from typing import List, Optional, Any from sqlalchemy import and_, exists from db import db from models.artist_team import ArtistTeam, ArtistTeamUser from models.labels import Label from models.user import user_label from sqlalchemy import func from shared.schemas import LabelResponseModel class LabelsRepository: def get_all_labels_for_user(self, user_id: int) -> List[LabelResponseModel]: labels = ( db.session.query(Label) .filter(exists().where(and_(user_label.c.label_id == Label.id, user_label.c.user_id == user_id))) .order_by(Label.name) .all() ) return [LabelResponseModel(label) for label in labels] def get_all_labels(self) -> List[LabelResponseModel]: labels = db.session.query(Label).order_by(Label.id).all() return [LabelResponseModel(label) for label in labels] def get_default_label_for_user(self, user_id: int) -> Optional[Any]: """ORM result or None """ return ( db.session.query( Label.id.label("label_id"), Label.name.label("label_name"), ) .select_from(Label) .join(user_label, and_(user_label.c.label_id == Label.id, user_label.c.user_id == user_id)) .outerjoin(ArtistTeam) .outerjoin( ArtistTeamUser, and_(ArtistTeam.id == ArtistTeamUser.artist_team_id, ArtistTeamUser.user_id == user_id) ) .group_by(Label.id) .order_by(func.count(ArtistTeamUser.id).desc(), Label.name.asc()) .first() ) def get_label_by_id(self, label_id): return db.session.query(Label).filter(Label.id == label_id).one_or_none()