Database examples¶

In [ ]:

from chowda.db import engine
from chowda.models import SonyCiAsset, MediaFile
from sqlmodel import Session, select, col, func
from re import search, split

from chowda.db import engine from chowda.models import SonyCiAsset, MediaFile from sqlmodel import Session, select, col, func from re import search, split

Get single asset by ID¶

In [ ]:

with Session(engine) as session:
    results = session.get(SonyCiAsset, '42c85f4305f04a2d8240084a801061d0')

with Session(engine) as session: results = session.get(SonyCiAsset, '42c85f4305f04a2d8240084a801061d0')

Search with regex¶

In [ ]:

statement = select(SonyCiAsset).where(col(SonyCiAsset.name).op('REGEXP')('^\ufeff'))

with Session(engine) as session:
    results = session.exec(statement).all()

statement = select(SonyCiAsset).where(col(SonyCiAsset.name).op('REGEXP')('^\ufeff')) with Session(engine) as session: results = session.exec(statement).all()

Find Media Files with more than one asset¶

In [ ]:

duplicates = (
    select(MediaFile)
    .join(SonyCiAsset)
    .group_by(MediaFile.guid)
    .having(func.count(SonyCiAsset.id) > 1)
)

with Session(engine) as session:
    results = session.exec(duplicates).all()

duplicates = ( select(MediaFile) .join(SonyCiAsset) .group_by(MediaFile.guid) .having(func.count(SonyCiAsset.id) > 1) ) with Session(engine) as session: results = session.exec(duplicates).all()

Duplicates with matching checksums¶

In [ ]:

matching = []
unmatching = []

duplicates = (
    select(MediaFile)
    .join(SonyCiAsset)
    .group_by(MediaFile.guid)
    .having(func.count(SonyCiAsset.id) > 1)
)

matching_statement = duplicates.having(
    func.count(func.distinct(SonyCiAsset.md5Checksum)) == 1
)
unmatching_statement = duplicates.having(
    func.count(func.distinct(SonyCiAsset.md5Checksum)) > 1
)

with Session(engine) as session:
    matching = session.exec(matching_statement).all()
    unmatching = session.exec(unmatching_statement).all()

print(f'Matching: {len(matching)}, Unmatching: {len(unmatching)}')

matching = [] unmatching = [] duplicates = ( select(MediaFile) .join(SonyCiAsset) .group_by(MediaFile.guid) .having(func.count(SonyCiAsset.id) > 1) ) matching_statement = duplicates.having( func.count(func.distinct(SonyCiAsset.md5Checksum)) == 1 ) unmatching_statement = duplicates.having( func.count(func.distinct(SonyCiAsset.md5Checksum)) > 1 ) with Session(engine) as session: matching = session.exec(matching_statement).all() unmatching = session.exec(unmatching_statement).all() print(f'Matching: {len(matching)}, Unmatching: {len(unmatching)}')

Aggregations¶

In [ ]:

folder_statement = matching_statement.having(
    # at least one related asset is in the 'Workspace' folder
    # func.bool_or(SonyCiAsset.folder['name'].as_string() == 'Workspace')

    # how many match
    func.count().filter(SonyCiAsset.folder['name'].as_string() == 'Workspace') > 0
)

with Session(engine) as session:
    matching = session.exec(folder_statement).all()
    len(matching)

folder_statement = matching_statement.having( # at least one related asset is in the 'Workspace' folder # func.bool_or(SonyCiAsset.folder['name'].as_string() == 'Workspace') # how many match func.count().filter(SonyCiAsset.folder['name'].as_string() == 'Workspace') > 0 ) with Session(engine) as session: matching = session.exec(folder_statement).all() len(matching)