"""Business logic for global participant endpoints.""" from typing import Any from ddtrace import tracer from neo4j import Session as Neo4jSession from contributor.logic.access import has_full_catalog_access, label_profile_ids from contributor.queries.neo4j import global_participants as queries @tracer.wrap() def get_global_participant( session: Neo4jSession, global_participant_id: str, profiles: list[dict], ) -> dict[str, Any] | None: results = get_global_participants(session, [global_participant_id], profiles) return results[0] if results else None @tracer.wrap() def get_global_participants( session: Neo4jSession, global_participant_ids: list[str], profiles: list[dict], ) -> list[dict[str, Any]]: is_admin = has_full_catalog_access(session, profiles) return queries.get_by_ids( session=session, ids=global_participant_ids, profile_ids=label_profile_ids(profiles), is_admin=is_admin, ) @tracer.wrap() def get_contributors_for_global_participant( session: Neo4jSession, global_participant_id: str, profiles: list[dict], ) -> list[dict[str, Any]] | None: is_admin = has_full_catalog_access(session, profiles) return queries.get_contributors( session=session, global_participant_id=global_participant_id, profile_ids=label_profile_ids(profiles), is_admin=is_admin, )