"""Business logic for track endpoints.""" from typing import Any from ddtrace import tracer from neo4j import Session as Neo4jSession from pydantic import UUID4 from contributor.logic.access import has_full_catalog_access, label_profile_ids from contributor.queries.mysql import track_display_artist as display_artist_queries from contributor.queries.neo4j import tracks as queries @tracer.wrap() def get_track_participations( session: Neo4jSession, contributor_uuid: UUID4, profiles: list[dict], ) -> dict[str, Any] | None: is_admin = has_full_catalog_access(session, profiles) result = queries.get_by_label_participant_uuid( session=session, uuid=str(contributor_uuid), profile_ids=label_profile_ids(profiles), is_admin=is_admin, ) if result is None: return None result["participations"] = [ p for p in result["participations"] if p.get("participated_as") != "feature_to_primary" ] return result @tracer.wrap() def get_track_participations_by_track_ids( session: Neo4jSession, track_ids: list[int], profiles: list[dict], ) -> list[dict[str, Any]]: is_admin = has_full_catalog_access(session, profiles) results = ( queries.get_by_track_ids( session=session, track_ids=track_ids, profile_ids=label_profile_ids(profiles), is_admin=is_admin, ) or [] ) display_artist_results = display_artist_queries.get_by_track_ids( track_ids=track_ids, ) display_artists_by_track_id = { result["track"]["tuid"]: result["display_artists"] for result in display_artist_results } for result in results: result["participations"] = [ p for p in result["participations"] if p is not None and p.get("participated_as") != "feature_to_primary" ] track_id = result["track"]["tuid"] result["display_artists"] = display_artists_by_track_id.get(track_id, []) return results @tracer.wrap() def get_track_participations_for_track( session: Neo4jSession, track_id: int, profiles: list[dict], ) -> dict[str, Any] | None: results = get_track_participations_by_track_ids( session=session, track_ids=[track_id], profiles=profiles, ) return results[0] if results else None