"""Administrative/provider endpoints (section 17): trigger ingestion and check job status. Separately protected by an admin API key — never mounted on the public read path. """ from __future__ import annotations import uuid from arq import ArqRedis from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import not_found from app.core.database import get_db from app.core.queue import get_queue from app.core.security import require_admin_api_key from app.repositories.provider_repository import get_ingestion_job from app.services.ingestion.runner import get_or_create_ingestion_job router = APIRouter( prefix="/admin", tags=["admin"], dependencies=[Depends(require_admin_api_key)], ) class IngestMovieRequest(BaseModel): tmdb_id: str class IngestPersonRequest(BaseModel): tmdb_id: str class IngestSeriesRequest(BaseModel): tmdb_id: str # When true, ingest_series_task also enqueues an ingest_season job for # every season the series has, once the series itself finishes. # Defaults to false: a show can have dozens of seasons, and cascading # automatically means dozens of TMDb calls fired off from one request — # opt in rather than surprise the caller with that. include_seasons: bool = False class IngestSeasonRequest(BaseModel): tmdb_id: str season_number: int class IngestCollectionRequest(BaseModel): tmdb_id: str class IngestionEnqueuedOut(BaseModel): job_id: str arq_job_id: str status: str = "enqueued" class IngestionJobOut(BaseModel): id: str job_type: str entity_type: str | None external_id: str | None status: str attempts: int error: str | None result_aurora_id: str | None async def _enqueue( queue: ArqRedis, *, job_type: str, entity_type: str, external_id: str, task_name: str, task_args: tuple = (), ) -> IngestionEnqueuedOut: # The IngestionJob row is created synchronously (and idempotently — the # same external_id reuses the same row) so the response's job_id is # immediately valid against GET /ingest/jobs/{job_id}. arq's own job id # is a separate, unrelated identifier used only for queue bookkeeping. job_id = await get_or_create_ingestion_job(job_type, entity_type, external_id) arq_job = await queue.enqueue_job(task_name, str(job_id), *task_args) if arq_job is None: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue ingestion job (queue busy)." ) return IngestionEnqueuedOut(job_id=str(job_id), arq_job_id=arq_job.job_id) @router.post("/ingest/movie", response_model=IngestionEnqueuedOut) async def enqueue_movie_ingestion( body: IngestMovieRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: return await _enqueue( queue, job_type="movie_sync", entity_type="movie", external_id=body.tmdb_id, task_name="ingest_movie_task", task_args=(body.tmdb_id,), ) @router.post("/ingest/person", response_model=IngestionEnqueuedOut) async def enqueue_person_ingestion( body: IngestPersonRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: return await _enqueue( queue, job_type="person_sync", entity_type="person", external_id=body.tmdb_id, task_name="ingest_person_task", task_args=(body.tmdb_id,), ) @router.post("/ingest/series", response_model=IngestionEnqueuedOut) async def enqueue_series_ingestion( body: IngestSeriesRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: return await _enqueue( queue, job_type="series_sync", entity_type="series", external_id=body.tmdb_id, task_name="ingest_series_task", task_args=(body.tmdb_id, body.include_seasons), ) @router.post("/ingest/season", response_model=IngestionEnqueuedOut) async def enqueue_season_ingestion( body: IngestSeasonRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: # Season ingestion requires the series to already be ingested — see # app/services/ingestion/series_ingestion.py:ingest_season. return await _enqueue( queue, job_type="season_sync", entity_type="season", external_id=f"{body.tmdb_id}:{body.season_number}", task_name="ingest_season_task", task_args=(body.tmdb_id, body.season_number), ) @router.post("/ingest/collection", response_model=IngestionEnqueuedOut) async def enqueue_collection_ingestion( body: IngestCollectionRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: # Membership only links movies already ingested into Aurora (never a # stub) -- see app/services/ingestion/collection_ingestion.py -- so # ingesting the franchise's movies first (or re-running this afterward) # is what fills the list in. return await _enqueue( queue, job_type="collection_sync", entity_type="collection", external_id=body.tmdb_id, task_name="ingest_collection_task", task_args=(body.tmdb_id,), ) @router.get("/ingest/jobs/{job_id}", response_model=IngestionJobOut) async def get_ingestion_job_status( job_id: uuid.UUID, session: AsyncSession = Depends(get_db) ) -> IngestionJobOut: job = await get_ingestion_job(session, job_id) if job is None: raise not_found("ingestion_job", str(job_id)) return IngestionJobOut( id=str(job.id), job_type=job.job_type, entity_type=job.entity_type, external_id=job.external_id, status=job.status.value, attempts=job.attempts, error=job.error, result_aurora_id=str(job.result_aurora_id) if job.result_aurora_id else None, )