"""SQLAlchemy model and lookup for the track_display_artists table.""" from typing import Any from sqlalchemy import Enum, select from sqlalchemy.dialects.mysql import INTEGER from sqlalchemy.orm import Mapped, Session, mapped_column from contributor.connectors.mysql import BaseModel, db_session_wrap from contributor.queries.mysql.contributor import Contributor class TrackDisplayArtist(BaseModel): """Track display artist model.""" __tablename__ = "track_display_artists" id: Mapped[int] = mapped_column( INTEGER(unsigned=True), primary_key=True, autoincrement=True, ) track_id: Mapped[int] = mapped_column(INTEGER(unsigned=True), nullable=False) contributor_id: Mapped[int] = mapped_column(INTEGER(unsigned=True), nullable=False) display_artist_role: Mapped[str] = mapped_column( Enum( "featuring", "primary", "feature_to_primary", name="display_artist_role", ), nullable=False, ) sequence_number: Mapped[int] = mapped_column( INTEGER(unsigned=True), nullable=False, default=0 ) @db_session_wrap def get_by_track_ids(session: Session, *, track_ids: list[int]) -> list[dict[str, Any]]: result = session.execute( select( TrackDisplayArtist.track_id, TrackDisplayArtist.display_artist_role, TrackDisplayArtist.sequence_number, TrackDisplayArtist.id, Contributor.neo4j_participant_uuid, ) .join(Contributor, Contributor.id == TrackDisplayArtist.contributor_id) .where(TrackDisplayArtist.track_id.in_(track_ids)) .order_by( TrackDisplayArtist.track_id, TrackDisplayArtist.sequence_number, TrackDisplayArtist.id, ) ).all() display_artists_by_track_id: dict[int, list[dict[str, Any]]] = {} for track_id, role, _, _, contributor_uuid in result: display_artists_by_track_id.setdefault(track_id, []).append( { "contributor": {"uuid": contributor_uuid}, "role": role, } ) return [ { "track": {"tuid": track_id}, "display_artists": display_artists_by_track_id.get(track_id, []), } for track_id in track_ids if track_id in display_artists_by_track_id ]