"""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 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 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 @router.post("/ingest/movie", response_model=IngestionEnqueuedOut) async def enqueue_movie_ingestion( body: IngestMovieRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: job = await queue.enqueue_job("ingest_movie_task", body.tmdb_id) if job is None: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue ingestion job (queue busy)." ) return IngestionEnqueuedOut(job_id=job.job_id, arq_job_id=job.job_id) @router.post("/ingest/person", response_model=IngestionEnqueuedOut) async def enqueue_person_ingestion( body: IngestPersonRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: job = await queue.enqueue_job("ingest_person_task", body.tmdb_id) if job is None: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue ingestion job (queue busy)." ) return IngestionEnqueuedOut(job_id=job.job_id, arq_job_id=job.job_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, )