NixReaperandClaude Sonnet 5
669034896a
CI / lint (push) Canceled after 0s
CI / typecheck (push) Canceled after 0s
CI / unit-tests (push) Canceled after 0s
CI / integration-tests (push) Canceled after 0s
CI / container-build (push) Canceled after 0s
CI / dependency-audit (push) Canceled after 0s
POST /api/v1/admin/ingest/movie returned arq's own internal queue
job id, but GET /api/v1/admin/ingest/jobs/{job_id} looked up our
IngestionJob DB row by its own UUID -- a completely different id that
didn't even exist yet at enqueue time (the row was only created once
the worker picked up the task). Every status poll 404'd.
Fix: create the IngestionJob row synchronously in the enqueue endpoint
(reusing the existing idempotency-key logic, so repeat enqueues for
the same tmdb_id reuse the same row rather than creating duplicates),
pass its real id into the arq task, and have the worker update that
row instead of creating a new one.
Found by actually running the milestone end-to-end: enqueue a real
TMDb ingestion, then poll its status.
Co-Authored-By: Claude Sonnet 5 <[email protected]>
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""Orchestrates one ingestion job end-to-end: create/track the `IngestionJob`
|
|
row, run the provider fetch + normalize + upsert pipeline, and record
|
|
success/failure. Used by both the admin trigger endpoint (section 17) and
|
|
the arq worker task (section 15) so the two entry points share identical
|
|
job-lifecycle and idempotency behavior.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import session_scope
|
|
from app.core.logging import get_logger
|
|
from app.models.provider import IngestionJob
|
|
from app.providers.tmdb import TMDbProvider
|
|
from app.repositories.provider_repository import (
|
|
create_ingestion_job,
|
|
get_ingestion_job,
|
|
get_or_create_provider,
|
|
mark_job_failed,
|
|
mark_job_running,
|
|
mark_job_succeeded,
|
|
)
|
|
from app.services.ingestion.movie_ingestion import ingest_movie
|
|
from app.services.ingestion.person_ingestion import ingest_person
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
async def _require_job(session: AsyncSession, job_id: uuid.UUID) -> IngestionJob:
|
|
job = await get_ingestion_job(session, job_id)
|
|
if job is None:
|
|
# Only possible if the row was deleted out-of-band; the job was
|
|
# created moments earlier in this same call chain.
|
|
raise RuntimeError(f"ingestion job {job_id} vanished mid-run")
|
|
return job
|
|
|
|
|
|
async def get_or_create_ingestion_job(job_type: str, entity_type: str, external_id: str) -> uuid.UUID:
|
|
"""Create the DB-tracked `IngestionJob` row (or return the existing one
|
|
for this provider+entity+external_id, keyed by `idempotency_key`)
|
|
without running the pipeline.
|
|
|
|
Called synchronously from the admin endpoint, *before* enqueuing to arq,
|
|
specifically so the HTTP response can return a `job_id` that
|
|
`GET /admin/ingest/jobs/{job_id}` can immediately resolve — arq's own
|
|
queue-internal job id is a different, unrelated identifier and was
|
|
never valid here.
|
|
"""
|
|
|
|
idempotency_key = f"tmdb:{entity_type}:{external_id}"
|
|
async with session_scope() as session:
|
|
provider_row = await get_or_create_provider(session, "tmdb", "The Movie Database (TMDb)")
|
|
job = await create_ingestion_job(
|
|
session,
|
|
provider_id=provider_row.id,
|
|
job_type=job_type,
|
|
idempotency_key=idempotency_key,
|
|
entity_type=entity_type,
|
|
external_id=external_id,
|
|
)
|
|
return job.id
|
|
|
|
|
|
async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID:
|
|
"""Run the pipeline for an `IngestionJob` row that already exists (see
|
|
`get_or_create_ingestion_job`) — this never creates the row itself."""
|
|
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_running(session, job)
|
|
|
|
provider = TMDbProvider()
|
|
try:
|
|
async with session_scope() as session:
|
|
movie_id = await ingest_movie(session, provider, external_id, ingestion_job_id=job_id)
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_succeeded(session, job, movie_id)
|
|
return movie_id
|
|
except Exception as exc:
|
|
logger.error("movie_ingestion_failed", external_id=external_id, error=str(exc))
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_failed(session, job, str(exc))
|
|
raise
|
|
finally:
|
|
await provider.aclose()
|
|
|
|
|
|
async def run_person_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID:
|
|
"""Run the pipeline for an `IngestionJob` row that already exists (see
|
|
`get_or_create_ingestion_job`) — this never creates the row itself."""
|
|
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_running(session, job)
|
|
|
|
provider = TMDbProvider()
|
|
try:
|
|
async with session_scope() as session:
|
|
person_id = await ingest_person(session, provider, external_id, ingestion_job_id=job_id)
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_succeeded(session, job, person_id)
|
|
return person_id
|
|
except Exception as exc:
|
|
logger.error("person_ingestion_failed", external_id=external_id, error=str(exc))
|
|
async with session_scope() as session:
|
|
job = await _require_job(session, job_id)
|
|
await mark_job_failed(session, job, str(exc))
|
|
raise
|
|
finally:
|
|
await provider.aclose()
|