Updated ingest_wistia_events.py to use a specific manifest_prefix

Author: johnathon-smith

README.md

@@ -13,6 +13,15 @@ workflow. The ingestion job publishes `INGESTION_MANIFEST_URI` for its run. The
 validation job reads that property, so it processes the exact manifest produced by
 the preceding ingestion job.
 
+The ingestion job stores manifests separately from temporary ingestion data. Its
+default manifest prefix is:
+
+```text
+metadata/wistia/events/manifests
+```
+
+Override it with the ingestion job parameter `--MANIFEST_PREFIX`.
+
 For a manual validation run outside the workflow, provide:
 
 ```text

glue_jobs/ingest_wistia_events.py

@@ -8,6 +8,7 @@
 
 Optional arguments:
     --S3_PREFIX          Default: ingestion/wistia/events
+    --MANIFEST_PREFIX    Default: metadata/wistia/events/manifests
     --SECRET_REGION      Default: the Glue job's AWS region
     --WORKFLOW_NAME      Supplied by AWS Glue when run in a workflow
     --WORKFLOW_RUN_ID    Supplied by AWS Glue when run in a workflow
@@ -62,6 +63,7 @@ class JobConfig:
     secret_region: str | None
     s3_bucket: str
     s3_prefix: str
+    manifest_prefix: str
     media_ids: tuple[str, ...]
     workflow_name: str | None
     workflow_run_id: str | None
@@ -109,14 +111,19 @@ def load_config() -> JobConfig:
         ["JOB_NAME", "SECRET_ID", "S3_BUCKET", "MEDIA_IDS"],
     )
     prefix = parse_optional_argument("S3_PREFIX", "ingestion/wistia/events")
+    manifest_prefix = parse_optional_argument(
+        "MANIFEST_PREFIX", "metadata/wistia/events/manifests"
+    )
     assert prefix is not None
+    assert manifest_prefix is not None
 
     return JobConfig(
         job_name=required["JOB_NAME"],
         secret_id=required["SECRET_ID"],
         secret_region=parse_optional_argument("SECRET_REGION"),
         s3_bucket=required["S3_BUCKET"],
         s3_prefix=prefix.strip("/"),
+        manifest_prefix=manifest_prefix.strip("/"),
         media_ids=parse_media_ids(required["MEDIA_IDS"]),
         workflow_name=parse_optional_argument("WORKFLOW_NAME"),
         workflow_run_id=parse_optional_argument("WORKFLOW_RUN_ID"),
@@ -400,8 +407,8 @@ def write_manifest(
     results: list[dict[str, Any]],
 ) -> str:
     manifest_key = (
-        f"{config.s3_prefix}/extraction_date={extraction_time:%Y-%m-%d}/"
-        f"_manifests/manifest_{extraction_time:%Y%m%dT%H%M%SZ}_{run_id}.json"
+        f"{config.manifest_prefix}/extraction_date={extraction_time:%Y-%m-%d}/"
+        f"manifest_{extraction_time:%Y%m%dT%H%M%SZ}_{run_id}.json"
     )
     manifest = {
         "api_url": API_URL,

tests/test_ingest_wistia_events.py

@@ -1,7 +1,9 @@
 import importlib.util
+import json
 import sys
 import types
 import unittest
+from datetime import date, datetime, timezone
 from pathlib import Path
 from unittest.mock import Mock
 
@@ -32,6 +34,7 @@ def config(self, workflow_name=None, workflow_run_id=None):
             secret_region=None,
             s3_bucket="bucket",
             s3_prefix="ingestion/wistia/events",
+            manifest_prefix="metadata/wistia/events/manifests",
             media_ids=("media-1", "media-2"),
             workflow_name=workflow_name,
             workflow_run_id=workflow_run_id,
@@ -74,5 +77,53 @@ def test_partial_workflow_context_fails(self):
             )
 
 
+class ManifestTests(unittest.TestCase):
+    def test_manifest_uses_metadata_prefix(self):
+        config = ingestion.JobConfig(
+            job_name="ingestion",
+            secret_id="secret",
+            secret_region=None,
+            s3_bucket="bucket",
+            s3_prefix="ingestion/wistia/events",
+            manifest_prefix="metadata/wistia/events/manifests",
+            media_ids=("media-1",),
+            workflow_name=None,
+            workflow_run_id=None,
+        )
+        s3_client = Mock()
+        manifest_uri = ingestion.write_manifest(
+            s3_client=s3_client,
+            config=config,
+            extraction_time=datetime(2026, 6, 5, 17, 30, tzinfo=timezone.utc),
+            run_id="run-123",
+            start_date=date(2024, 6, 5),
+            end_date=date(2026, 6, 5),
+            results=[
+                {
+                    "event_count": 1,
+                    "media_id": "media-1",
+                    "page_count": 1,
+                    "s3_uri": "s3://bucket/ingestion/events.jsonl.gz",
+                    "status": "succeeded",
+                }
+            ],
+        )
+
+        self.assertEqual(
+            "s3://bucket/metadata/wistia/events/manifests/"
+            "extraction_date=2026-06-05/"
+            "manifest_20260605T173000Z_run-123.json",
+            manifest_uri,
+        )
+        put_call = s3_client.put_object.call_args.kwargs
+        self.assertEqual("bucket", put_call["Bucket"])
+        self.assertEqual(
+            "metadata/wistia/events/manifests/extraction_date=2026-06-05/"
+            "manifest_20260605T173000Z_run-123.json",
+            put_call["Key"],
+        )
+        self.assertEqual("run-123", json.loads(put_call["Body"])["run_id"])
+
+
 if __name__ == "__main__":
     unittest.main()