"""SQLAlchemy model and lookup for the track_artist table.""" from sqlalchemy import String, select, update from sqlalchemy.dialects.mysql import INTEGER from sqlalchemy.orm import Mapped, Session, mapped_column from contributor.connectors.mysql import BaseModel from contributor.queries.mysql.project import Project from contributor.queries.mysql.release import Release from contributor.queries.mysql.track import Track class TrackArtist(BaseModel): __tablename__ = "track_artist" id: Mapped[int] = mapped_column( INTEGER(unsigned=True), primary_key=True, autoincrement=True, ) track_id: Mapped[int] = mapped_column(INTEGER(unsigned=True), nullable=False) name: Mapped[str | None] = mapped_column(String(255), nullable=True) artist_info_id: Mapped[int | None] = mapped_column( INTEGER(unsigned=True), nullable=True ) def rename_track_artists( *, session: Session, current_artist_name: str, vendor_id: int, new_artist_name: str, ) -> None: """Bulk update track artists with a given name + vendor_id to a new name.""" session.execute( update(TrackArtist) .where( TrackArtist.track_id.in_( select(Track.id) .join(Release, Release.release_id == Track.release_id) .join(Project, Project.project_id == Release.project_id) .where(Project.vendor_id == vendor_id) .scalar_subquery() ) ) .where(TrackArtist.name == current_artist_name) .values(name=new_artist_name) ) def reassign_artist_info( *, session: Session, artist_name: str, artist_info_id: int, duplicate_artist_info_id: int, ) -> None: """ Reassign track artists from duplicate_artist_info_id to artist_info_id, updating the artist name at the same time. """ session.execute( update(TrackArtist) .where(TrackArtist.artist_info_id == duplicate_artist_info_id) .values(artist_info_id=artist_info_id, name=artist_name) )