"""Neo4j queries for sound recording participations.""" from typing import Any from ddtrace import tracer from neo4j import Session get_sound_recording_participations_by_label_participant_uuid = """ MATCH (lp:LabelParticipant {uuid: $uuid}) WHERE $is_admin OR EXISTS { MATCH (p:Profile {profileType: 'LabelProfile'}) -[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->() -[:HAS_LABEL_PARTICIPANT]->(lp) WHERE p.profileId IN $profile_ids } WITH DISTINCT lp OPTIONAL MATCH (lp)-[:HAS_LABEL_SOUND_RECORDING_PARTICIPATION]->(lsrp:LabelSoundRecordingParticipation), (lsrp)-[:ON_LABEL_SOUND_RECORDING]->(lsr:LabelSoundRecording), (lsrp)-[:WITH_PARTICIPATION_ROLE]->(pr:ParticipationRole) OPTIONAL MATCH (pr)-[:BELONGS_TO]->(prc:ParticipationRoleCategory) WITH lp, lsr, lsrp, pr, prc ORDER BY lsrp.sequenceNumber RETURN {uuid: lp.uuid} AS contributor, collect( CASE WHEN lsrp IS NOT NULL THEN { contributor: {uuid: lp.uuid}, label_sound_recording: {id: lsr.id}, role: { apple_role_name: pr.appleRoleName, category: {name: prc.name, uuid: prc.uuid}, ddex_role_name: pr.ddexRoleName, name: pr.name }, sequence_number: lsrp.sequenceNumber } END ) AS participations """ @tracer.wrap() def get_by_label_participant_uuid( session: Session, uuid: str, profile_ids: list[int], is_admin: bool = False, ) -> dict[str, Any] | None: result = session.run( get_sound_recording_participations_by_label_participant_uuid, uuid=uuid, profile_ids=profile_ids, is_admin=is_admin, ) record = result.single() if record is None: return None participations = [ { "contributor": {"uuid": p["contributor"]["uuid"]}, "label_sound_recording": p["label_sound_recording"], "role": { "apple_role_name": p["role"]["apple_role_name"], "category": { "name": p["role"]["category"]["name"], "uuid": p["role"]["category"]["uuid"], }, "ddex_role_name": p["role"]["ddex_role_name"], "name": p["role"]["name"], }, "sequence_number": p["sequence_number"], } for p in record["participations"] if p is not None ] return { "contributor": record["contributor"], "participations": participations, }