From de20e6c8cebb878986e8c73e875dcabec31f450f Mon Sep 17 00:00:00 2001 From: domverse Date: Thu, 16 Jul 2026 20:15:57 +0200 Subject: [PATCH] US02-04: Expose Job APIs and Activity Events (#57) --- photo_pipeline/api/app.py | 3 +- photo_pipeline/api/routes/jobs.py | 116 +++++++++++++++++++++++++++++ photo_pipeline/schemas/__init__.py | 3 +- photo_pipeline/schemas/jobs.py | 12 +++ photo_pipeline/services/jobs.py | 16 +++- tests/integration/test_jobs_api.py | 108 +++++++++++++++++++++++++++ tests/integration/test_jobs_sse.py | 69 +++++++++++++++++ tests/story_traceability.json | 4 + 8 files changed, 328 insertions(+), 3 deletions(-) create mode 100644 photo_pipeline/api/routes/jobs.py create mode 100644 photo_pipeline/schemas/jobs.py create mode 100644 tests/integration/test_jobs_api.py create mode 100644 tests/integration/test_jobs_sse.py diff --git a/photo_pipeline/api/app.py b/photo_pipeline/api/app.py index d90ffa1..9be1229 100644 --- a/photo_pipeline/api/app.py +++ b/photo_pipeline/api/app.py @@ -14,7 +14,7 @@ from pathlib import Path from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from photo_pipeline.api.routes import duplicates, health, inventory, thumbnails +from photo_pipeline.api.routes import duplicates, health, inventory, jobs, thumbnails from photo_pipeline.config import Config from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations from photo_pipeline.logging import configure_logging @@ -44,6 +44,7 @@ def create_app(config: Config | None = None) -> FastAPI: app.include_router(health.router, prefix="/api/v1") app.include_router(inventory.router, prefix="/api/v1") app.include_router(duplicates.router, prefix="/api/v1") + app.include_router(jobs.router, prefix="/api/v1") app.include_router(thumbnails.router, prefix="/api/v1") # Static single-page app (hash-routed). Mounted last so /api/v1 wins. if FRONTEND_DIR.is_dir(): diff --git a/photo_pipeline/api/routes/jobs.py b/photo_pipeline/api/routes/jobs.py new file mode 100644 index 0000000..1ba42b4 --- /dev/null +++ b/photo_pipeline/api/routes/jobs.py @@ -0,0 +1,116 @@ +"""Job command, status, blocker, and activity-event API. + +Operational endpoints return JSON with the shared error envelope; the event stream +returns SSE with resumable ``id:`` cursors (SQLite rowid). A client that drops the +stream reconnects with ``Last-Event-ID`` and receives exactly the durable events it +missed — no duplicates, none lost. The same events are available as plain JSON via +``?after=`` for a polling fallback. +""" + +from __future__ import annotations + +import json + +import anyio +from fastapi import APIRouter, Query, Request +from fastapi.responses import JSONResponse, StreamingResponse + +from photo_pipeline.schemas import JobStartRequest +from photo_pipeline.services.jobs import ( + ACTIVE_STATES, + InvalidTransition, + JobBlocked, + JobError, + JobService, +) + +router = APIRouter(tags=["jobs"]) + + +def _service(request: Request) -> JobService: + return JobService(request.app.state.session_factory) + + +def _error(status: int, code: str, message: str) -> JSONResponse: + return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}}) + + +@router.post("/jobs") +def start_job(body: JobStartRequest, request: Request): + service = _service(request) + try: + return service.enqueue( + body.job_type, + lock=body.lock, + idempotency_key=body.idempotency_key, + items=body.items, + ) + except JobBlocked as error: + return _error(409, error.code, str(error)) + + +@router.get("/jobs/{job_id}") +def get_job(job_id: str, request: Request): + service = _service(request) + snapshot = service.get(job_id) + if snapshot is None: + return _error(404, "not_found", f"unknown job {job_id}") + return {**snapshot, "progress": service.progress(job_id)} + + +@router.post("/jobs/{job_id}/cancel") +def cancel_job(job_id: str, request: Request): + service = _service(request) + if service.get(job_id) is None: + return _error(404, "not_found", f"unknown job {job_id}") + try: + return service.cancel(job_id) + except InvalidTransition as error: + return _error(409, "invalid_transition", str(error)) + except JobError as error: + return _error(409, "job_error", str(error)) + + +@router.get("/jobs/{job_id}/blockers") +def job_blockers(job_id: str, request: Request, lock: str = Query(...)): + return {"lock": lock, "blockers": _service(request).blockers(lock)} + + +@router.get("/jobs/{job_id}/events") +def list_events(job_id: str, request: Request, after: int = 0): + service = _service(request) + snapshot = service.get(job_id) + if snapshot is None: + return _error(404, "not_found", f"unknown job {job_id}") + return {"state": snapshot["state"], "events": service.events_after(job_id, after)} + + +@router.get("/jobs/{job_id}/events/stream") +async def stream_events(job_id: str, request: Request, after: int = 0): + service = _service(request) + if service.get(job_id) is None: + return _error(404, "not_found", f"unknown job {job_id}") + + header = request.headers.get("last-event-id") + cursor = int(header) if header and header.isdigit() else after + + async def generate(): + nonlocal cursor + # Bounded so a stream can't run forever; a terminal job closes immediately. + # ponytail: sync DB reads inside the async loop are fine for a local + # single-user app; move to a threadpool/pubsub if it ever multiplexes. + for _ in range(600): + for event in service.events_after(job_id, cursor): + cursor = event["seq"] + yield ( + f"id: {event['seq']}\n" + f"event: {event['type']}\n" + f"data: {json.dumps({'message': event['message']})}\n\n" + ) + snapshot = service.get(job_id) + if snapshot is None or snapshot["state"] not in ACTIVE_STATES: + yield "event: done\ndata: {}\n\n" + return + await anyio.sleep(0.1) + + return StreamingResponse(generate(), media_type="text/event-stream") diff --git a/photo_pipeline/schemas/__init__.py b/photo_pipeline/schemas/__init__.py index f116276..0c0822d 100644 --- a/photo_pipeline/schemas/__init__.py +++ b/photo_pipeline/schemas/__init__.py @@ -1,5 +1,6 @@ """Pydantic API request/response contracts.""" from photo_pipeline.schemas.duplicates import DecisionRequest +from photo_pipeline.schemas.jobs import JobStartRequest -__all__ = ["DecisionRequest"] +__all__ = ["DecisionRequest", "JobStartRequest"] diff --git a/photo_pipeline/schemas/jobs.py b/photo_pipeline/schemas/jobs.py new file mode 100644 index 0000000..38d6319 --- /dev/null +++ b/photo_pipeline/schemas/jobs.py @@ -0,0 +1,12 @@ +"""Request contracts for job commands.""" + +from __future__ import annotations + +from pydantic import BaseModel + + +class JobStartRequest(BaseModel): + job_type: str + lock: str | None = None + idempotency_key: str | None = None + items: list[str] = [] diff --git a/photo_pipeline/services/jobs.py b/photo_pipeline/services/jobs.py index fbd552d..71d7d50 100644 --- a/photo_pipeline/services/jobs.py +++ b/photo_pipeline/services/jobs.py @@ -23,7 +23,7 @@ from collections import Counter from datetime import datetime, timedelta, timezone from collections.abc import Iterable, Sequence -from sqlalchemy import select, update +from sqlalchemy import select, text, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import sessionmaker @@ -357,6 +357,20 @@ class JobService: ).scalars() return [{"type": e.event_type, "message": e.message} for e in rows] + def events_after(self, job_id: str, after: int = 0) -> list[dict]: + """Durable events with ``seq`` > ``after``, ordered — the resumable cursor for + SSE and polling. SQLite's monotonic rowid is the sequence (job_events is a + rowid table), so no extra column is needed.""" + with self._session_factory() as session: + rows = session.execute( + text( + "SELECT rowid AS seq, event_type, message FROM job_events " + "WHERE job_id = :jid AND rowid > :after ORDER BY rowid" + ), + {"jid": job_id, "after": after}, + ).mappings().all() + return [{"seq": r["seq"], "type": r["event_type"], "message": r["message"]} for r in rows] + # ── helpers ────────────────────────────────────────────────────────────────── @staticmethod def _event(session, job_id: str, event_type: str, message: str | None = None) -> None: diff --git a/tests/integration/test_jobs_api.py b/tests/integration/test_jobs_api.py new file mode 100644 index 0000000..fe3153b --- /dev/null +++ b/tests/integration/test_jobs_api.py @@ -0,0 +1,108 @@ +"""Job API contract: typed commands, status codes, error envelope, idempotency, +cancellation, blockers, and OpenAPI schema.""" + +import pytest +from fastapi.testclient import TestClient + +from photo_pipeline.api.app import create_app +from photo_pipeline.config import Config +from photo_pipeline.db import run_migrations +from photo_pipeline.services.jobs import ItemState, JobService, JobState + + +@pytest.fixture +def client(tmp_path): + (tmp_path / "data").mkdir() + config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")}) + run_migrations(config.database_url) + app = create_app(config) + with TestClient(app) as test_client: + test_client.app = app + yield test_client + + +def _drive_to_success(app, job_id, items): + service = JobService(app.state.session_factory) + service.claim(["scan"], "w1") + for item in items: + service.set_item(job_id, item, ItemState.RUNNING) + service.set_item(job_id, item, ItemState.SUCCEEDED) + service.transition(job_id, JobState.SUCCEEDED, worker_id="w1") + + +def test_start_status_and_progress(client): + start = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]}) + assert start.status_code == 200 + job = start.json() + assert job["state"] == "queued" + + status = client.get(f"/api/v1/jobs/{job['id']}") + assert status.status_code == 200 + body = status.json() + assert body["progress"] == {"total": 2, "done": 0, "by_state": {"queued": 2}} + + +def test_idempotent_start_returns_same_job(client): + first = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json() + second = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json() + assert first["id"] == second["id"] + + +def test_lock_conflict_returns_409(client): + client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"}) + conflict = client.post("/api/v1/jobs", json={"job_type": "rename", "lock": "library_write"}) + assert conflict.status_code == 409 + assert conflict.json()["error"]["code"] == "lock_held" + + +def test_cancel_queued_then_terminal(client): + job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json() + cancelled = client.post(f"/api/v1/jobs/{job['id']}/cancel") + assert cancelled.status_code == 200 + assert cancelled.json()["state"] == "cancelled" + # Cancelling a terminal job conflicts. + again = client.post(f"/api/v1/jobs/{job['id']}/cancel") + assert again.status_code == 409 + + +def test_blockers_endpoint(client): + client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"}) + job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json() + blockers = client.get(f"/api/v1/jobs/{job['id']}/blockers?lock=library_write").json() + assert len(blockers["blockers"]) == 1 + + +def test_unknown_job_uses_error_envelope(client): + for resp in ( + client.get("/api/v1/jobs/ghost"), + client.post("/api/v1/jobs/ghost/cancel"), + client.get("/api/v1/jobs/ghost/events"), + ): + assert resp.status_code == 404 + assert resp.json()["error"]["code"] == "not_found" + + +def test_polling_events_with_cursor(client): + job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a"]}).json() + _drive_to_success(client.app, job["id"], ["a"]) + all_events = client.get(f"/api/v1/jobs/{job['id']}/events").json()["events"] + seqs = [e["seq"] for e in all_events] + assert seqs == sorted(seqs) and len(seqs) >= 3 + after_first = client.get(f"/api/v1/jobs/{job['id']}/events?after={seqs[0]}").json()["events"] + assert [e["seq"] for e in after_first] == seqs[1:] + + +def test_openapi_exposes_job_contracts(client): + paths = client.get("/openapi.json").json()["paths"] + assert "/api/v1/jobs" in paths + assert "post" in paths["/api/v1/jobs"] + assert "/api/v1/jobs/{job_id}" in paths + assert "/api/v1/jobs/{job_id}/cancel" in paths + assert "/api/v1/jobs/{job_id}/events/stream" in paths + + +def test_status_content_type_is_json(client): + job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json() + assert client.get(f"/api/v1/jobs/{job['id']}").headers["content-type"].startswith( + "application/json" + ) diff --git a/tests/integration/test_jobs_sse.py b/tests/integration/test_jobs_sse.py new file mode 100644 index 0000000..cfa877e --- /dev/null +++ b/tests/integration/test_jobs_sse.py @@ -0,0 +1,69 @@ +"""SSE activity stream: resumable event ids, reconnect without duplicate/lost +events, and parity with the polling fallback.""" + +import pytest +from fastapi.testclient import TestClient + +from photo_pipeline.api.app import create_app +from photo_pipeline.config import Config +from photo_pipeline.db import run_migrations +from photo_pipeline.services.jobs import ItemState, JobService, JobState + + +@pytest.fixture +def client(tmp_path): + (tmp_path / "data").mkdir() + config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")}) + run_migrations(config.database_url) + app = create_app(config) + with TestClient(app) as test_client: + test_client.app = app + yield test_client + + +def _finished_job(client): + job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]}).json() + service = JobService(client.app.state.session_factory) + service.claim(["scan"], "w1") + for item in ("a", "b"): + service.set_item(job["id"], item, ItemState.RUNNING) + service.set_item(job["id"], item, ItemState.SUCCEEDED) + service.transition(job["id"], JobState.SUCCEEDED, worker_id="w1") + return job["id"] + + +def _read_ids(client, job_id, headers=None): + ids, saw_done = [], False + with client.stream( + "GET", f"/api/v1/jobs/{job_id}/events/stream", headers=headers or {} + ) as stream: + assert stream.headers["content-type"].startswith("text/event-stream") + for line in stream.iter_lines(): + if line.startswith("id:"): + ids.append(int(line.split(":", 1)[1])) + elif line.startswith("event: done"): + saw_done = True + return ids, saw_done + + +def test_sse_streams_all_events_then_closes(client): + job_id = _finished_job(client) + ids, saw_done = _read_ids(client, job_id) + assert ids == sorted(ids) and len(ids) >= 3 + assert saw_done # terminal job closes the stream + + +def test_sse_reconnect_has_no_duplicate_or_lost_events(client): + job_id = _finished_job(client) + full, _ = _read_ids(client, job_id) + resume_from = full[1] + resumed, saw_done = _read_ids(client, job_id, headers={"Last-Event-ID": str(resume_from)}) + assert resumed == [i for i in full if i > resume_from] # no dup, none lost + assert saw_done + + +def test_sse_matches_polling(client): + job_id = _finished_job(client) + sse_ids, _ = _read_ids(client, job_id) + polled = client.get(f"/api/v1/jobs/{job_id}/events").json()["events"] + assert sse_ids == [e["seq"] for e in polled] diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 68b879f..d405bf9 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -53,6 +53,10 @@ "tests/unit/test_lock_order.py", "tests/integration/test_worker.py", "tests/e2e/test_worker_kill.py" + ], + "US02-04": [ + "tests/integration/test_jobs_api.py", + "tests/integration/test_jobs_sse.py" ] } }