phase6-Prefect

jobsource/flow.py

@@ -1,10 +1,126 @@
 """Prefect flow definition and interval schedule.
 
-Scaffold stub -- not implemented yet.
+Run with:
+    python -m jobsource.flow
+
+This starts a long-running process that:
+1. Boots a local Prefect server (if ``PREFECT_API_URL`` is not already set).
+2. Registers the ``job-source-batch`` deployment on an interval schedule
+   (default: daily, controlled by ``SCHEDULE_INTERVAL_SECONDS``).
+3. Polls for scheduled and manually-triggered runs, executing them in-process.
+
+The flow delegates entirely to :func:`~jobsource.pipeline.run_batch`, which is
+idempotent — the dedup step means retrying a failed flow run processes only the
+jobs that haven't been committed yet, so flow-level retries are safe.
+
+**Why a persistent server instead of Prefect's ephemeral server?**
+Prefect 3.7.4's ephemeral server is incompatible with FastAPI ≥ 0.137: it
+deletes sub-router route lists after inclusion (memory optimisation), but
+FastAPI 0.137 re-reads those lists lazily during request routing, causing all
+API paths to return 404.  The persistent server (``prefect server start``)
+does not apply that optimisation, so it works correctly.
 """
-# TODO (scheduling): implement per CLAUDE.md "Orchestration/scheduling: Prefect".
-# Wrap run_batch() in a @flow with:
-#   - Retries on the flow level.
-#   - An interval schedule (configurable; default daily).
-# Run with: python -m jobsource.flow
-# Cron fallback (no daemon): */0 6 * * * cd <repo> && ./.venv/bin/python -m jobsource.main --batch-size 50
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import time
+
+import httpx
+from prefect import flow
+
+from .config import get_settings
+from .pipeline import BatchSummary, run_batch
+
+# Read scheduling config once at module load (get_settings() is a cached singleton).
+# Prefect @flow decorator arguments must be concrete values, not lazy callables.
+_settings = get_settings()
+
+_DEFAULT_SERVER_URL = "http://127.0.0.1:4200"
+_SERVER_READY_TIMEOUT = 30  # seconds
+
+
+@flow(
+    name="job-source-batch",
+    retries=_settings.flow_retries,
+    retry_delay_seconds=_settings.flow_retry_delay_seconds,
+    log_prints=True,
+)
+def job_source_batch(
+    batch_size: int | None = None,
+    search_terms: list[str] | None = None,
+    location: str | None = None,
+    hours_old: int | None = None,
+) -> BatchSummary:
+    """Fetch, cascade, persist, and export one batch of LinkedIn jobs.
+
+    All parameters override the corresponding config/env values when provided;
+    ``None`` means "use config default".  Returns the :class:`BatchSummary`
+    produced by :func:`~jobsource.pipeline.run_batch`.
+
+    Flow-level retries are safe because ``run_batch`` deduplicates on
+    ``job_id`` — a retry only processes jobs not yet committed.
+    """
+    return run_batch(
+        batch_size=batch_size,
+        search_terms=search_terms,
+        location=location,
+        hours_old=hours_old,
+    )
+
+
+def _start_prefect_server() -> subprocess.Popen[bytes]:
+    """Start ``prefect server start`` as a background subprocess."""
+    return subprocess.Popen(
+        [sys.executable, "-m", "prefect", "server", "start",
+         "--host", "127.0.0.1", "--port", "4200"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+    )
+
+
+def _wait_for_server(url: str, timeout: int) -> bool:
+    """Return True once the Prefect server health endpoint responds OK."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            r = httpx.get(f"{url}/api/health", timeout=2.0)
+            if r.status_code == 200:
+                return True
+        except Exception:
+            pass
+        time.sleep(1)
+    return False
+
+
+if __name__ == "__main__":
+    _server_proc: subprocess.Popen[bytes] | None = None
+
+    if not os.environ.get("PREFECT_API_URL"):
+        # No external server configured — boot the persistent local server.
+        # The Prefect 3.7.4 ephemeral server is incompatible with FastAPI ≥ 0.137
+        # (see module docstring), so we start a persistent server instead.
+        print(f"Starting local Prefect server at {_DEFAULT_SERVER_URL} …")
+        _server_proc = _start_prefect_server()
+        if not _wait_for_server(_DEFAULT_SERVER_URL, _SERVER_READY_TIMEOUT):
+            _server_proc.terminate()
+            raise RuntimeError(
+                f"Prefect server did not become ready within {_SERVER_READY_TIMEOUT}s. "
+                "Start it manually with `prefect server start` and set "
+                "PREFECT_API_URL=http://127.0.0.1:4200/api before re-running."
+            )
+        os.environ["PREFECT_API_URL"] = f"{_DEFAULT_SERVER_URL}/api"
+        print(f"Prefect server ready.  API → {os.environ['PREFECT_API_URL']}")
+
+    try:
+        # Serve on an interval schedule; blocks until Ctrl-C.
+        job_source_batch.serve(
+            name="job-source-batch",
+            interval=_settings.schedule_interval_seconds,
+        )
+    finally:
+        if _server_proc is not None:
+            print("Stopping local Prefect server …")
+            _server_proc.terminate()
+            _server_proc.wait(timeout=10)