feat: add ckanext.datapusher_plus.prefect_enabled to run jobs without Prefect

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'

That is not a Prefect outage — `submit_flow_run`'s `from
prefect.deployments import run_deployment` 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 amount of server
configuration helps.

Add `ckanext.datapusher_plus.prefect_enabled` (default true). Setting it
to false turns Prefect off entirely: `datapusher_submit` enqueues the job
on CKAN's own background queue and the new `jobs/local_runner.py` runs the
same nine ingestion stages in-process, with nothing on the path importing
`prefect`. Operators run `ckan jobs worker` instead of a Prefect server,
worker, and work pool.

The local runner keeps the Jobs/Logs tables, the datapusher_hook
callbacks, the complete-with-skip contract, the flow_timeout deadline, and
the post-database-failure datastore cleanup. It does not provide per-stage
retries, result caching, the run graph/artifacts/events, or
human-in-the-loop PII review — with Prefect off, a job crossing
pii_review_threshold aborts before any datastore write rather than
suspending for an approval that can never arrive. A database-stage failure
deliberately leaves the datastore alone (that stage can fail before
touching anything, and dropping there would destroy data the run never
wrote). The task_status row records the RQ job id as `rq_job_id` rather
than `flow_run_id`, so nothing builds a Prefect-UI deep link out of it.

The Prefect-free half of the pipeline moves out of prefect_flow.py into
the new jobs/pipeline_core.py — the CKAN status callback, input
validation, RuntimeContext construction, the stage invoker and its
StageAbort signal, and the datastore rollback body — shared verbatim by
both runners. prefect_flow keeps its historical private names as aliases,
so custom flows composed from its @task primitives are unaffected; tests
that patched prefect_flow.dsu / .QSVCommand / .Path now patch
pipeline_core.

Unit suite: 307 passed, 1 failed (test_quoted_csv_inference_matrix, a
pre-existing qsv-binary-dependent failure present on the base commit too).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: minhajuddin2510

CHANGELOG.md

@@ -6,7 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+* New `ckanext.datapusher_plus.prefect_enabled` setting (default `true`). Setting it to `false` turns Prefect off entirely: `datapusher_submit` enqueues the job on CKAN's own background-job queue and the new `ckanext/datapusher_plus/jobs/local_runner.py` executes the same nine ingestion stages in-process, so no Prefect server, worker, work pool — or `import prefect` — is involved. Operators run `ckan jobs worker` instead. This is the fix for deployments where importing Prefect *itself* fails, e.g. a CKAN process running with `HOME=/root` but no write access there, which surfaces as `Error submitting job to DataPusher: [Errno 13] Permission denied: '/root/.prefect/profiles.toml'` (the alternative fix, if you want to keep Prefect, is to point `PREFECT_HOME` at a writable directory). The local runner keeps the `Jobs`/`Logs` tables, the `datapusher_hook` callbacks, the complete-with-skip contract, the `flow_timeout` deadline, and the post-database-failure datastore cleanup; it does *not* provide per-stage retries, result caching, the Prefect run graph/artifacts/events, or human-in-the-loop PII review — with Prefect off, a job crossing `pii_review_threshold` aborts before any datastore write rather than suspending for approval. The task_status row records the RQ job id as `rq_job_id` (not `flow_run_id`), so nothing builds a Prefect-UI deep link out of it. See [Running without Prefect](README.md#running-without-prefect).
+
 ### Changed
+* The Prefect-free half of the pipeline moved out of `jobs/prefect_flow.py` into the new `jobs/pipeline_core.py` (the CKAN status callback, input validation, `RuntimeContext` construction, the stage invoker and its `StageAbort` signal, and the datastore rollback body), shared verbatim by the Prefect flow and the local runner. `prefect_flow` keeps its historical private names as aliases, so custom flows composed from its `@task` primitives are unaffected. Tests that patched `prefect_flow.dsu` / `.QSVCommand` / `.Path` now patch `pipeline_core`.
 * Auto-indexing now uses a closed cardinality range `[auto_index_min_threshold, auto_index_threshold]` instead of a single upper bound (issue #142). Defaults bumped: `auto_index_threshold` `3` → `10` and the new `auto_index_min_threshold` defaults to `3`. The new lower bound skips the useless-single-value-column case @EricSoroos flagged in #142 (a 1-value text column previously produced a 10–40MB B-tree the Postgres planner would never choose). The upper bump from 3 → 10 widens the DataTables-SearchBuilder filtering sweet spot to cover typical enum-shaped columns (status = 3–10 values, common enums under ~10). **Operator-facing impact at default settings:** columns with cardinality 1–2 lose their auto-index (intentional — the indexes were dead weight); columns with cardinality 4–10 gain an auto-index. No migration is required; the change re-applies on the next resubmit. To restore pre-#142 behavior (no lower floor) set `ckanext.datapusher_plus.auto_index_min_threshold = 0`. The setting `auto_index_threshold = -1` ("index every column") still works but now hits the `min_threshold = 3` floor unless paired with `auto_index_min_threshold = 0`.
 * **BREAKING** Bumped `MINIMUM_QSV_VERSION` from `4.0.0` to `20.0.0`. Operators must upgrade their `qsv` binary at the path configured by `ckanext.datapusher_plus.qsv_bin` before deploying this version — DP+ will refuse to start otherwise. See [qsv 20.0.0 release notes](https://github.com/dathere/qsv/releases/tag/20.0.0) and the migration notes below. ([README install snippet](README.md#option-2-install-prebuilt-qsv-binaries) updated accordingly.)
 * **BREAKING** Bumped `MINIMUM_QSV_VERSION` from `20.0.0` to `20.1.0` (and bumped the qsv version installed by `Dockerfile.worker`, `.github/workflows/ci.yml`, and `.github/workflows/main.yml` to match). Operators must upgrade their `qsv` binary at the path configured by `ckanext.datapusher_plus.qsv_bin` before deploying this version — DP+ will refuse to start otherwise with `JobError: At least qsv version 20.1.0 required. Found 20.0.0.`. qsv 20.1.0 itself introduces no breaking changes against 20.0.0 (per the [20.1.0 release notes](https://github.com/dathere/qsv/releases/tag/20.1.0): "pipelines built on 20.0.0 will upgrade in place"), so the upgrade is binary-swap-only — no data re-ingestion is required. The reason this is still flagged as **BREAKING** is the minimum-version gate, not the qsv behavior. User-visible improvement justifying the floor bump: qsv-dateparser 0.14 → 0.15 in qsv 20.1.0 adds recognition for ISO 8601 `T`-separated datetimes without a timezone suffix (e.g. `2024-10-11T14:30:00`) — qsv 20.0.0 misclassified these as `String` during `qsv stats --infer-dates`, which in DP+ surfaced as a date-typed column being demoted to text on certain CSV shapes (one of the gaps documented in the issue #173 regression test, now closed). The regression-test matrix in `tests/test_issue_173_date_format_inference.py` is updated to the new baseline.

CLAUDE.md

@@ -66,7 +66,17 @@ prefect_flow.py    → Orchestration. Per-stage @task functions (each delegates
 __init__.py        → Public surface via PEP 562 lazy __getattr__ (defers the Prefect
                      import so CKAN admin commands don't spin up a Prefect server).
                      Exposes `datapusher_plus_flow`, `push_to_datastore` (v2 shim),
-                     `datapusher_plus_to_datastore` (alias), `callback_datapusher_hook`.
+                     `datapusher_plus_to_datastore` (alias), `callback_datapusher_hook`,
+                     `run_job`.
+pipeline_core.py   → Prefect-free core shared by both runners: callback_datapusher_hook,
+                     validate_input, build_runtime_context, run_stage + StageAbort,
+                     resource_is_datastore_dump, rollback_datastore_writes, resolve_int.
+                     **Nothing here may import prefect.**
+local_runner.py    → The `prefect_enabled = false` path. `enqueue_job` puts the job on
+                     CKAN's RQ queue (`ckan jobs worker` runs it); `run_job` executes the
+                     nine stages sequentially over one live RuntimeContext, owning the
+                     same Jobs-row state machine and callbacks as the flow. No retries,
+                     caching, artifacts, events, or PII suspend-for-review.
 context.py         → ProcessingContext — per-run mutable state shared across stages.
 runtime_context.py → JobInput (frozen, JSON-serializable flow input), the per-stage
                      `*Result` dataclasses (DownloadResult, AnalyzeResult, …), the
@@ -96,6 +106,8 @@ stages/
 
 Operators can register a custom flow via `ckanext.datapusher_plus.prefect_flow`; the per-stage `@task` functions in `prefect_flow.py` are the public composable primitives.
 
+Prefect can also be turned off entirely with `ckanext.datapusher_plus.prefect_enabled = false`, which routes submissions to `jobs/local_runner.py` on CKAN's RQ worker. When touching the pipeline, keep shared logic in `jobs/pipeline_core.py` so both runners stay in sync — and keep that module free of any `prefect` import, since the disabled mode exists for hosts where importing Prefect itself fails.
+
 ### Key Modules
 
 - **plugin.py** — CKAN plugin entry point, implements IConfigurer, IConfigurable, IActions, IAuthFunctions, IPackageController, IResourceUrlChange, IResourceController, ITemplateHelpers, IBlueprint, IClick (+ IFormRedirect conditionally)
@@ -168,3 +180,4 @@ Key settings in `ckan.ini` (see config.py and config_declaration.yaml for the fu
 - `ckanext.datapusher_plus.prefer_dmy` — Date format preference (DMY vs MDY)
 - `ckanext.datapusher_plus.enable_druf` — Enable DRUF workflow
 - `ckanext.datapusher_plus.enable_form_redirect` — Enable IFormRedirect interface
+- `ckanext.datapusher_plus.prefect_enabled` — Orchestrate with Prefect (default: true); `false` runs jobs in-process on `ckan jobs worker`

CONFIG.md

@@ -58,6 +58,27 @@ When DRUF is enabled, the following templates are overridden:
 - Works with standard CKAN installations
 - Compatible with ckanext-scheming
 
+### Turning Prefect off
+
+By default (v3.0+) ingestion jobs are orchestrated by a Prefect server + worker. Setting `prefect_enabled = false` runs them in-process on CKAN's own background-job worker instead, over the same ingestion stages, with nothing in the path importing `prefect`.
+
+**Configuration:**
+```ini
+# Orchestrate jobs with Prefect (default: true)
+ckanext.datapusher_plus.prefect_enabled = false
+```
+
+**What it does:**
+- `datapusher_submit` enqueues the job on CKAN's RQ queue instead of creating a Prefect flow run
+- `ckanext/datapusher_plus/jobs/local_runner.py` executes the nine stages sequentially in the worker process
+- The `Jobs`/`Logs` tables, the job-status page, and the `datapusher_hook` callbacks behave identically
+
+**Requirements:**
+- A running CKAN worker: `ckan -c /etc/ckan/default/ckan.ini jobs worker`
+- No Prefect server, worker, or work pool (`datapusher_plus prefect-deploy` refuses to run in this mode)
+
+**What you give up:** per-stage retries, result caching / re-run-from-failed-stage, the Prefect run graph, artifacts and `datapusher.*` events, and human-in-the-loop PII review (a job crossing `pii_review_threshold` aborts before any datastore write instead of waiting for approval). See [Running without Prefect](README.md#running-without-prefect) for the full comparison — including the common trigger for wanting it, a `PermissionError` on `$PREFECT_HOME/profiles.toml` when CKAN cannot write `$HOME/.prefect`.
+
 ## Example Configuration
 
 Add these lines to your CKAN configuration file (e.g., `/etc/ckan/default/ckan.ini`):
@@ -68,6 +89,10 @@ ckanext.datapusher_plus.enable_druf = true
 
 # Enable IFormRedirect for better form redirects (recommended with DRUF)
 ckanext.datapusher_plus.enable_form_redirect = true
+
+# Run ingestions in-process on CKAN's job worker instead of Prefect
+# (default: true — leave unset to keep Prefect orchestration)
+ckanext.datapusher_plus.prefect_enabled = false
 ```
 
 **Recommended combinations:**

README.md

@@ -570,7 +570,7 @@ ckan -c /etc/ckan/default/ckan.ini datapusher_plus submit {dataset_id}
 
 ## Prefect orchestration (v3.0+)
 
-DataPusher+ v3.0 replaces the v2 RQ-based background worker with a [Prefect 3](https://docs.prefect.io/v3/) flow. RQ is no longer used by DP+ itself (CKAN continues to ship RQ for unrelated extensions).
+DataPusher+ v3.0 replaces the v2 RQ-based background worker with a [Prefect 3](https://docs.prefect.io/v3/) flow. RQ is no longer used by DP+ itself (CKAN continues to ship RQ for unrelated extensions) — unless you set `ckanext.datapusher_plus.prefect_enabled = false`, which runs ingestions on CKAN's RQ worker instead of Prefect; see [Running without Prefect](#running-without-prefect).
 
 ### Why Prefect
 
@@ -733,6 +733,7 @@ The **default** DP+ flow does NOT call these subflows — it inlines the underly
 
 | Key | Default | Purpose |
 |---|---|---|
+| `ckanext.datapusher_plus.prefect_enabled` | `true` | Orchestrate ingestions with Prefect. `false` runs them in-process on CKAN's own job worker with no Prefect import at all — see [Running without Prefect](#running-without-prefect). |
 | `ckanext.datapusher_plus.prefect_deployment_name` | `datapusher-plus/datapusher-plus` | Fully-qualified Prefect deployment name (`<flow>/<deployment>`). |
 | `ckanext.datapusher_plus.prefect_work_pool` | `datapusher-plus` | Work-pool name workers subscribe to. |
 | `ckanext.datapusher_plus.prefect_flow` | _(unset)_ | `module.path:flow_name` entrypoint of a custom flow. |
@@ -765,11 +766,61 @@ takes effect on the next flow run without a worker restart.
 
 Resolution order: Prefect Variable -> env var -> `ckan.ini` -> built-in default. Variable lookup failures (Prefect server unreachable, name absent, value not int-parseable) silently fall through to the next priority — operators with no Prefect Variables set see no behaviour change.
 
+## Running without Prefect
+
+Prefect can be turned off entirely:
+
+```ini
+ckanext.datapusher_plus.prefect_enabled = false
+```
+
+Submissions are then enqueued on **CKAN's own background-job queue** and executed in-process by `ckanext/datapusher_plus/jobs/local_runner.py`, which runs the same nine ingestion stages in the same order. Nothing in the submit or job path imports `prefect`.
+
+Instead of a Prefect server + worker, you run CKAN's built-in worker:
+
+```bash
+ckan -c /etc/ckan/default/ckan.ini jobs worker
+```
+
+That is the only operational change — `ckan datapusher_plus submit` / `resubmit`, the DRUF workflow, the job-status page, the `Jobs`/`Logs` tables, and the `datapusher_hook` callbacks all behave as before. `ckan datapusher_plus prefect-deploy` refuses to run in this mode (there is no deployment to register).
+
+### When you'd want this
+
+* **Prefect can't be run at all** — including the case where merely *importing* it fails. A CKAN process running with `HOME=/root` but no write access there fails submission with:
+
+  ```
+  ERROR [ckanext.datapusher_plus.logic.action] Error submitting job to DataPusher: [Errno 13] Permission denied: '/root/.prefect/profiles.toml'
+  ```
+
+  Prefect writes its profile store to `$PREFECT_HOME` (default `$HOME/.prefect`) the first time it is imported. If you'd rather keep Prefect, the alternative fix is to point `PREFECT_HOME` at a directory the CKAN user can write and restart CKAN and the worker:
+
+  ```bash
+  PREFECT_HOME=/var/lib/ckan/prefect
+  ```
+
+* **Small or single-node deployments** where a second orchestration service isn't worth the operational surface.
+* **Air-gapped or locked-down hosts** where the Prefect server isn't permitted.
+
+### What you give up
+
+| Capability | With Prefect | With `prefect_enabled = false` |
+|---|---|---|
+| Per-stage retries / backoff | Yes | No — the job fails and is resubmitted |
+| Result caching, re-run from failed stage | Yes | No |
+| Run graph, artifacts, `datapusher.*` events | Yes | No (events are no-ops) |
+| Human-in-the-loop PII review (`pii_review_threshold`) | Suspends for approval | Aborts the job before any datastore write |
+| Datastore cleanup after a failed write group | Transactional rollback | Same cleanup, minus the database stage's own failure (see below) |
+| Horizontal scaling | Add Prefect workers | Add `ckan jobs worker` processes |
+
+Rollback difference in detail: when a stage *after* the database load fails, the local runner drops the half-built datastore table and restores a stashed Data Dictionary, exactly as the Prefect `on_rollback` hook does. When the **database stage itself** raises, it leaves the datastore alone — that stage can fail before touching anything (e.g. "could not connect to the Datastore"), and dropping there would destroy data the run never wrote.
+
 ### Troubleshooting
 
 | Symptom | Likely cause | Fix |
 |---|---|---|
 | `datapusher_submit` returns `False` with a Prefect connection error in the CKAN log | The Prefect server is unreachable from CKAN | Check `PREFECT_API_URL` and that the Prefect server is healthy at `<API>/health`. |
+| `Error submitting job to DataPusher: [Errno 13] Permission denied: '/root/.prefect/profiles.toml'` | CKAN's process can't write `$PREFECT_HOME`, so `import prefect` fails | Set `PREFECT_HOME` to a writable directory, or turn Prefect off with `ckanext.datapusher_plus.prefect_enabled = false` (see [Running without Prefect](#running-without-prefect)). |
+| With `prefect_enabled = false`, jobs stay `pending` forever | No CKAN background worker is running | Start `ckan -c /etc/ckan/default/ckan.ini jobs worker`. |
 | Flow run sits in `Scheduled` forever | No worker is polling the configured work pool | Start `prefect worker start -p datapusher-plus` on a host with the `datapusher-plus` package installed. |
 | Flow run goes straight to `Failed` with "QSV binary not found" | The worker process can't see the qsv binary | Set `ckanext.datapusher_plus.qsv_bin` in the CKAN config the worker reads, or install qsv in the worker's PATH. |
 | Re-run from a failed task re-downloads the file | Result storage block isn't registered, so persisted results aren't being read | Re-run `ckan datapusher_plus prefect-deploy` — it calls `ensure_result_storage_block`. |

ckanext/datapusher_plus/cli.py

@@ -274,6 +274,17 @@ def prefect_deploy(work_pool: str | None):
     ``datapusher_plus_flow`` otherwise. This is how operators register
     custom ingestion flows without modifying DP+.
     """
+    import ckanext.datapusher_plus.config as conf
+
+    if not conf.prefect_enabled():
+        error_shout(
+            "ckanext.datapusher_plus.prefect_enabled is false — this "
+            "deployment runs ingestions in-process on CKAN's RQ worker, "
+            "so there is no Prefect deployment to register. Run "
+            "`ckan jobs worker` instead, or set prefect_enabled = true."
+        )
+        raise click.Abort()
+
     try:
         from prefect import flow as _flow_decorator  # noqa: F401
         from prefect.deployments.runner import RunnerDeployment  # noqa: F401
@@ -457,15 +468,26 @@ def migrate_from_rq(resubmit: bool, yes: bool):
     session.commit()
     click.echo(f"Reset {reset_count} stale ``pending`` task_status rows.")
 
-    # Sanity-check Prefect server reachability.
-    try:
-        import ckanext.datapusher_plus.prefect_client as prefect_client
+    # Sanity-check Prefect server reachability — unless the operator
+    # migrated off RQ *without* adopting Prefect, in which case there is
+    # no server to reach and jobs run on CKAN's own worker.
+    import ckanext.datapusher_plus.config as conf
 
-        prefect_client.get_running_resource_ids()
-        click.echo("Prefect server is reachable.")
-    except Exception as e:
-        error_shout(f"Cannot reach Prefect server: {e}")
-        raise click.Abort()
+    if not conf.prefect_enabled():
+        click.echo(
+            "ckanext.datapusher_plus.prefect_enabled is false — skipping "
+            "the Prefect reachability check; jobs will run in-process on "
+            "CKAN's RQ worker (`ckan jobs worker`)."
+        )
+    else:
+        try:
+            import ckanext.datapusher_plus.prefect_client as prefect_client
+
+            prefect_client.get_running_resource_ids()
+            click.echo("Prefect server is reachable.")
+        except Exception as e:
+            error_shout(f"Cannot reach Prefect server: {e}")
+            raise click.Abort()
 
     # Optional: resubmit each drained resource through the new path.
     if resubmit:

ckanext/datapusher_plus/config.py

@@ -37,6 +37,37 @@
 # TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL
 UPLOAD_LOG_LEVEL = tk.config.get("ckanext.datapusher_plus.upload_log_level", "INFO")
 
+
+# Orchestration backend.
+#
+# ``True`` (default) submits every ingestion to Prefect. ``False`` turns
+# Prefect off completely: jobs are enqueued on CKAN's own RQ background
+# queue and executed in-process by ``jobs/local_runner.py``, and nothing
+# in the request or job path imports ``prefect``. That matters for
+# deployments where importing Prefect itself fails — e.g. a CKAN process
+# running with ``HOME=/root`` but no write access there, where Prefect's
+# settings bootstrap raises ``PermissionError: '/root/.prefect/profiles.toml'``.
+#
+# Deliberately a function rather than a module-level constant: the flag
+# gates a code path in the web request (``datapusher_submit``) and in the
+# CLI, so operators flipping it in ``ckan.ini`` should not need a
+# restart-order-of-operations lesson. Reading it live also keeps
+# ``config_declaration.yaml``'s declared default authoritative under
+# CKAN 2.10+.
+def prefect_enabled() -> bool:
+    """Return whether ingestion jobs are orchestrated by Prefect.
+
+    Falls back to ``True`` in contexts where CKAN config is not loaded
+    (bare tooling imports), matching the shipped default.
+    """
+    try:
+        value = tk.config.get("ckanext.datapusher_plus.prefect_enabled")
+    except Exception:
+        return True
+    if value is None or value == "":
+        return True
+    return tk.asbool(value)
+
 # Supported formats
 FORMATS = tk.config.get(
     "ckanext.datapusher_plus.formats",