From 11f10affbd300517c5f3aa80a7f04064ccb9401a Mon Sep 17 00:00:00 2001 From: ldy Date: Thu, 18 Jun 2026 00:32:26 -0400 Subject: [PATCH] phase6-Prefect --- .env.example | 6 ++ README.md | 45 ++++++++++++--- jobsource/config.py | 5 ++ jobsource/flow.py | 130 +++++++++++++++++++++++++++++++++++++++++--- jobsource/main.py | 24 ++++++-- 5 files changed, 192 insertions(+), 18 deletions(-) diff --git a/.env.example b/.env.example index 7016b9f..32b3bb2 100644 --- a/.env.example +++ b/.env.example @@ -40,3 +40,9 @@ OUTPUT_CSV=output/results.csv # == Browser agent == ENABLE_BROWSER_AGENT=true BROWSER_HEADLESS=true + +# == Scheduling (Prefect flow) == +# Interval between scheduled runs in seconds (86400 = daily) +SCHEDULE_INTERVAL_SECONDS=86400 +FLOW_RETRIES=1 +FLOW_RETRY_DELAY_SECONDS=60 diff --git a/README.md b/README.md index 965be0f..4e3d867 100644 --- a/README.md +++ b/README.md @@ -20,18 +20,49 @@ cp .env.example .env # fill keys as available ## Run ```bash -# one batch +# one batch (no server required) python -m jobsource.main --batch-size 20 --search "software engineer" --location "United States" - -# scheduled run (Prefect) -python -m jobsource.flow - -# cron fallback (no daemon): -# */0 6 * * * cd && ./.venv/bin/python -m jobsource.main --batch-size 50 ``` `--search` is repeatable. Run `python -m jobsource.main --help` for all options. +### Scheduled run (Prefect) + +```bash +# Boots a local Prefect server automatically, then serves the flow on an +# interval schedule (default: daily, set SCHEDULE_INTERVAL_SECONDS to change). +python -m jobsource.flow +``` + +If you already have a Prefect server running elsewhere, point the flow at it: + +```bash +export PREFECT_API_URL=http://127.0.0.1:4200/api +python -m jobsource.flow +``` + +To trigger a run manually while the flow is being served: + +```bash +prefect deployment run 'job-source-batch/job-source-batch' +``` + +> **Note — Prefect 3.7.4 + FastAPI ≥ 0.137 compatibility**: Prefect's built-in +> ephemeral server is incompatible with FastAPI 0.137+ (route lists are cleared +> after inclusion but re-read lazily during routing). `python -m jobsource.flow` +> works around this by starting a persistent `prefect server start` process +> instead of the ephemeral server. + +### No-daemon cron fallback + +If you prefer a plain cron job with no running process, add this line to your +crontab (`crontab -e`), adjusting the path to your repo: + +``` +# daily at 06:00 — no Prefect daemon required +0 6 * * * cd /path/to/repo && ./.venv/bin/python -m jobsource.main --batch-size 50 +``` + ## Tests ```bash diff --git a/jobsource/config.py b/jobsource/config.py index d328a8c..f42612d 100644 --- a/jobsource/config.py +++ b/jobsource/config.py @@ -57,6 +57,11 @@ class Settings(BaseSettings): enable_browser_agent: bool = True browser_headless: bool = True + # -- Scheduling (Prefect flow) ----------------------------------------- + schedule_interval_seconds: int = 86400 # default: daily + flow_retries: int = 1 + flow_retry_delay_seconds: int = 60 + @lru_cache def get_settings() -> Settings: diff --git a/jobsource/flow.py b/jobsource/flow.py index 43d38ec..93793a7 100644 --- a/jobsource/flow.py +++ b/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 && ./.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) diff --git a/jobsource/main.py b/jobsource/main.py index 0141c60..1144729 100644 --- a/jobsource/main.py +++ b/jobsource/main.py @@ -1,12 +1,19 @@ -"""CLI entry point: `python -m jobsource.main`. +"""CLI entry point: ``python -m jobsource.main``. -Scaffold stub. Argument parsing is wired so `--help` works; the actual batch -run lands in a later step (see jobsource/pipeline.py). Imports only stdlib so -`--help` works before the heavier dependencies are installed. +Parses arguments and delegates to :func:`~jobsource.pipeline.run_batch`. +All heavy imports are deferred past the argument parse so ``--help`` exits +immediately with no dependency overhead. + +Usage examples:: + + python -m jobsource.main --help + python -m jobsource.main --batch-size 20 --search "software engineer" --location "United States" + python -m jobsource.main --search "data engineer" --search "ML engineer" --hours-old 48 """ from __future__ import annotations import argparse +import logging import sys @@ -46,6 +53,15 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) + + # Configure root logger so pipeline/cascade INFO and WARNING messages are + # visible on the terminal. Deferred past parse_args so --help stays instant. + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + from jobsource.pipeline import run_batch # deferred: keeps --help fast before heavy deps run_batch( batch_size=args.batch_size,