From b4d1e8f75503a4f0dbf4ff526dda3219a316ae36 Mon Sep 17 00:00:00 2001 From: NixReaper Date: Sat, 5 Sep 2026 02:03:25 -0500 Subject: [PATCH] Add opt-in season cascade to series ingestion Ingesting a series only ever produced season stubs -- getting actual episodes meant calling POST /admin/ingest/season once per season by hand, which is exactly what manual end-to-end testing kept requiring for every multi-season show. POST /api/v1/admin/ingest/series now takes include_seasons (default false, since a show can have dozens of seasons and that's dozens of TMDb calls fired from one request). When true, ingest_series_task fans out one ingest_season job per season once the series succeeds, using arq's own ctx['redis'] connection for the job (the same pattern arq's docs recommend for job chaining) rather than a second unrelated connection. Each fanned-out season job still goes through get_or_create_ingestion_job, so re-running a cascaded series ingestion reuses the same season job rows instead of creating duplicates. Co-Authored-By: Claude Sonnet 5 --- app/api/v1/admin.py | 8 +- app/services/ingestion/runner.py | 12 +++ app/workers/tasks.py | 24 ++++- docs/api.md | 2 +- docs/database.md | 9 +- tests/integration/test_ingestion_cascade.py | 99 +++++++++++++++++++++ 6 files changed, 148 insertions(+), 6 deletions(-) create mode 100644 tests/integration/test_ingestion_cascade.py diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index 32eaf8b..6a2bec3 100644 --- a/app/api/v1/admin.py +++ b/app/api/v1/admin.py @@ -36,6 +36,12 @@ class IngestPersonRequest(BaseModel): 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): @@ -120,7 +126,7 @@ async def enqueue_series_ingestion( entity_type="series", external_id=body.tmdb_id, task_name="ingest_series_task", - task_args=(body.tmdb_id,), + task_args=(body.tmdb_id, body.include_seasons), ) diff --git a/app/services/ingestion/runner.py b/app/services/ingestion/runner.py index 3250052..6e76b8c 100644 --- a/app/services/ingestion/runner.py +++ b/app/services/ingestion/runner.py @@ -25,6 +25,7 @@ from app.repositories.provider_repository import ( mark_job_running, mark_job_succeeded, ) +from app.repositories.tv_repository import get_seasons_for_series from app.services.ingestion.movie_ingestion import ingest_movie from app.services.ingestion.person_ingestion import ingest_person from app.services.ingestion.series_ingestion import ingest_season, ingest_series @@ -138,3 +139,14 @@ async def run_season_ingestion( session, provider, series_external_id, season_number, ingestion_job_id=job_id ), ) + + +async def get_series_season_numbers(series_id: uuid.UUID) -> list[int]: + """All season numbers a just-ingested series has, for cascading into + per-season ingestion jobs (see `app/workers/tasks.py:ingest_series_task`, + `include_seasons`). A short-lived session of its own since this runs + after `run_series_ingestion`'s own session has already closed.""" + + async with session_scope() as session: + seasons = await get_seasons_for_series(session, series_id) + return [season.season_number for season in seasons] diff --git a/app/workers/tasks.py b/app/workers/tasks.py index a6ef234..6e55904 100644 --- a/app/workers/tasks.py +++ b/app/workers/tasks.py @@ -14,6 +14,8 @@ from typing import Any from app.core.logging import get_logger from app.services.ingestion.runner import ( + get_or_create_ingestion_job, + get_series_season_numbers, run_movie_ingestion, run_person_ingestion, run_season_ingestion, @@ -33,8 +35,28 @@ async def ingest_person_task(ctx: dict[str, Any], job_id: str, external_id: str) return str(person_id) -async def ingest_series_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str: +async def ingest_series_task( + ctx: dict[str, Any], job_id: str, external_id: str, include_seasons: bool = False +) -> str: series_id = await run_series_ingestion(uuid.UUID(job_id), external_id) + + if include_seasons: + # Fan out one ingest_season job per season this series has, using + # the same arq connection this task is running on (arq hands every + # job's ctx a "redis" entry precisely for chaining/fan-out like + # this) — not a second, unrelated connection. + redis = ctx["redis"] + for season_number in await get_series_season_numbers(series_id): + season_job_id = await get_or_create_ingestion_job( + "season_sync", "season", f"{external_id}:{season_number}" + ) + await redis.enqueue_job( + "ingest_season_task", str(season_job_id), external_id, season_number + ) + logger.info( + "series_seasons_cascaded", series_id=str(series_id), external_id=external_id + ) + return str(series_id) diff --git a/docs/api.md b/docs/api.md index 63e9874..dce8fc2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -120,7 +120,7 @@ you want network-level isolation too. |---|---|---|---| | POST | `/api/v1/admin/ingest/movie` | `{"tmdb_id": "603"}` | Enqueues an arq job; returns `{"job_id": "...", "status": "enqueued"}` immediately (section 15: ingestion never runs inline in a request). | | POST | `/api/v1/admin/ingest/person` | `{"tmdb_id": "6384"}` | Same, for a person profile. | -| POST | `/api/v1/admin/ingest/series` | `{"tmdb_id": "1396"}` | Ingests the series and season *stubs* (number, title, air date, poster) — no episodes yet. | +| POST | `/api/v1/admin/ingest/series` | `{"tmdb_id": "1396", "include_seasons": false}` | Ingests the series and season *stubs* (number, title, air date, poster) — no episodes yet. With `include_seasons: true`, also enqueues an `ingest/season` job for every season once the series finishes (opt-in: some shows have dozens of seasons, each a separate TMDb call). | | POST | `/api/v1/admin/ingest/season` | `{"tmdb_id": "1396", "season_number": 1}` | Ingests one season's full episode list. Requires the series to already be ingested (`ingest/series` first) — fails with a clear error otherwise. | | GET | `/api/v1/admin/ingest/jobs/{job_id}` | — | Poll job status: `pending` → `running` → `succeeded`/`failed`/`retrying`. | diff --git a/docs/database.md b/docs/database.md index 836dd4c..f19c6fb 100644 --- a/docs/database.md +++ b/docs/database.md @@ -171,6 +171,9 @@ indexes on the alternate-title tables. See - `ingest_series` only upserts season *stubs* (number, title, air date, poster) — episodes require a separate `ingest_season` call per season (TMDb has no single endpoint returning a whole series' episodes), and - that call requires the series to already exist. There's no orchestration - yet that walks all of a series' seasons automatically after - `ingest_series` succeeds. + that call requires the series to already exist. `POST + /api/v1/admin/ingest/series` can cascade this automatically + (`include_seasons: true` fans out one `ingest_season` job per season from + `ingest_series_task` once the series succeeds, via arq's own `ctx['redis']` + — see `app/workers/tasks.py`), but it's opt-in: some shows have dozens of + seasons, each a separate TMDb call, so cascading isn't the default. diff --git a/tests/integration/test_ingestion_cascade.py b/tests/integration/test_ingestion_cascade.py new file mode 100644 index 0000000..585cbd3 --- /dev/null +++ b/tests/integration/test_ingestion_cascade.py @@ -0,0 +1,99 @@ +"""Tests for the series -> season ingestion cascade +(app/workers/tasks.py:ingest_series_task, include_seasons=True). + +run_series_ingestion and get_series_season_numbers are monkeypatched so this +doesn't need a real TMDbProvider/network call — the thing actually at risk +here is the fan-out wiring itself (does it enqueue the right task with the +right arguments for every season), not the ingestion pipeline, which +test_series_ingestion.py already covers. +""" + +from __future__ import annotations + +import uuid + +import pytest + +from app.workers import tasks + + +class _FakeArqRedis: + def __init__(self) -> None: + self.enqueued: list[tuple[str, tuple]] = [] + + async def enqueue_job(self, task_name: str, *args): + self.enqueued.append((task_name, args)) + return object() # stand-in for arq's Job — callers don't use it here + + +@pytest.mark.asyncio +async def test_series_task_does_not_cascade_by_default(monkeypatch): + fake_series_id = uuid.uuid4() + + async def fake_run_series_ingestion(job_id, external_id): + return fake_series_id + + monkeypatch.setattr(tasks, "run_series_ingestion", fake_run_series_ingestion) + + redis = _FakeArqRedis() + result = await tasks.ingest_series_task({"redis": redis}, str(uuid.uuid4()), "1396") + + assert result == str(fake_series_id) + assert redis.enqueued == [] + + +@pytest.mark.asyncio +async def test_series_task_cascades_one_season_job_per_season(monkeypatch, db_session): + fake_series_id = uuid.uuid4() + + async def fake_run_series_ingestion(job_id, external_id): + return fake_series_id + + async def fake_get_series_season_numbers(series_id): + assert series_id == fake_series_id + return [0, 1, 2] + + monkeypatch.setattr(tasks, "run_series_ingestion", fake_run_series_ingestion) + monkeypatch.setattr(tasks, "get_series_season_numbers", fake_get_series_season_numbers) + + redis = _FakeArqRedis() + result = await tasks.ingest_series_task( + {"redis": redis}, str(uuid.uuid4()), "1396", include_seasons=True + ) + + assert result == str(fake_series_id) + assert len(redis.enqueued) == 3 + for task_name, args in redis.enqueued: + assert task_name == "ingest_season_task" + job_id_arg, external_id_arg, season_number_arg = args + assert external_id_arg == "1396" + uuid.UUID(job_id_arg) # must be a real, parseable job id + + seasons_enqueued = sorted(args[2] for _, args in redis.enqueued) + assert seasons_enqueued == [0, 1, 2] + + +@pytest.mark.asyncio +async def test_series_task_cascade_reuses_job_on_repeat_enqueue(monkeypatch, db_session): + """Re-cascading the same series (e.g. a retried job) must reuse the same + season IngestionJob rows, not create duplicates — get_or_create_ingestion_job + is idempotency-keyed by (entity_type, external_id).""" + + fake_series_id = uuid.uuid4() + + async def fake_run_series_ingestion(job_id, external_id): + return fake_series_id + + async def fake_get_series_season_numbers(series_id): + return [1] + + monkeypatch.setattr(tasks, "run_series_ingestion", fake_run_series_ingestion) + monkeypatch.setattr(tasks, "get_series_season_numbers", fake_get_series_season_numbers) + + redis = _FakeArqRedis() + await tasks.ingest_series_task({"redis": redis}, str(uuid.uuid4()), "9999", include_seasons=True) + await tasks.ingest_series_task({"redis": redis}, str(uuid.uuid4()), "9999", include_seasons=True) + + first_job_id = redis.enqueued[0][1][0] + second_job_id = redis.enqueued[1][1][0] + assert first_job_id == second_job_id