Models

Models

SQLModels for DB and validation

Batch

Bases: SQLModel

Source code in chowda/models.py

class Batch(SQLModel, table=True):
    __tablename__ = 'batches'
    id: Optional[int] = Field(primary_key=True, default=None)
    name: str
    description: str
    pipeline_id: Optional[int] = Field(default=None, foreign_key='pipelines.id')
    pipeline: Optional['Pipeline'] = Relationship(back_populates='batches')
    media_files: List[MediaFile] = Relationship(
        back_populates='batches', link_model=MediaFileBatchLink
    )
    output_mmifs: List['MMIF'] = Relationship(
        back_populates='batch_output',
        sa_relationship_kwargs={
            "primaryjoin": "Batch.id==MMIF.batch_output_id",
        },
    )

    input_mmifs: List['MMIF'] = Relationship(
        back_populates='batch_inputs',
        link_model=MMIFBatchInputLink,
    )
    metaflow_runs: List['MetaflowRun'] = Relationship(back_populates='batch')

    def unstarted_guids(self) -> set:
        """Returns the set of GUIDs that are not currently running"""
        ids: set = {media_file.guid for media_file in self.media_files}
        running_guids: set = {run.media_file.guid for run in self.metaflow_runs}
        return ids - running_guids

    async def __admin_repr__(self, request: Request) -> str:
        return f'{self.name or self.id}'

    async def __admin_select2_repr__(self, request: Request) -> str:
        return f'<span><strong>{self.name or self.id}</span>'

unstarted_guids()

Returns the set of GUIDs that are not currently running

Source code in chowda/models.py

def unstarted_guids(self) -> set:
    """Returns the set of GUIDs that are not currently running"""
    ids: set = {media_file.guid for media_file in self.media_files}
    running_guids: set = {run.media_file.guid for run in self.metaflow_runs}
    return ids - running_guids

MMIF

Bases: SQLModel

MMIF model

Attributes:

Name	Type	Description
id	Optional[int]	Primary key
created_at	Optional[datetime]	Creation timestamp
media_file_id	Optional[str]	GUID
media_file	Optional[MediaFile]	MediaFile
metaflow_run_id	Optional[str]	MetaflowRun ID
metaflow_run	Optional[MetaflowRun]	MetaflowRun
batch_output	Optional[Batch]	Batch that generated this MMIF
batch_inputs	List[Batch]	Batch that uses this as an input
mmif_location	Optional[str]	S3 URL of the mmif

Source code in chowda/models.py

class MMIF(SQLModel, table=True):
    """MMIF model

    Attributes:
        id: Primary key
        created_at: Creation timestamp
        media_file_id: GUID
        media_file: MediaFile
        metaflow_run_id: MetaflowRun ID
        metaflow_run: MetaflowRun
        batch_output: Batch that generated this MMIF
        batch_inputs: Batch that uses this as an input
        mmif_location: S3 URL of the mmif
    """

    __tablename__ = 'mmifs'
    id: Optional[int] = Field(primary_key=True, default=None, index=True)
    created_at: Optional[datetime] = Field(
        sa_column=Column(DateTime(timezone=True), default=datetime.utcnow)
    )
    media_file_id: Optional[str] = Field(
        default=None, foreign_key='media_files.guid', index=True
    )
    media_file: Optional[MediaFile] = Relationship(back_populates='mmifs')
    metaflow_run_id: Optional[str] = Field(default=None, foreign_key='metaflow_runs.id')
    metaflow_run: Optional[MetaflowRun] = Relationship(back_populates='mmif')
    batch_output_id: Optional[int] = Field(default=None, foreign_key='batches.id')
    batch_output: Optional[Batch] = Relationship(
        back_populates='output_mmifs',
        sa_relationship_kwargs={
            "primaryjoin": "MMIF.batch_output_id==Batch.id",
        },
    )
    batch_inputs: List[Batch] = Relationship(
        back_populates='input_mmifs',
        link_model=MMIFBatchInputLink,
    )

    mmif_location: Optional[str] = Field(default=None)

    async def __admin_repr__(self, request: Request):
        return (
            f'{self.metaflow_run.batch.name}'
            if self.metaflow_run and self.metaflow_run.batch
            else self.id
        )

    async def __admin_select2_repr__(self, request: Request) -> str:
        text = (
            self.metaflow_run.batch.name
            if self.metaflow_run and self.metaflow_run.batch
            else self.id
        )
        return f'<span>{text}</span>'

MediaFile

Bases: SQLModel

Media file model

Attributes:

Name	Type	Description
guid	Optional[str]	MediaFile GUID
assets	List[SonyCiAsset]	List of SonyCiAssets
collections	List[Collection]	List of Collections
batches	List[Batch]	List of Batches
metaflow_runs	List[MetaflowRun]	List of MetaflowRuns
mmifs	List[MMIF]	List of MMIFs

Source code in chowda/models.py

class MediaFile(SQLModel, table=True):
    """Media file model

    Attributes:
        guid: MediaFile GUID
        assets: List of SonyCiAssets
        collections: List of Collections
        batches: List of Batches
        metaflow_runs: List of MetaflowRuns
        mmifs: List of MMIFs
    """

    __tablename__ = 'media_files'
    guid: Optional[str] = Field(primary_key=True, default=None, index=True)
    mmifs: List['MMIF'] = Relationship(back_populates='media_file')
    assets: List['SonyCiAsset'] = Relationship(back_populates='media_files')
    collections: List['Collection'] = Relationship(
        back_populates='media_files', link_model=MediaFileCollectionLink
    )
    batches: List['Batch'] = Relationship(
        back_populates='media_files', link_model=MediaFileBatchLink
    )
    metaflow_runs: List['MetaflowRun'] = Relationship(back_populates='media_file')

    def metaflow_runs_for_batch(self, batch_id: int):
        return [
            metaflow_run
            for metaflow_run in self.metaflow_runs
            if metaflow_run.batch_id == batch_id
        ]

    def last_metaflow_run_for_batch(self, batch_id: int):
        # TODO: is getting the last one sufficient, or do we need to add sortable
        # timestamps?
        runs = self.metaflow_runs_for_batch(batch_id=batch_id)
        return runs[-1] if len(runs) > 0 else None

    async def __admin_repr__(self, request: Request):
        return self.guid

    async def __admin_select2_repr__(self, request: Request) -> str:
        return f'<span><strong>{self.guid}</strong></span>'

MediaType

Bases: Enum

Media type enum Type of Media: video or audio. This is not the same as the format of the media file.

FIXME:

Enum class attributes are the values that are stored in the database. The value of the class attribute is the value returned by SonyCi (for validation) But starlette-admin + SQLAlchemy send the value to the db, not the name. Therefore, we need to Capatalize the name to make it match the db value.

Source code in chowda/models.py

class MediaType(enum.Enum):
    """Media type enum
    Type of Media: video or audio.
    This is not the same as the format of the media file.

    # FIXME:
    Enum class attributes are the values that are stored in the database.
    The value of the class attribute is the value returned by SonyCi (for validation)
    But starlette-admin + SQLAlchemy send the value to the db, not the name.
    Therefore, we need to Capatalize the name to make it match the db value.
    """

    Video = 'Video'
    Audio = 'Audio'

User

Bases: SQLModel

User model

Attributes:

Name	Type	Description
id	Optional[int]	Primary key
email	EmailStr	User email
first_name	str	User first name
last_name	str	User last name

Source code in chowda/models.py

class User(SQLModel, table=True):
    """User model

    Attributes:
        id: Primary key
        email: User email
        first_name: User first name
        last_name: User last name
    """

    __tablename__ = 'users'
    id: Optional[int] = Field(primary_key=True)
    email: EmailStr = Field(unique=True, index=True, sa_type=AutoString)
    first_name: str = Field(min_length=3, index=True)
    last_name: str = Field(min_length=3, index=True)

    async def __admin_repr__(self, request: Request):
        return f'{self.first_name} {self.last_name}'