Models

Models

SQLModels for DB and validation

AssetType

Bases: Enum

Asset type enum Type of SonyCi asset: audio, video, or document / other.

This is not the same as the MediaType of the asset (Audio or Video).

Note: Assets with "status: Uploading" have no type

Source code in chowda/models.py

class AssetType(enum.Enum):
    """Asset type enum
    Type of SonyCi asset: audio, video, or document / other.

    This is not the same as the `MediaType` of the asset (Audio or Video).

    Note: Assets with "status: Uploading" have no type
    """

    Audio = 'Audio'
    Video = 'Video'
    Image = 'Image'
    Document = 'Document'
    TimedText = 'TimedText'
    Other = 'Other'
    NoneType = None

Batch

Bases: SQLModel

Source code in chowda/models.py

class Batch(SQLModel, table=True):
    __tablename__ = 'batches'
    id: int | None = Field(primary_key=True, default=None)
    name: str
    description: str
    pipeline_id: int | None = 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	int | None	Primary key
created_at	datetime | None	Creation timestamp
media_file_id	str | None	GUID
media_file	MediaFile | None	MediaFile
metaflow_run_id	str | None	MetaflowRun ID
metaflow_run	MetaflowRun | None	MetaflowRun
batch_output	Batch | None	Batch that generated this MMIF
batch_inputs	list[Batch]	Batch that uses this as an input
mmif_location	str | None	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: int | None = Field(primary_key=True, default=None, index=True)
    created_at: datetime | None = Field(
        sa_column=Column(DateTime(timezone=True), default=datetime.utcnow)
    )
    media_file_id: str | None = Field(
        default=None, foreign_key='media_files.guid', index=True
    )
    media_file: MediaFile | None = Relationship(back_populates='mmifs')
    metaflow_run_id: str | None = Field(default=None, foreign_key='metaflow_runs.id')
    metaflow_run: MetaflowRun | None = Relationship(back_populates='mmif')
    batch_output_id: int | None = Field(default=None, foreign_key='batches.id')
    batch_output: Batch | None = 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: str | None = 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	str | None	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: str | None = Field(primary_key=True, default=None, index=True)
    mmifs: list['MMIF'] = Relationship(back_populates='media_file')
    assets: list['SonyCiAsset'] = Relationship(back_populates='media_files')
    trashbin: list['SonyCiTrashbin'] = 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.
    """

    Audio = 'Audio'
    Video = 'Video'

SonyCiArchiveStatus

Bases: Enum

SonyCi archive status

Accepted values: - Not archived - Archive in progress - Archived - Restore in progress - Restored

Source code in chowda/models.py

class SonyCiArchiveStatus(enum.Enum):
    """SonyCi archive status

    Accepted values:
    - Not archived
    - Archive in progress
    - Archived
    - Restore in progress
    - Restored
    """

    NotArchived = 'Not archived'
    ArchiveInProgress = 'Archive in progress'
    Archived = 'Archived'
    RestoreInProgress = 'Restore in progress'
    Restored = 'Restored'

SonyCiAsset

Bases: SonyCiAssetBase

SonyCiAsset model

Source code in chowda/models.py

class SonyCiAsset(SonyCiAssetBase, table=True):
    """SonyCiAsset model"""

    __tablename__ = 'sonyci_assets'

    media_files: MediaFile | None = Relationship(back_populates='assets')
    events: list[SonyCiEvent] = Relationship(
        back_populates='assets', link_model=SonyCiEventAssetLink
    )

SonyCiAssetBase

Bases: SQLModel

Shared columns for SonyCi assets and trashbin entries.

Non-table base mixin. Uses sa_type (not sa_column) so each concrete table subclass gets its own Column instances. Relationships are declared on the subclasses, since a Column/relationship belongs to a single mapper.

Source code in chowda/models.py

class SonyCiAssetBase(SQLModel):
    """Shared columns for SonyCi assets and trashbin entries.

    Non-table base mixin. Uses ``sa_type`` (not ``sa_column``) so each concrete
    table subclass gets its own Column instances. Relationships are declared on
    the subclasses, since a Column/relationship belongs to a single mapper.
    """

    id: str | None = Field(primary_key=True, index=True, default=None)
    name: str = Field(index=True)
    size: int = Field(sa_type=postgresql.BIGINT)
    createdOn: datetime | None = Field(default=None, index=True)
    createdBy: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    modifiedOn: datetime | None = Field(default=None, index=True)
    lastActivityOn: datetime | None = Field(default=None, index=True)
    acquisitionSource: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    type: AssetType | None = Field(default=None, sa_type=Enum(AssetType))
    format: str | None = Field(default=None, index=True)
    folder: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    md5Checksum: str | None = Field(default=None, index=True)
    # SonyCi returns the enum *values* (e.g. 'Not archived'); table models skip
    # validation, so those raw strings are stored as-is. values_callable makes
    # the DB enum store values instead of member names so they round-trip.
    status: SonyCiAssetStatus | None = Field(
        default=None, sa_type=Enum(SonyCiAssetStatus, values_callable=_enum_values)
    )
    archiveStatus: SonyCiArchiveStatus | None = Field(
        default=None, sa_type=Enum(SonyCiArchiveStatus, values_callable=_enum_values)
    )
    restoreStatus: SonyCiRestoreStatus | None = Field(
        default=None, sa_type=Enum(SonyCiRestoreStatus, values_callable=_enum_values)
    )
    isTrashed: bool | None = Field(default=None, index=True)
    runtime: float | None = Field(default=None, index=True)
    totalFolderCount: int | None = Field(default=None, index=True)
    asset_metadata: list[dict[str, Any]] | None = Field(
        default=None, sa_type=postgresql.ARRAY(JSON)
    )
    uploadCompleteDate: datetime | None = Field(default=None, index=True)
    uploadTransferType: SonyCiUploadTransferType | None = Field(
        default=None, sa_type=Enum(SonyCiUploadTransferType)
    )
    hasPlayableProxies: bool | None = Field(default=None, index=True)
    generatingPlayableProxies: bool | None = Field(default=None, index=True)
    thumbnails: list[dict[str, Any]] | None = Field(
        default=None, sa_type=postgresql.ARRAY(JSON)
    )
    proxies: list[dict[str, Any]] | None = Field(
        default=None, sa_type=postgresql.ARRAY(JSON)
    )
    filmstrips: list[dict[str, Any]] | None = Field(
        default=None, sa_type=postgresql.ARRAY(JSON)
    )
    technicalMetadata: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    technicalMetadataUpdates: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    waveforms: list[dict[str, Any]] | None = Field(
        default=None, sa_type=postgresql.ARRAY(JSON)
    )
    media_file_id: str | None = Field(
        default=None, foreign_key='media_files.guid', index=True
    )

    @property
    def thumbnails_by_type(self):
        return {thumbnail['type']: thumbnail for thumbnail in self.thumbnails}

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

SonyCiAssetStatus

Bases: Enum

SonyCi Asset status Status of SonyCi asset

Typical statuses

Created - The asset record has been created in our database. Waiting - Once uploaded the asset's status will transition to Waiting to indicate it is waiting for backend processing jobs to begin (thumbnails, preview proxies, calculating MD5 checksum, extract technical metadata, check for viruses, etc.). Processing - Once those jobs begin its status moves to Processing. Complete - The status will go to Complete when all applicable thumbnails and preview proxies are finished (note: other jobs like calculating MD5 checksum, gathering technical metadata, and performing other verifications may not be completed even though the status is Complete). Failed - If a failure happens during upload the status will go to Failed.

Other statuses

Limited - There were problems generating thumbnails and/or preview proxies, The source file is still available for download. Virus Detected - A virus was found and the file is not downloadable and generally cannot be used in our system. Executable Detected - An executable file was found and the file is not downloadable and generally cannot be used in our system. Deleted - The file has been deleted.

Source code in chowda/models.py

class SonyCiAssetStatus(enum.Enum):
    """SonyCi Asset status
    Status of SonyCi asset

    ## Typical statuses
    Created - The asset record has been created in our database.
    Waiting - Once uploaded the asset's status will transition to Waiting to indicate it is waiting for backend processing jobs to begin (thumbnails, preview proxies, calculating MD5 checksum, extract technical metadata, check for viruses, etc.).
    Processing - Once those jobs begin its status moves to Processing.
    Complete - The status will go to Complete when all applicable thumbnails and preview proxies are finished (note: other jobs like calculating MD5 checksum, gathering technical metadata, and performing other verifications may not be completed even though the status is Complete).
    Failed - If a failure happens during upload the status will go to Failed.

    ## Other statuses
    Limited - There were problems generating thumbnails and/or preview proxies, The source file is still available for download.
    Virus Detected - A virus was found and the file is not downloadable and generally cannot be used in our system.
    Executable Detected - An executable file was found and the file is not downloadable and generally cannot be used in our system.
    Deleted - The file has been deleted.
    """

    Created = 'Created'
    Complete = 'Complete'
    Deleted = 'Deleted'
    ExecutableDetected = 'Executable Detected'
    Failed = 'Failed'
    Limited = 'Limited'
    Processing = 'Processing'
    Uploading = 'Uploading'
    VirusDetected = 'Virus Detected'
    Waiting = 'Waiting'

SonyCiEvent

Bases: SQLModel

SonyCiEvent model

A webhook notification received from SonyCi.

Attributes:

Name	Type	Description
id	str | None	SonyCi event id
type	SonyCiEventType | None	Type of the event
createdOn	datetime | None	When SonyCi created the event
createdBy	dict[str, Any] | None	SonyCi user that triggered the event
payload	dict[str, Any] | None	The complete event as received from SonyCi
created_at	datetime | None	When Chowda received the event
assets	list[SonyCiAsset]	SonyCiAssets referenced by the event

Source code in chowda/models.py

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

    A webhook notification received from SonyCi.

    Attributes:
        id: SonyCi event id
        type: Type of the event
        createdOn: When SonyCi created the event
        createdBy: SonyCi user that triggered the event
        payload: The complete event as received from SonyCi
        created_at: When Chowda received the event
        assets: SonyCiAssets referenced by the event
    """

    __tablename__ = 'sonyci_events'

    id: str | None = Field(primary_key=True, default=None, index=True)
    type: SonyCiEventType | None = Field(
        default=None, sa_type=Enum(SonyCiEventType), index=True
    )
    createdOn: datetime | None = Field(default=None, index=True)
    createdBy: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    payload: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    created_at: datetime | None = Field(
        sa_column=Column(DateTime(timezone=True), default=datetime.utcnow, index=True)
    )
    assets: list['SonyCiAsset'] = Relationship(
        back_populates='events', link_model=SonyCiEventAssetLink
    )

    async def __admin_repr__(self, request: Request):
        return f'{self.type.value if self.type else "Event"} {self.id}'

SonyCiEventType

Bases: Enum

SonyCi webhook event type

The type of notification sent by SonyCi to the webhook API.

Source code in chowda/models.py

class SonyCiEventType(enum.Enum):
    """SonyCi webhook event type

    The type of notification sent by SonyCi to the webhook API.
    """

    MoveAsset = 'MoveAsset'
    CopyAsset = 'CopyAsset'
    TrashAsset = 'TrashAsset'
    DeleteAsset = 'DeleteAsset'
    DeleteElement = 'DeleteElement'
    JobStatusChange = 'JobStatusChange'
    AssetProcessingFinished = 'AssetProcessingFinished'
    ElementProcessingFinished = 'ElementProcessingFinished'
    AssetArchiveStatusChange = 'AssetArchiveStatusChange'
    AssetRestoreStatusChange = 'AssetRestoreStatusChange'
    LockElement = 'LockElement'
    UnlockElement = 'UnlockElement'
    CopyAssetsToWorkspace = 'CopyAssetsToWorkspace'
    CopyAssetsToCatalog = 'CopyAssetsToCatalog'
    AssetMetadataChange = 'AssetMetadataChange'
    ElementMetadataChange = 'ElementMetadataChange'

SonyCiRestoreStatus

Bases: Enum

SonyCi restore status

Accepted values

• Not restored
• Restore in progress
• Restore failed
• Restored

Source code in chowda/models.py

class SonyCiRestoreStatus(enum.Enum):
    """SonyCi restore status

    Accepted values:
     - Not restored
     - Restore in progress
     - Restore failed
     - Restored
    """

    NotRestored = 'Not restored'
    RestoreInProgress = 'Restore in progress'
    RestoreFailed = 'Restore failed'
    Restored = 'Restored'

SonyCiTrashbin

Bases: SonyCiAssetBase

A SonyCiAsset that has been trashed in SonyCi.

Source code in chowda/models.py

class SonyCiTrashbin(SonyCiAssetBase, table=True):
    """A SonyCiAsset that has been trashed in SonyCi."""

    __tablename__ = 'sonyci_trashbin'

    statusDescription: dict[str, Any] | None = Field(default=None, sa_type=JSON)
    trashedOn: datetime | None = Field(default=None, index=True)

    media_files: MediaFile | None = Relationship(back_populates='trashbin')

SonyCiUploadTransferType

Bases: Enum

SonyCi upload transfer type Indicates how the asset was uploaded.

Valid values are:

• SinglepartHttp
• MultipartHttp
• Aspera
• Copy
• FTP
• WorkspaceSend
• ClipFromSource

Source code in chowda/models.py

class SonyCiUploadTransferType(enum.Enum):
    """SonyCi upload transfer type
    Indicates how the asset was uploaded.

    Valid values are:

    - SinglepartHttp
    - MultipartHttp
    - Aspera
    - Copy
    - FTP
    - WorkspaceSend
    - ClipFromSource
    """

    SinglepartHttp = 'SinglepartHttp'
    MultipartHttp = 'MultipartHttp'
    Aspera = 'Aspera'
    Copy = 'Copy'
    FTP = 'FTP'
    WorkspaceSend = 'WorkspaceSend'
    ClipFromSource = 'ClipFromSource'

User

Bases: SQLModel

User model

Attributes:

Name	Type	Description
id	int | None	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: int | None = 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}'