91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
|
"""Shared data access for the polymorphic satellite tables — credits,
|
||
|
|
artwork, videos, external IDs — that attach to more than one entity type via
|
||
|
|
an (entity_type, entity_id) pair rather than a typed foreign key.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
from sqlalchemy.orm import selectinload
|
||
|
|
|
||
|
|
from app.models.artwork import Artwork
|
||
|
|
from app.models.credit import Credit
|
||
|
|
from app.models.enums import ArtworkType
|
||
|
|
from app.models.external_id import ExternalID
|
||
|
|
from app.models.provider import Provider
|
||
|
|
from app.models.video import Video
|
||
|
|
|
||
|
|
|
||
|
|
async def get_external_ids(
|
||
|
|
session: AsyncSession, entity_type: str, entity_id: uuid.UUID
|
||
|
|
) -> dict[str, str]:
|
||
|
|
stmt = (
|
||
|
|
select(Provider.slug, ExternalID.external_id)
|
||
|
|
.join(Provider, Provider.id == ExternalID.provider_id)
|
||
|
|
.where(ExternalID.entity_type == entity_type, ExternalID.entity_id == entity_id)
|
||
|
|
)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
return {slug: external_id for slug, external_id in result.all()}
|
||
|
|
|
||
|
|
|
||
|
|
async def get_external_ids_bulk(
|
||
|
|
session: AsyncSession, entity_type: str, entity_ids: list[uuid.UUID]
|
||
|
|
) -> dict[uuid.UUID, dict[str, str]]:
|
||
|
|
if not entity_ids:
|
||
|
|
return {}
|
||
|
|
stmt = (
|
||
|
|
select(ExternalID.entity_id, Provider.slug, ExternalID.external_id)
|
||
|
|
.join(Provider, Provider.id == ExternalID.provider_id)
|
||
|
|
.where(ExternalID.entity_type == entity_type, ExternalID.entity_id.in_(entity_ids))
|
||
|
|
)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
out: dict[uuid.UUID, dict[str, str]] = {eid: {} for eid in entity_ids}
|
||
|
|
for entity_id, slug, external_id in result.all():
|
||
|
|
out.setdefault(entity_id, {})[slug] = external_id
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
async def get_artwork(
|
||
|
|
session: AsyncSession,
|
||
|
|
entity_type: str,
|
||
|
|
entity_id: uuid.UUID,
|
||
|
|
artwork_type: ArtworkType | None = None,
|
||
|
|
) -> list[Artwork]:
|
||
|
|
stmt = (
|
||
|
|
select(Artwork)
|
||
|
|
.options(selectinload(Artwork.provider))
|
||
|
|
.where(Artwork.entity_type == entity_type, Artwork.entity_id == entity_id)
|
||
|
|
.order_by(Artwork.is_primary.desc(), Artwork.vote_average.desc().nulls_last())
|
||
|
|
)
|
||
|
|
if artwork_type is not None:
|
||
|
|
stmt = stmt.where(Artwork.artwork_type == artwork_type)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
return list(result.scalars().all())
|
||
|
|
|
||
|
|
|
||
|
|
async def get_videos(session: AsyncSession, entity_type: str, entity_id: uuid.UUID) -> list[Video]:
|
||
|
|
stmt = (
|
||
|
|
select(Video)
|
||
|
|
.options(selectinload(Video.provider))
|
||
|
|
.where(Video.entity_type == entity_type, Video.entity_id == entity_id)
|
||
|
|
.order_by(Video.is_official.desc(), Video.published_at.desc().nulls_last())
|
||
|
|
)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
return list(result.scalars().all())
|
||
|
|
|
||
|
|
|
||
|
|
async def get_credits(
|
||
|
|
session: AsyncSession, entity_type: str, entity_id: uuid.UUID
|
||
|
|
) -> list[Credit]:
|
||
|
|
stmt = (
|
||
|
|
select(Credit)
|
||
|
|
.options(selectinload(Credit.person))
|
||
|
|
.where(Credit.entity_type == entity_type, Credit.entity_id == entity_id)
|
||
|
|
.order_by(Credit.billing_order.asc().nulls_last())
|
||
|
|
)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
return list(result.scalars().all())
|