"""TrackArtist Model.""" from oto import response from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy.orm import Session from product.connectors import mysql from product.constants import error class TrackArtist(mysql.BaseModel): """TrackArtist Model class.""" __tablename__ = 'track_artist' track_artist_id = Column( 'id', Integer, primary_key=True, autoincrement=True) tuid = Column('track_id', Integer) artist_role = Column( 'type', String(50), nullable=False, default='performer') name = Column(String(255), nullable=False) def get_artist_ids_for_tracks(tuids): """Get track artist_ids for list of track ids (for delete). Args: tuids (list): unique identifiers for tracks. Returns: Response: containing list of artist ids or [] """ with mysql.db_session() as session: track_artists = session.query(TrackArtist.track_artist_id) \ .filter(TrackArtist.tuid.in_(tuids)) \ .all() response_data = [each.track_artist_id for each in track_artists] return response.Response(message=response_data) def validate_artists_belong_to_tuid(tuid, artist_ids): """Validate if all artists belong to this tuid. Args: tuid (int): unique identifiers for track. artist_ids (list): track artist ids. Returns: Response: success/error response object. """ with mysql.db_session() as session: valid_artist_count = session.query(TrackArtist.track_artist_id) \ .filter_by(tuid=tuid) \ .filter(TrackArtist.track_artist_id.in_(artist_ids)) \ .count() if valid_artist_count != len(artist_ids): return response.create_error_response( error.ERROR_CODE_INVALID_DATA, "Invalid track artist ids. They don\'t belong to this track.") return response.Response(message=True) def delete(track_ids: list[int], session: Session) -> bool: """Delete track artists by list of track_ids. Args: track_ids (list): unique identifiers for tracks. Returns: bool: True if deleted successfully else False. """ result: int = session.query(TrackArtist).filter(TrackArtist.tuid.in_(track_ids)).delete() return result > 0