import secrets from datetime import datetime, timezone from typing import Annotated, List, Optional import sentry_sdk from fastapi import ( APIRouter, Depends, FastAPI, HTTPException, Query, Request, Security, status, ) from fastapi.responses import JSONResponse from fastapi.security import APIKeyHeader from pydantic import BaseModel, BeforeValidator, ConfigDict, field_serializer from pydantic.alias_generators import to_camel from sentry_sdk.integrations.asgi import SentryAsgiMiddleware from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from starlette.middleware.base import BaseHTTPMiddleware from playlist_sync.config import settings from playlist_sync.models.application import Application from playlist_sync.models.enums import FieldCopyMode, MusicService, ServiceType from playlist_sync.models.insert_media import InsertMedia from playlist_sync.models.service_account import ServiceAccount from playlist_sync.models.sync_log import PlaylistSynchronizationLog from playlist_sync.models.sync_task import PlaylistSynchronization from playlist_sync.services.application_service import ApplicationService from playlist_sync.services.database import get_session from playlist_sync.services.sync_service import ( ServiceAccountService, SyncLogService, SyncTaskService, ) # --------------------------------------------------------------------------- # Authentication # --------------------------------------------------------------------------- _api_key_header = APIKeyHeader(name="FiltrAuthentication", auto_error=False) async def verify_api_key(api_key: str = Security(_api_key_header)) -> None: """Validate the FiltrAuthentication header against FILTR_API_KEY env var. Mirrors IsValidFiltrApiKey() in Sony.Filtr.AdminAPI/Config/Bootstrapper.cs. Uses constant-time comparison to prevent timing-based key enumeration. When FILTR_API_KEY is not configured the dependency is a no-op (local dev / tests). """ configured_key = settings.FILTR_API_KEY if not configured_key: return if not api_key or not secrets.compare_digest(api_key, configured_key): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing FiltrAuthentication key", headers={"WWW-Authenticate": "ApiKey"}, ) app = FastAPI(title="Playlist Duplication API") async def _exception_handler(request: Request, call_next: object) -> object: """Catch unhandled exceptions and return a plain 500. Sentry captures the exception automatically via SentryAsgiMiddleware. Mirrors the default exception_handler pattern in ows-moneyhub and ows-data-export. """ try: return await call_next(request) # type: ignore[operator] except Exception: return JSONResponse( content={"detail": "Internal server error"}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) app.add_middleware(BaseHTTPMiddleware, dispatch=_exception_handler) # type: ignore[arg-type] @app.middleware("http") async def lowercase_path_middleware(request: Request, call_next: object) -> object: """Normalise URL path to lowercase before routing. Makes all route paths case-insensitive (e.g. /PLAYLISTSYNC == /PlaylistSync == /playlistsync). Mirrors the default behaviour of the .NET Nancy framework used in the original service. Only the path is lowercased; query parameters, headers, and bodies are untouched. """ request.scope["path"] = request.scope["path"].lower() if "raw_path" in request.scope: request.scope["raw_path"] = request.scope["path"].encode("latin-1") return await call_next(request) # type: ignore[operator] # Router that enforces API key auth on all routes registered to it. # /health is registered directly on `app` above to remain auth-exempt. protected = APIRouter(dependencies=[Depends(verify_api_key)]) if settings.SENTRY_DSN: sentry_sdk.init( dsn=settings.SENTRY_DSN, environment=settings.SENTRY_ENVIRONMENT, traces_sample_rate=0.0, ) app.add_middleware(SentryAsgiMiddleware) # --------------------------------------------------------------------------- # Request / response schemas # --------------------------------------------------------------------------- def _bit_to_bool(v: object) -> bool: r"""Convert a MySQL BIT(1) value to bool. aiomysql returns BIT(1) columns as bytes (b'\x00' = false, b'\x01' = true). Plain bool() is wrong here: bool(b'\x00') == True because non-empty bytes are truthy. """ if isinstance(v, (bytes, bytearray)): return v not in (b"\x00", b"") return bool(v) class AppMarketResponse(BaseModel): """Response schema for GET /apollo-api/app-markets/. Mirrors the ApplicationMarketSchema serializer in apollo-api. strServiceList is stored as a comma-separated string and is returned as a list. """ model_config = ConfigDict(from_attributes=True) id: Optional[int] = None name: Optional[str] = None cultureInfo: Optional[str] = None spotifyRegionCode: Optional[str] = None active: bool = True gaCountryName: Optional[str] = None defaultService: Optional[str] = None services: List[str] = [] workoutMarket: bool = False includeOtherPlaylists: bool = False @classmethod def from_application(cls, app: "Application") -> "AppMarketResponse": raw_services = app.service_list or "" services = [s for s in raw_services.split(",") if s] if raw_services else [] return cls( id=app.id, name=app.name, cultureInfo=app.language_id, spotifyRegionCode=app.spotify_region_code, active=app.active, gaCountryName=app.ga_country_name, defaultService=app.default_service, services=services, workoutMarket=_bit_to_bool(app.workout_market), includeOtherPlaylists=_bit_to_bool(app.include_other_playlists), ) class PlaylistSyncCreate(BaseModel): """Request body for POST /PlaylistSync/{country_code}/playlists. Accepts both camelCase and snake_case field names in the request body. """ model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) from_playlist_id: str to_playlist_id: str to_service_account_id: int title: Optional[str] = None description: Optional[str] = None title_copy_mode: int = 0 description_copy_mode: int = 0 append_track_list: bool = False active: bool = True class PlaylistSyncUpdate(BaseModel): """Request body for PUT /PlaylistSync/{country_code}/playlists/{sync_id}. Only fields in sync_service._UPDATABLE_FIELDS are applied. Accepts both camelCase and snake_case field names in the request body. """ model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) from_playlist_id: Optional[str] = None to_playlist_id: Optional[str] = None to_service_account_id: Optional[int] = None title: Optional[str] = None description: Optional[str] = None active: Optional[bool] = None title_copy_mode: Optional[int] = None description_copy_mode: Optional[int] = None def _fmt_dt(dt: Optional[datetime]) -> Optional[str]: """Format a datetime as ISO 8601 with Z UTC suffix (matching .NET API output).""" if dt is None: return None return dt.strftime("%Y-%m-%dT%H:%M:%SZ") class InsertMediaResponse(BaseModel): """Insert-media item embedded in PlaylistSyncResponse.""" model_config = ConfigDict(from_attributes=True, populate_by_name=True) mediaId: str insertPosition: int class PlaylistSyncResponse(BaseModel): """Rich playlist sync response merging sync task, log, service account, and media. Mirrors the .NET API response shape returned by playlist sync GET endpoints. Fields from the latest sync log are null when no run has been recorded yet. """ # From PlaylistSynchronization id: Optional[int] = None applicationId: Optional[int] = None fromPlaylistId: str = "" fromServiceType: int = 0 fromMusicServiceId: int = 1 toPlaylistId: str = "" toServiceAccountId: int = 0 title: Optional[str] = None description: Optional[str] = None active: bool = True createdAt: Optional[str] = None lastUpdated: Optional[str] = None titleCopyMode: int = 0 descriptionCopyMode: int = 0 appendTrackList: bool = False sourceTitle: Optional[str] = None sourceImage: Optional[str] = None sourceTrackCount: int = 0 synchronizedTrackCount: int = 0 error: bool = False sourceServiceAccountId: Optional[str] = None sourceServiceAccountName: Optional[str] = None toPlaylistTitle: Optional[str] = None toPlaylistImage: Optional[str] = None # From service account toServiceType: Optional[int] = None # From latest sync log (null until first sync run) time: Optional[str] = None madeChange: Optional[bool] = None addedTracks: Optional[int] = None deletedTracks: Optional[int] = None deletedDuplicates: Optional[int] = None syncedTrackCount: Optional[int] = None triggeredManually: Optional[bool] = None errorText: Optional[str] = None errorType: Optional[str] = None # From tblPlaylistSynchronizationInsertMedia insertMedia: List[InsertMediaResponse] = [] def _build_sync_response( sync: PlaylistSynchronization, *, log: Optional[PlaylistSynchronizationLog] = None, to_service_type: Optional[int] = None, insert_media: Optional[List[InsertMedia]] = None, ) -> PlaylistSyncResponse: """Build a PlaylistSyncResponse from a sync task and its related data.""" resp = PlaylistSyncResponse( id=sync.id, applicationId=sync.application_id, fromPlaylistId=sync.from_playlist_id, fromServiceType=sync.from_service_type, fromMusicServiceId=sync.from_music_service_id, toPlaylistId=sync.to_playlist_id, toServiceAccountId=sync.to_service_account_id, title=sync.title, description=sync.description, active=sync.active, createdAt=_fmt_dt(sync.created_at), lastUpdated=_fmt_dt(sync.last_updated), titleCopyMode=sync.title_copy_mode, descriptionCopyMode=sync.description_copy_mode, appendTrackList=sync.append_track_list, sourceTitle=sync.source_title, sourceImage=sync.source_image, sourceTrackCount=sync.source_track_count, synchronizedTrackCount=sync.synchronized_track_count or 0, error=sync.error, sourceServiceAccountId=sync.source_service_account_id, sourceServiceAccountName=sync.source_service_account_name, toPlaylistTitle=sync.to_playlist_title or ( sync.title if sync.title_copy_mode == int(FieldCopyMode.UseSetting) else None ), toPlaylistImage=sync.to_playlist_image, toServiceType=to_service_type, insertMedia=[ InsertMediaResponse(mediaId=m.media_id, insertPosition=m.insert_position) for m in (insert_media or []) ], ) if log: resp.time = _fmt_dt(log.time) resp.madeChange = log.made_changes resp.addedTracks = log.added_tracks resp.deletedTracks = log.deleted_tracks resp.deletedDuplicates = log.deleted_duplicates resp.syncedTrackCount = log.target_track_count resp.triggeredManually = log.triggered_manually resp.errorText = log.error_message resp.errorType = log.error return resp async def _enrich_syncs( syncs: List[PlaylistSynchronization], task_service: "SyncTaskService", account_service: "ServiceAccountService", ) -> List[PlaylistSyncResponse]: """Batch-fetch related data and build PlaylistSyncResponse list.""" if not syncs: return [] sync_ids = [s.id for s in syncs if s.id is not None] account_ids = list({s.to_service_account_id for s in syncs}) latest_logs = await task_service.get_latest_logs_for_sync_ids(sync_ids) accounts = await account_service.get_by_ids(account_ids) insert_media_map = await task_service.get_insert_media_for_sync_ids(sync_ids) return [ _build_sync_response( sync, log=latest_logs.get(sync.id) if sync.id else None, to_service_type=accounts[sync.to_service_account_id].service_type if sync.to_service_account_id in accounts else None, insert_media=insert_media_map.get(sync.id, []) if sync.id else [], ) for sync in syncs ] # --------------------------------------------------------------------------- # Service Account schemas # --------------------------------------------------------------------------- _SERVICE_TYPE_TO_MUSIC_SERVICE: dict[int, int] = { int(ServiceType.Spotify): int(MusicService.Spotify), int(ServiceType.Deezer): int(MusicService.Deezer), int(ServiceType.YouTube): int(MusicService.YouTube), int(ServiceType.SoundCloud): int(MusicService.SoundCloud), } _MUSIC_SERVICE_TO_SERVICE_TYPE: dict[int, int] = { v: k for k, v in _SERVICE_TYPE_TO_MUSIC_SERVICE.items() } class ServiceAccountCreate(BaseModel): """Request body for POST /serviceAccounts/{country_code}. Either service_type or music_service_id must be provided; the other is derived automatically. Accepts both camelCase and snake_case field names in the request. """ model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) service_type: Optional[int] = None music_service_id: Optional[int] = None display_name: Optional[str] = None user_identifier: Annotated[str, BeforeValidator(lambda v: str(v))] access_token: Optional[str] = None access_token_expiry: Optional[int] = None refresh_token: Optional[str] = None channel_name: Optional[str] = None class ServiceAccountUpdate(BaseModel): """Request body for PUT /serviceAccounts/{country_code}/{service_account_id}. service_type, music_service_id, and application_id are immutable after creation. Accepts both camelCase and snake_case field names in the request body. """ model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) display_name: Optional[str] = None user_identifier: Optional[Annotated[str, BeforeValidator(lambda v: str(v))]] = None access_token: Optional[str] = None access_token_expiry: Optional[int] = None refresh_token: Optional[str] = None channel_name: Optional[str] = None class ServiceAccountResponse(BaseModel): """Service account response without sensitive tokens.""" model_config = ConfigDict( from_attributes=True, populate_by_name=True, alias_generator=to_camel ) id: int service_type: int music_service_id: int display_name: Optional[str] = None user_identifier: str updated_date: Optional[datetime] = None application_id: Optional[int] = None @field_serializer("updated_date") def serialize_updated_date(self, v: Optional[datetime]) -> Optional[str]: """Format datetime as ISO 8601 UTC with Z suffix (matches .NET API output).""" if v is None: return None return v.strftime("%Y-%m-%dT%H:%M:%SZ") class ServiceAccountWithTokensResponse(ServiceAccountResponse): """Service account response including OAuth tokens (only when explicitly needed).""" access_token: Optional[str] = None refresh_token: Optional[str] = None # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/health", tags=["Health"]) async def health_check(session: AsyncSession = Depends(get_session)) -> JSONResponse: """Deep health check — verifies database and Redis connectivity. Returns 200 when all checks pass, 503 when any dependency is degraded. Each component reports its own status so clients can identify the failing layer. """ checks: dict = {} healthy = True # --- Database --- try: await session.execute(text("SELECT 1")) checks["database"] = "ok" except Exception as exc: checks["database"] = f"error: {exc}" healthy = False # --- Redis --- try: import redis.asyncio as aioredis r = await aioredis.from_url(settings.REDIS_URL, socket_connect_timeout=2) await r.ping() await r.aclose() checks["redis"] = "ok" except Exception as exc: checks["redis"] = f"error: {exc}" healthy = False body = {"status": "healthy" if healthy else "degraded", **checks} http_status = status.HTTP_200_OK if healthy else status.HTTP_503_SERVICE_UNAVAILABLE return JSONResponse(content=body, status_code=http_status) @protected.get( "/apollo-api/app-markets/", response_model=List[AppMarketResponse], tags=["App Markets"], ) async def get_app_markets( session: AsyncSession = Depends(get_session), ) -> List[AppMarketResponse]: """Return all application market rows from tblApplicationInstance. Mirrors GET /apollo-api/app-markets/ in apollo-api (ApplicationMarkets.get). """ apps = await ApplicationService(session).get_all() return [AppMarketResponse.from_application(a) for a in apps] @protected.get( "/playlistsync/playlists", response_model=List[PlaylistSyncResponse], tags=["Playlist Sync"], ) async def list_all_syncs( session: AsyncSession = Depends(get_session), ) -> List[PlaylistSyncResponse]: """Return all playlist syncs (active and inactive).""" task_service = SyncTaskService(session) syncs = await task_service.get_all_no_filter() return await _enrich_syncs(syncs, task_service, ServiceAccountService(session)) @protected.get( "/playlistsync/{country_code}/playlists", response_model=List[PlaylistSyncResponse], tags=["Playlist Sync"], ) async def get_syncs_by_market( country_code: str, session: AsyncSession = Depends(get_session) ) -> List[PlaylistSyncResponse]: """Return all syncs (active and inactive) for a market, with latest log data.""" app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) task_service = SyncTaskService(session) syncs = await task_service.get_by_application_id_all(app_obj.id) # type: ignore[arg-type] return await _enrich_syncs(syncs, task_service, ServiceAccountService(session)) @protected.get( "/playlistsync/{country_code}/playlists/{sync_id}", response_model=PlaylistSyncResponse, tags=["Playlist Sync"], ) async def get_single_sync( country_code: str, sync_id: int, session: AsyncSession = Depends(get_session) ) -> PlaylistSyncResponse: app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) task_service = SyncTaskService(session) sync = await task_service.get_by_id(sync_id) if not sync: raise HTTPException(status_code=404, detail=f"Sync with id {sync_id} not found") if sync.application_id != app_obj.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Sync {sync_id} does not belong to application '{country_code}'", ) responses = await _enrich_syncs( [sync], task_service, ServiceAccountService(session) ) return responses[0] @protected.post( "/playlistsync/{country_code}/playlists", response_model=PlaylistSynchronization, status_code=status.HTTP_201_CREATED, tags=["Playlist Sync"], ) async def create_sync( country_code: str, sync_data: PlaylistSyncCreate, session: AsyncSession = Depends(get_session), ) -> PlaylistSynchronization: """Create a new playlist synchronization task. Validates that no active sync already targets the same to_playlist_id. application_id is derived from country_code; currently defaults to 0 until ApplicationInstanceManager integration is complete. """ service = SyncTaskService(session) existing = await service.get_by_to_playlist_id(sync_data.to_playlist_id) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"Sync to playlist {sync_data.to_playlist_id} already exists" f" (id: {existing.id})" ), ) app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) application_id = app_obj.id new_sync = PlaylistSynchronization( application_id=application_id, from_playlist_id=sync_data.from_playlist_id, from_service_type=int(ServiceType.Spotify), from_music_service_id=int(MusicService.Spotify), to_playlist_id=sync_data.to_playlist_id, to_service_account_id=sync_data.to_service_account_id, title=sync_data.title, description=sync_data.description, title_copy_mode=sync_data.title_copy_mode, description_copy_mode=sync_data.description_copy_mode, append_track_list=sync_data.append_track_list, created_at=datetime.now(timezone.utc), active=sync_data.active, to_playlist_title=sync_data.title if sync_data.title_copy_mode == int(FieldCopyMode.UseSetting) and sync_data.title else None, ) return await service.create_sync(new_sync) @protected.put( "/playlistsync/{country_code}/playlists/{sync_id}", response_model=PlaylistSynchronization, tags=["Playlist Sync"], ) async def update_sync( country_code: str, sync_id: int, update_data: PlaylistSyncUpdate, session: AsyncSession = Depends(get_session), ) -> PlaylistSynchronization: app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) service = SyncTaskService(session) existing_sync = await service.get_by_id(sync_id) if not existing_sync: raise HTTPException(status_code=404, detail="Sync task not found") if existing_sync.application_id != app_obj.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( f"Application does not have access to playlist sync with id {sync_id}" ), ) if update_data.to_playlist_id: existing = await service.get_by_to_playlist_id(update_data.to_playlist_id) if existing and existing.id != sync_id and existing.active: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"Active sync to playlist {update_data.to_playlist_id}" f" already exists (id: {existing.id})" ), ) updated = await service.update_sync( sync_id, update_data.model_dump(exclude_none=True) ) if not updated: raise HTTPException(status_code=404, detail="Sync task not found") return updated @protected.delete( "/playlistsync/{country_code}/playlists/{sync_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Playlist Sync"], ) async def delete_sync( country_code: str, sync_id: int, session: AsyncSession = Depends(get_session) ) -> None: app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) service = SyncTaskService(session) existing_sync = await service.get_by_id(sync_id) if not existing_sync: raise HTTPException(status_code=404, detail="Sync task not found") if existing_sync.application_id != app_obj.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( f"Application does not have access to playlist sync with id {sync_id}" ), ) await service.delete_sync(sync_id) @protected.post( "/playlistsync/{country_code}/playlists/{sync_id}/execute", tags=["Playlist Sync"] ) async def execute_sync( country_code: str, sync_id: int, force_all_sources: bool = Query(default=False, alias="forceAllSources"), session: AsyncSession = Depends(get_session), ) -> dict: """Trigger an immediate synchronization for a specific sync task. Validates the sync exists and belongs to the application, then dispatches a Celery task. The Celery task performs the actual platform API calls asynchronously. When ``forceAllSources=true`` is passed, all *active* sync records that share the same source playlist (``from_playlist_id``) as ``sync_id`` are dispatched — not just the one record. This powers the "Sync All Related" button on the frontend. """ app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) service = SyncTaskService(session) sync = await service.get_by_id(sync_id) if not sync: raise HTTPException(status_code=404, detail=f"Sync with id {sync_id} not found") if sync.application_id != app_obj.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( f"Application does not have access to playlist sync with id {sync_id}" ), ) from worker.tasks import execute_single_sync_task if force_all_sources: related = await service.get_active_by_from_playlist_id(sync.from_playlist_id) for s in related: execute_single_sync_task.delay(s.id, triggered_manually=True) return { "message": ( f"Execution triggered for {len(related)} sync(s) sharing" f" source playlist '{sync.from_playlist_id}'" ) } execute_single_sync_task.delay(sync_id, triggered_manually=True) return { "message": f"Execution of sync {sync_id} triggered for market {country_code}" } @protected.get( "/playlistsync/{country_code}/playlists/{sync_id}/log", response_model=List[PlaylistSynchronizationLog], tags=["Playlist Sync"], ) async def get_sync_logs( country_code: str, sync_id: int, limit: int = Query(default=100, ge=1, le=1000), offset: int = Query(default=0, ge=0), session: AsyncSession = Depends(get_session), ) -> List[PlaylistSynchronizationLog]: service = SyncLogService(session) return await service.get_logs_by_sync_id(sync_id, limit=limit, offset=offset) # --------------------------------------------------------------------------- # Service Account endpoints # --------------------------------------------------------------------------- @protected.get( "/serviceaccounts", response_model=List[ServiceAccountResponse], response_model_by_alias=True, tags=["Service Accounts"], ) async def list_all_service_accounts( session: AsyncSession = Depends(get_session), ) -> List[ServiceAccountResponse]: """Return all service accounts (without tokens).""" accounts = await ServiceAccountService(session).get_all() return [ServiceAccountResponse.model_validate(a) for a in accounts] @protected.get( "/serviceaccounts/{country_code}", response_model=List[ServiceAccountResponse], response_model_by_alias=True, tags=["Service Accounts"], ) async def list_service_accounts_by_market( country_code: str, include_tokens: bool = Query(default=False), session: AsyncSession = Depends(get_session), ) -> List[ServiceAccountResponse]: """Return service accounts for a market. Pass ``?include_tokens=true`` to include OAuth tokens in the response. """ app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) accounts = await ServiceAccountService(session).get_by_application_id(app_obj.id) # type: ignore[arg-type] if include_tokens: return [ServiceAccountWithTokensResponse.model_validate(a) for a in accounts] return [ServiceAccountResponse.model_validate(a) for a in accounts] @protected.post( "/serviceaccounts/{country_code}", response_model=ServiceAccountResponse, response_model_by_alias=True, status_code=status.HTTP_201_CREATED, tags=["Service Accounts"], ) async def create_service_account( country_code: str, data: ServiceAccountCreate, session: AsyncSession = Depends(get_session), ) -> ServiceAccountResponse: """Create a new service account for a market. Either service_type or music_service_id must be provided; the other is derived automatically. Returns 409 if a service account with the same (music_service_id, user_identifier) already exists. """ app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) service_type = data.service_type music_service_id = data.music_service_id if music_service_id is not None: derived_service_type = _MUSIC_SERVICE_TO_SERVICE_TYPE.get(music_service_id) if derived_service_type is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unknown music_service_id: {music_service_id}", ) service_type = derived_service_type elif service_type is not None: derived_music_service_id = _SERVICE_TYPE_TO_MUSIC_SERVICE.get(service_type) if derived_music_service_id is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unknown service_type: {service_type}", ) music_service_id = derived_music_service_id else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Either service_type or music_service_id must be provided.", ) svc = ServiceAccountService(session) existing = await svc.get_by_music_service_and_user_identifier( music_service_id, data.user_identifier ) if existing: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"Service account with user_identifier '{data.user_identifier}'" f" already exists for music_service_id {music_service_id}." ), ) account = ServiceAccount( service_type=service_type, music_service_id=music_service_id, display_name=data.display_name, user_identifier=data.user_identifier, access_token=data.access_token, access_token_expiry=data.access_token_expiry, refresh_token=data.refresh_token, channel_name=data.channel_name, application_id=app_obj.id, updated_date=datetime.now(timezone.utc), ) created = await svc.create(account) return ServiceAccountResponse.model_validate(created) @protected.put( "/serviceaccounts/{country_code}/{service_account_id}", response_model=ServiceAccountResponse, response_model_by_alias=True, tags=["Service Accounts"], ) async def update_service_account( country_code: str, service_account_id: int, data: ServiceAccountUpdate, session: AsyncSession = Depends(get_session), ) -> ServiceAccountResponse: """Update a service account. Note: service_type, music_service_id, and application_id are immutable. """ app_obj = await ApplicationService(session).get_by_country_code(country_code) if app_obj is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"No application found for country code '{country_code}'", ) svc = ServiceAccountService(session) account = await svc.get_by_id(service_account_id) if account is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Service account {service_account_id} not found", ) for field, value in data.model_dump(exclude_none=True).items(): setattr(account, field, value) account.updated_date = datetime.now(timezone.utc) saved = await svc.save(account) return ServiceAccountResponse.model_validate(saved) @protected.delete( "/serviceaccounts/{service_account_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Service Accounts"], ) async def delete_service_account( service_account_id: int, session: AsyncSession = Depends(get_session), ) -> None: """Delete a service account by ID.""" svc = ServiceAccountService(session) account = await svc.get_by_id(service_account_id) if account is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Service account {service_account_id} not found", ) await svc.delete(service_account_id) app.include_router(protected)