59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
"""Database operations for MediaAsset — SQLModel."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
from uuid import UUID
|
|
|
|
from sqlmodel import select
|
|
|
|
from .connection import get_session
|
|
from .models import MediaAsset
|
|
|
|
|
|
def list_assets(status: Optional[str] = None, search: Optional[str] = None) -> list[MediaAsset]:
|
|
with get_session() as session:
|
|
stmt = select(MediaAsset)
|
|
if status:
|
|
stmt = stmt.where(MediaAsset.status == status)
|
|
if search:
|
|
stmt = stmt.where(MediaAsset.filename.ilike(f"%{search}%"))
|
|
return list(session.exec(stmt).all())
|
|
|
|
|
|
def get_asset(id: UUID) -> MediaAsset | None:
|
|
with get_session() as session:
|
|
return session.get(MediaAsset, id)
|
|
|
|
|
|
def get_asset_filenames() -> set[str]:
|
|
with get_session() as session:
|
|
return set(session.exec(select(MediaAsset.filename)).all())
|
|
|
|
|
|
def create_asset(*, filename: str, file_path: str, file_size: int) -> MediaAsset:
|
|
asset = MediaAsset(filename=filename, file_path=file_path, file_size=file_size)
|
|
with get_session() as session:
|
|
session.add(asset)
|
|
session.commit()
|
|
session.refresh(asset)
|
|
return asset
|
|
|
|
|
|
def update_asset(id: UUID, **fields) -> None:
|
|
with get_session() as session:
|
|
asset = session.get(MediaAsset, id)
|
|
if not asset:
|
|
return
|
|
for k, v in fields.items():
|
|
setattr(asset, k, v)
|
|
session.commit()
|
|
|
|
|
|
def delete_asset(id: UUID) -> None:
|
|
with get_session() as session:
|
|
asset = session.get(MediaAsset, id)
|
|
if asset:
|
|
session.delete(asset)
|
|
session.commit()
|