US05-02: Orchestrate Album Upload Batches (#72)

photo_pipeline/jobs/domain_handlers.py

@@ -1,10 +1,10 @@
-"""Domain job handlers: safety scoring and content analysis (US02-06).
+"""Domain job handlers: safety scoring, content analysis, uploads (US02-06, US05-02).
 
-Importing this module registers the ``safety_score`` and ``analysis`` job types so
-the generic worker can run them per item (one item = one ``asset_id``). Each handler
+Importing this module registers the ``safety_score``, ``analysis``, and
+``upload_batch`` job types so the generic worker can run them per item. Each handler
 delegates to its service, which owns the real work and the privacy gate. Handlers
 are idempotent: re-scoring or re-analyzing one asset is safe after an interrupted
-attempt.
+attempt, and an upload batch refuses to re-run an attempt whose outcome is unknown.
 
 Providers/models are the service defaults here (real NsfwModel / vision provider);
 tests exercise the services directly with injected fakes rather than the worker.
@@ -12,12 +12,15 @@ tests exercise the services directly with injected fakes rather than the worker.
 
 from __future__ import annotations
 
-from photo_pipeline.jobs.handlers import JobContext, register
+from photo_pipeline.jobs.handlers import Cancelled, JobContext, register
 
 SAFETY_SCORE = "safety_score"
 ANALYSIS = "analysis"
+UPLOAD_BATCH = "upload_batch"
 # Both mutate the library's metadata/derived state; one at a time (concept §one job).
 LIBRARY_WRITE_LOCK = "library_write"
+# The uploader lane: one album batch at a time (concept §16).
+UPLOAD_LOCK = "upload"
 
 
 def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
@@ -32,5 +35,22 @@ def _analysis_item(asset_id: str, ctx: JobContext) -> None:
     AnalysisService(ctx.session_factory).run([asset_id])
 
 
+def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:
+    """One item = one album batch. The upload itself is long and external, so the
+    handler hands the job's cancellation check to the service, which stops the
+    uploader and leaves a resumable batch."""
+    from photo_pipeline.config import Config
+    from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
+
+    config = ctx.config if ctx.config is not None else Config.from_env()
+    service = UploadBatchService(ctx.session_factory, config=config)
+    batch = service.run(batch_id, worker_id=ctx.worker_id, cancelled=ctx.cancelled)
+    if batch["state"] == BatchState.CANCELLED:
+        raise Cancelled(f"upload batch {batch_id} was cancelled")
+    if batch["state"] in (BatchState.FAILED, BatchState.UNKNOWN):
+        raise RuntimeError(f"upload batch {batch_id} is {batch['state']}: {batch['error_code']}")
+
+
 register(SAFETY_SCORE, _safety_score_item)
 register(ANALYSIS, _analysis_item)
+register(UPLOAD_BATCH, _upload_batch_item)

photo_pipeline/jobs/handlers.py

@@ -34,6 +34,9 @@ class JobContext:
     fencing_token: int
     service: "JobService"
     session_factory: object | None = None
+    # Handlers that talk to an external tool (uploads) need the typed configuration;
+    # the worker passes its own so a test stack is never read from the environment.
+    config: object | None = None
 
     def cancelled(self) -> bool:
         from photo_pipeline.services.jobs import JobState

photo_pipeline/jobs/worker.py

@@ -30,6 +30,7 @@ class Worker:
         *,
         job_types: Sequence[str] | None = None,
         lease_seconds: int = 60,
+        config: object | None = None,
     ) -> None:
         self._session_factory = session_factory
         self.service = JobService(session_factory)
@@ -37,6 +38,7 @@ class Worker:
         self.worker_id = worker_id
         self.job_types = list(job_types if job_types is not None else self.handlers.keys())
         self.lease_seconds = lease_seconds
+        self.config = config
 
     def run_once(self) -> str | None:
         """Recover stragglers, then claim and fully process one job. Returns its id."""
@@ -51,7 +53,9 @@ class Worker:
 
     def _process(self, job_id: str, job_type: str, token: int) -> None:
         handler = self.handlers[job_type]
-        ctx = JobContext(job_id, self.worker_id, token, self.service, self._session_factory)
+        ctx = JobContext(
+            job_id, self.worker_id, token, self.service, self._session_factory, self.config
+        )
         self._reset_interrupted_items(job_id, token)
 
         cancelled = False