feat: add prefect_enabled config option to run ingestions without Prefect

[minhajuddin2510]

Why

Submitting a resource fails outright on deployments where CKAN cannot write $PREFECT_HOME:

ERROR [ckanext.datapusher_plus.logic.action] Error submitting job to DataPusher: [Errno 13] Permission denied: '/root/.prefect/profiles.toml'

This is not a Prefect outage. prefect_client.submit_flow_run() does from prefect.deployments import run_deployment, and that import runs Prefect's settings bootstrap, which creates $PREFECT_HOME/profiles.toml (default $HOME/.prefect). A CKAN process running with HOME=/root but no write access there cannot even import the library, so no server configuration fixes it.

What changed

New setting, default true:

ckanext.datapusher_plus.prefect_enabled = false

With it off, datapusher_submit enqueues the job on CKAN's own background queue and the new jobs/local_runner.py executes the same nine ingestion stages in-process — no Prefect server, worker, work pool, or import prefect anywhere on the path. Operators run ckan jobs worker instead.

	file
In-process runner: enqueue_job (RQ submit), run_job (the nine stages), get_running_resource_ids (RQ-queue scan for the duplicate-submission check)	jobs/local_runner.py (new)
Prefect-free core shared by both runners: status callback, input validation, RuntimeContext construction, stage invoker + StageAbort, datastore rollback, resolve_int	jobs/pipeline_core.py (new)
_orchestrator() / _submit_job() routing	logic/action.py
Events become no-ops; prefect-deploy refuses to run in this mode; migrate-from-rq skips the reachability check; startup log line	jobs/events.py, cli.py, plugin.py
Config declaration, README section, CONFIG.md, CHANGELOG, CLAUDE.md	docs

Unchanged in local mode: Jobs/Logs tables, the job-status page, datapusher_hook callbacks (running → complete/error), CLI submit/resubmit, DRUF, the complete-with-skip contract, and the flow_timeout deadline (also passed to RQ as the job timeout — RQ's own default is 180s, which would kill any real ingestion mid-COPY).

Reviewer notes — three deliberate behavior decisions

1. PII review aborts instead of suspending. pii_review_threshold > 0 asks a human to approve a PII-flagged run. Nothing can hold a job open without Prefect, so the only safe reading of that intent is "do not load it" — the job errors before any datastore write. Documented in the README table and in the error message itself.
2. A database-stage failure does not drop the datastore table. A failure after the load drops the half-built table and restores the stashed Data Dictionary, matching database_task.on_rollback. But the database stage can raise before touching anything ("Could not connect to the Datastore"), and dropping there would destroy intact data the run never wrote. Both paths are tested.
3. The RQ job id is recorded as rq_job_id, not flow_run_id. Otherwise a deployment with a stale prefect_ui_base would render a Prefect UI deep-link pointing at an RQ id. datapusher_status already returns job_url: None when flow_run_id is absent, so its response shape is unchanged.

Refactor note

prefect_flow.py shed the Prefect-free half into pipeline_core.py (~200 lines moved, not rewritten) so there is one implementation per behavior. prefect_flow binds the moved names to its historical private spellings (_StageAbort, _build_runtime_context, _stage_run, …), so custom flows composed from its @task primitives are unaffected. Tests that patched prefect_flow.dsu / .QSVCommand / .Path now patch pipeline_core — same module objects, owner-correct target.

Testing

Unit suite (pytest tests/ --ignore=tests/integration): 307 passed, 1 failed vs 279 passed, 1 failed on the base commit. The single failure is test_quoted_csv_inference_matrix, pre-existing and unrelated (needs a real qsv 20.x binary locally).

28 new tests in tests/test_local_runner.py: the config switch, stage order, complete-with-skip, datastore-dump short-circuit, both rollback paths, both PII-gate paths, the RQ helpers, and datapusher_submit routing (including which key the task_status row gets). One of them shells out to a subprocess to assert that importing the disabled path pulls in zero prefect modules — the invariant this whole PR rests on.

Integration tests were not run (no docker-compose stack in the dev environment); a reviewer with the stack up should sanity-check that the default Prefect path is untouched end-to-end.

Alternative for operators who want to keep Prefect

Point PREFECT_HOME at a directory the CKAN user owns (e.g. /var/lib/ckan/prefect) in both the CKAN and worker environments. Documented alongside the new mode in the README troubleshooting table.

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR adds a ckanext.datapusher_plus.prefect_enabled configuration switch (default true) to support running DP+ ingestions without importing Prefect at all, falling back to CKAN’s RQ worker for orchestration when disabled. It introduces a Prefect-free shared pipeline core so both the Prefect flow and the local runner reuse the same callback/validation/context/rollback logic.

Changes:

• Add prefect_enabled config + declarative config entry, and route datapusher_submit to either Prefect (prefect_client) or RQ (jobs/local_runner.py).
• Introduce jobs/pipeline_core.py (Prefect-free shared pipeline helpers) and jobs/local_runner.py (in-process 9-stage runner + RQ helpers).
• Update docs and tests to cover the no-Prefect mode and refactored patch targets.

Reviewed changes

Copilot reviewed 17 out of 17 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/test_prefect_flow.py	Updates patch targets to pipeline_core after refactor.
tests/test_local_runner.py	New unit coverage for prefect_enabled=false path, including “no Prefect import” invariant.
tests/test_dictionary_stash.py	Updates monkeypatch targets to pipeline_core datastore helpers.
README.md	Documents prefect_enabled option, no-Prefect mode behavior, and troubleshooting.
CONFIG.md	Adds configuration guidance for disabling Prefect and operational requirements.
CLAUDE.md	Updates architecture notes to include pipeline_core + local_runner.
ckanext/datapusher_plus/plugin.py	Logs an operator-facing startup INFO line when Prefect is disabled.
ckanext/datapusher_plus/logic/action.py	Adds _orchestrator() / _submit_job() routing and records rq_job_id vs flow_run_id.
ckanext/datapusher_plus/jobs/prefect_flow.py	Moves Prefect-free functionality into pipeline_core and aliases historical private names.
ckanext/datapusher_plus/jobs/pipeline_core.py	New Prefect-free shared implementation (callback, validate, context build, stage invocation, rollback, int resolution).
ckanext/datapusher_plus/jobs/local_runner.py	New RQ-backed in-process ingestion runner + queue scanning helpers.
ckanext/datapusher_plus/jobs/events.py	Makes event emission a no-op when Prefect is disabled (to avoid importing Prefect).
ckanext/datapusher_plus/jobs/init.py	Exposes run_job and ensures callback helper is sourced from Prefect-free core.
ckanext/datapusher_plus/config.py	Adds prefect_enabled() runtime config accessor.
ckanext/datapusher_plus/config_declaration.yaml	Declares ckanext.datapusher_plus.prefect_enabled with default and description.
ckanext/datapusher_plus/cli.py	Prevents prefect-deploy when Prefect is disabled; adjusts migrate messaging/checks.
CHANGELOG.md	Documents the new configuration option and refactor.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.