From 043a7812e573204c20286c60a3fa3ac0cb5c1ebc Mon Sep 17 00:00:00 2001 From: Mohammed Minhajuddin <68331751+minhajuddin2510@users.noreply.github.com> Date: Wed, 5 Aug 2026 04:17:36 -0400 Subject: [PATCH] feat: add ckanext.datapusher_plus.prefect_enabled to run jobs without Prefect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CHANGELOG.md | 4 + CLAUDE.md | 15 +- CONFIG.md | 25 + README.md | 53 +- ckanext/datapusher_plus/cli.py | 38 +- ckanext/datapusher_plus/config.py | 31 + .../datapusher_plus/config_declaration.yaml | 24 + ckanext/datapusher_plus/jobs/__init__.py | 41 +- ckanext/datapusher_plus/jobs/events.py | 8 + ckanext/datapusher_plus/jobs/local_runner.py | 486 +++++++++++++++ ckanext/datapusher_plus/jobs/pipeline_core.py | 326 ++++++++++ ckanext/datapusher_plus/jobs/prefect_flow.py | 274 +------- ckanext/datapusher_plus/logic/action.py | 61 +- ckanext/datapusher_plus/plugin.py | 13 + tests/test_dictionary_stash.py | 12 +- tests/test_local_runner.py | 584 ++++++++++++++++++ tests/test_prefect_flow.py | 25 +- 17 files changed, 1729 insertions(+), 291 deletions(-) create mode 100644 ckanext/datapusher_plus/jobs/local_runner.py create mode 100644 ckanext/datapusher_plus/jobs/pipeline_core.py create mode 100644 tests/test_local_runner.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 310b2d49..d896353a 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/CLAUDE.md b/CLAUDE.md index 2138aacf..de42a722 100644 --- a/CLAUDE.md +++ b/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` diff --git a/CONFIG.md b/CONFIG.md index d6ffd3db..fe2ba4fc 100644 --- a/CONFIG.md +++ b/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:** diff --git a/README.md b/README.md index 862b21e7..0743fdde 100644 --- a/README.md +++ b/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 (`/`). | | `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 `/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`. | diff --git a/ckanext/datapusher_plus/cli.py b/ckanext/datapusher_plus/cli.py index 9b5058af..5d3efae7 100644 --- a/ckanext/datapusher_plus/cli.py +++ b/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: diff --git a/ckanext/datapusher_plus/config.py b/ckanext/datapusher_plus/config.py index d7231bd8..deec34e2 100644 --- a/ckanext/datapusher_plus/config.py +++ b/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", diff --git a/ckanext/datapusher_plus/config_declaration.yaml b/ckanext/datapusher_plus/config_declaration.yaml index 448dc8ca..dae7e34b 100644 --- a/ckanext/datapusher_plus/config_declaration.yaml +++ b/ckanext/datapusher_plus/config_declaration.yaml @@ -456,6 +456,30 @@ groups: # --------------------------------------------------------------- # Prefect orchestration (v3.0+) # --------------------------------------------------------------- + - key: ckanext.datapusher_plus.prefect_enabled + type: bool + editable: true + default: true + description: | + Whether ingestion jobs are orchestrated by Prefect (the + default). Set to ``false`` to turn Prefect off entirely: jobs + are then enqueued on CKAN's own background-job queue and run + in-process by ``jobs/local_runner.py`` over the same ingestion + stages, and nothing in the submit or job path imports + ``prefect``. Requires a running ``ckan jobs worker`` instead + of a Prefect server + worker. + + Use it when you cannot run Prefect — including the case where + importing it fails, e.g. a CKAN process whose ``$PREFECT_HOME`` + (``$HOME/.prefect``) is not writable and raises + ``PermissionError: '/root/.prefect/profiles.toml'`` on submit. + + What you give up with Prefect off: per-stage retries, task + result caching / re-run-from-failed-stage, the run graph, + artifacts and ``datapusher.*`` events, and human-in-the-loop + PII review (``pii_review_threshold`` then aborts a flagged job + before any datastore write instead of waiting for approval). + - key: ckanext.datapusher_plus.prefect_deployment_name editable: true default: datapusher-plus/datapusher-plus diff --git a/ckanext/datapusher_plus/jobs/__init__.py b/ckanext/datapusher_plus/jobs/__init__.py index 56e7f7ab..5c1a4e17 100644 --- a/ckanext/datapusher_plus/jobs/__init__.py +++ b/ckanext/datapusher_plus/jobs/__init__.py @@ -8,10 +8,16 @@ ``prefect_flow``. Custom flows registered via ``ckanext.datapusher_plus.prefect_flow`` should import from there. +Operators who cannot (or would rather not) run Prefect set +``ckanext.datapusher_plus.prefect_enabled = false``; ingestions then run +through ``local_runner.run_job`` on CKAN's own RQ worker, over the same +stage classes, with Prefect out of the picture entirely. The pieces both +runners share live in ``pipeline_core``. + This module exposes a small public surface (``datapusher_plus_flow``, ``push_to_datastore``, ``datapusher_plus_to_datastore``, -``callback_datapusher_hook``) but does NOT import Prefect at module -load. Eager imports here would pull in the Prefect runtime every time +``callback_datapusher_hook``, ``run_job``) but does NOT import Prefect at +module load. Eager imports here would pull in the Prefect runtime every time CKAN loads the DP+ plugin (during ``ckan db init``, ``ckan plugins info``, etc.) — and Prefect spins up an ephemeral server when no ``PREFECT_API_URL`` is configured, polluting stdout with log lines and @@ -28,13 +34,16 @@ def push_to_datastore( ) -> Optional[str]: """Backward-compat shim for the v2 ``push_to_datastore`` callable. - Constructs a ``JobInput`` from the legacy arg shape and invokes the - Prefect flow. Useful for tests that drive the flow as a plain Python - function without going through a Prefect worker. + Constructs a ``JobInput`` from the legacy arg shape and runs the + ingestion in-process: through the Prefect flow by default, or + through ``local_runner.run_job`` when + ``ckanext.datapusher_plus.prefect_enabled`` is false (in which case + nothing here imports Prefect). Useful for tests and scripts that + drive a job as a plain Python function. """ # Lazy: avoid pulling Prefect into CKAN admin commands that import # this module but never run a job. - from ckanext.datapusher_plus.jobs.prefect_flow import datapusher_plus_flow + import ckanext.datapusher_plus.config as conf from ckanext.datapusher_plus.jobs.runtime_context import JobInput metadata = input.get("metadata", {}) @@ -45,6 +54,14 @@ def push_to_datastore( input=input, dry_run=dry_run, ) + + if not conf.prefect_enabled(): + from ckanext.datapusher_plus.jobs.local_runner import run_job + + return run_job(job_input) + + from ckanext.datapusher_plus.jobs.prefect_flow import datapusher_plus_flow + return datapusher_plus_flow(job_input) @@ -60,9 +77,18 @@ def __getattr__(name: str): return datapusher_plus_flow if name == "callback_datapusher_hook": - from ckanext.datapusher_plus.jobs.prefect_flow import callback_datapusher_hook + # Sourced from the Prefect-free core: the callback is identical + # on both runners, and resolving it must not drag Prefect in on + # the ``prefect_enabled = false`` path. + from ckanext.datapusher_plus.jobs.pipeline_core import ( + callback_datapusher_hook, + ) return callback_datapusher_hook + if name == "run_job": + from ckanext.datapusher_plus.jobs.local_runner import run_job + + return run_job raise AttributeError( f"module 'ckanext.datapusher_plus.jobs' has no attribute {name!r}" ) @@ -84,4 +110,5 @@ def __dir__(): "datapusher_plus_to_datastore", "push_to_datastore", "callback_datapusher_hook", + "run_job", ] diff --git a/ckanext/datapusher_plus/jobs/events.py b/ckanext/datapusher_plus/jobs/events.py index 749763e4..517d7b8e 100644 --- a/ckanext/datapusher_plus/jobs/events.py +++ b/ckanext/datapusher_plus/jobs/events.py @@ -21,6 +21,14 @@ def _safe_emit(event: str, resource_id: str, payload: Dict[str, Any]) -> None: """Best-effort emit; never let a failed event fail the flow.""" + # There is nothing to emit to when Prefect is turned off, and the + # import below is exactly what that mode exists to avoid (Prefect's + # settings bootstrap writes to ``$PREFECT_HOME``). The local runner + # reaches this via ``quarantine.apply_quarantine``. + import ckanext.datapusher_plus.config as conf + + if not conf.prefect_enabled(): + return try: from prefect.events import emit_event diff --git a/ckanext/datapusher_plus/jobs/local_runner.py b/ckanext/datapusher_plus/jobs/local_runner.py new file mode 100644 index 00000000..a363730e --- /dev/null +++ b/ckanext/datapusher_plus/jobs/local_runner.py @@ -0,0 +1,486 @@ +# -*- coding: utf-8 -*- +# flake8: noqa: E501 +""" +In-process ingestion runner for ``prefect_enabled = false``. + +This is the fallback path DataPusher+ takes when an operator turns +Prefect off in ``ckan.ini``:: + + ckanext.datapusher_plus.prefect_enabled = false + +``datapusher_submit`` then enqueues :func:`run_job` on CKAN's own RQ +background queue instead of creating a Prefect flow run, and a plain +``ckan jobs worker`` executes the same nine stage classes the Prefect +flow runs — sequentially, in one process, sharing one +``RuntimeContext``. No Prefect server, no Prefect worker, no Prefect +work pool, and — importantly — no ``import prefect`` anywhere in the +path: on a host where ``$PREFECT_HOME`` is unwritable, that import is +itself the failure (``PermissionError: '/root/.prefect/profiles.toml'``). + +What is identical to the Prefect flow: + +* the nine stages and the order they run in; +* the ``Jobs``-row state machine (pending → completed / errored) and the + ``Logs`` table written through ``utils.StoringHandler``; +* the ``datapusher_hook`` callbacks (running / complete / error) that + drive default views, ``IDataPusher.after_upload``, and auto-resubmit; +* the "stage returned ``None``" complete-with-skip contract; +* the soft between-stage deadline from + ``ckanext.datapusher_plus.flow_timeout``; +* dropping the datastore table (and restoring a stashed Data Dictionary) + when a stage after the database load fails. + +What Prefect gives you that this does not — the reason it stays the +default: + +* per-task retries and backoff (a flaky download fails the job here); +* task result caching / re-run-from-failed-stage; +* the run graph, artifacts, and ``datapusher.*`` events; +* human-in-the-loop PII review — ``pii_review_threshold`` cannot suspend + a job here, so crossing the gate aborts before any datastore write + rather than waiting for an approval that can never arrive; +* per-stage rollback granularity: when the *database* stage itself + raises, this runner leaves the datastore alone rather than dropping a + table it may never have touched (a failure to even connect must not + destroy intact data). Cleanup happens for failures *after* the load, + which is where the table is known to be half-built. + +Retry policy is therefore RQ's: the job fails, and the operator (or +``ckan datapusher_plus resubmit``) submits it again. +""" + +from __future__ import annotations + +import logging +import sys +import tempfile +import time +import traceback +from dataclasses import asdict +from typing import Any, Dict, Optional, Set, Union + +import sqlalchemy as sa + +import ckan.plugins.toolkit as tk + +import ckanext.datapusher_plus.config as conf +import ckanext.datapusher_plus.helpers as dph +import ckanext.datapusher_plus.utils as utils +from ckanext.datapusher_plus import __version__ as _dpp_version +import ckanext.datapusher_plus.dictionary_stash as dict_stash +from ckanext.datapusher_plus.jobs import quarantine +from ckanext.datapusher_plus.jobs.pipeline_core import ( + StageAbort, + build_runtime_context, + callback_datapusher_hook, + resolve_int, + resource_is_datastore_dump, + rollback_datastore_writes, + run_stage, + validate_input, +) +from ckanext.datapusher_plus.jobs.runtime_context import ( + JobInput, + reset_runtime_context, + set_runtime_context, +) +from ckanext.datapusher_plus.jobs.stages.ai_suggestions import AISuggestionsStage +from ckanext.datapusher_plus.jobs.stages.analysis import AnalysisStage +from ckanext.datapusher_plus.jobs.stages.database import DatabaseStage +from ckanext.datapusher_plus.jobs.stages.download import DownloadStage +from ckanext.datapusher_plus.jobs.stages.format_converter import FormatConverterStage +from ckanext.datapusher_plus.jobs.stages.formula import FormulaStage +from ckanext.datapusher_plus.jobs.stages.indexing import IndexingStage +from ckanext.datapusher_plus.jobs.stages.metadata import MetadataStage +from ckanext.datapusher_plus.jobs.stages.validation import ValidationStage + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Submission +# --------------------------------------------------------------------------- + + +def enqueue_job(job_input: JobInput, *, timeout: Optional[int] = None) -> str: + """Enqueue an ingestion on CKAN's RQ background queue. + + The local-mode counterpart of ``prefect_client.submit_flow_run``: + non-blocking, returns as soon as Redis has accepted the job. A + ``ckan jobs worker`` (CKAN's built-in worker — no extra service) + picks it up and calls :func:`run_job`. + + The payload is the ``asdict``-ed ``JobInput``, i.e. exactly what the + Prefect deployment receives as its ``job_input`` parameter, so both + paths carry the same data contract. + + Args: + job_input: the job to run. + timeout: job timeout in seconds, handed to RQ. Pass one: + ``None`` leaves RQ's own default (180s) in place, which + kills any ingestion of consequence mid-COPY. + ``datapusher_submit`` passes + ``ckanext.datapusher_plus.flow_timeout`` here, the same + value the runner enforces between stages. + + Returns: + The RQ job id, which the caller records on the CKAN + ``task_status`` row. + + Raises: + Whatever ``enqueue_job`` raises when Redis is unreachable — + the caller surfaces it exactly as it does a Prefect failure. + """ + rq_kwargs: Dict[str, Any] = {} + if timeout is not None: + rq_kwargs["timeout"] = timeout + + job = tk.enqueue_job( + run_job, + [asdict(job_input)], + title=f"DataPusher+ ingest {job_input.resource_id}", + rq_kwargs=rq_kwargs or None, + ) + return str(job.id) + + +def get_running_resource_ids() -> Set[str]: + """Return the set of CKAN resource_ids currently being ingested. + + The local-mode counterpart of + ``prefect_client.get_running_resource_ids``: scans CKAN's RQ queue + (both waiting and started jobs) for DP+ ingestions and pulls + ``resource_id`` out of each one's payload. ``datapusher_submit`` uses + it to skip duplicate submissions of a resource that is already + in-flight. + + Unlike the v2 regex-over-``job.description`` scan this reads the + enqueued payload directly, so it cannot be defeated by a change in + how RQ renders a job's description. + + Any error is logged and treated as "nothing running" — same + fail-open behaviour as the Prefect implementation when its server is + unreachable, so an unavailable Redis never blocks a submission. + """ + ids: Set[str] = set() + try: + import ckan.lib.jobs as rq_jobs + + queue = rq_jobs.get_queue() + jobs = list(queue.get_jobs()) + + # Waiting jobs only come from ``queue.get_jobs()``; a job that a + # worker already picked up has moved to the started registry and + # is exactly the case we most need to catch. + try: + from rq.registry import StartedJobRegistry + + registry = StartedJobRegistry(queue=queue) + for job_id in registry.get_job_ids(): + job = queue.fetch_job(job_id) + if job is not None: + jobs.append(job) + except Exception as e: + log.warning("Could not read RQ started-job registry: %s", e) + + for job in jobs: + args = job.args or () + payload = args[0] if args else None + if isinstance(payload, dict) and payload.get("resource_id"): + ids.add(payload["resource_id"]) + except Exception as e: + log.warning("Failed to query RQ for running resource_ids: %s", e) + return set() + return ids + + +# --------------------------------------------------------------------------- +# Execution +# --------------------------------------------------------------------------- + + +def _current_rq_job_id() -> Optional[str]: + """The id of the RQ job we are executing inside, if any.""" + try: + from rq import get_current_job + + job = get_current_job() + return str(job.id) if job is not None else None + except Exception: + return None + + +def _guard_pii_review(runtime) -> None: + """Abort before any datastore write when the PII review gate is crossed. + + ``ckanext.datapusher_plus.pii_review_threshold`` asks for a human to + approve an ingestion whose analysis flagged PII. The Prefect flow + honours that by suspending the run until an operator answers a form + in the Prefect UI; with Prefect off there is nothing that can hold a + job open and nothing to answer on, so the only reading of the + operator's intent that stays safe is "do not load it". + + No-op when the feature is off (threshold ``0``, the default) or when + analysis found no PII. + """ + threshold = resolve_int( + "DATAPUSHER_PLUS_PII_REVIEW_THRESHOLD", + "ckanext.datapusher_plus.pii_review_threshold", + 0, + ) + if threshold <= 0 or not runtime.pii_found: + return + + pii_count = runtime.pii_candidate_count + # Quick-screen mode only detects PII *presence* — pii_candidate_count + # is a degenerate 1 that no numeric threshold could ever exceed, so + # there the threshold acts purely as the feature's on/off switch. + if not conf.PII_QUICK_SCREEN and pii_count <= threshold: + return + + raise utils.JobError( + f"PII review gate crossed: {pii_count} candidate match(es) " + f"(threshold={threshold}). Human-in-the-loop review requires " + "Prefect, which is disabled " + "(ckanext.datapusher_plus.prefect_enabled = false), so the job " + "was aborted before any datastore write. Re-enable Prefect to " + "review and approve these runs, or set " + "ckanext.datapusher_plus.pii_review_threshold = 0 to load them " + "without review." + ) + + +def run_job(job_input: Union[JobInput, Dict[str, Any]]) -> Optional[str]: + """Ingest one CKAN resource into the datastore, in this process. + + The Prefect-free twin of ``datapusher_plus_flow``, and the callable + RQ invokes for every locally-run job. + + Returns ``None`` on success — matching the flow's return contract — + and re-raises on failure so RQ records the job as failed after the + ``Jobs`` row and the CKAN callback have been updated. + """ + # RQ hands back exactly what was enqueued (a dict); accept the + # dataclass too so callers and tests can drive this directly. + if isinstance(job_input, dict): + job_input = JobInput(**job_input) + + job_id = job_input.task_id + log.info( + f"DATAPUSHER+ v{_dpp_version} starting local (no-Prefect) job for " + f"resource {job_input.resource_id}" + ) + + # Register the job in the DP+ Jobs table at start. This is what + # ``datapusher_status`` and the CKAN UI read. + try: + dph.add_pending_job(job_id, **job_input.input) + except sa.exc.IntegrityError: + raise utils.JobError("Job already exists.") + # Column repurposed for the orchestrator's own run id — the Prefect + # flow stores its flow_run_id here, we store the RQ job id. + dph.set_aps_job_id(job_id, _current_rq_job_id() or "local") + + # Validate the input only after the Jobs row exists, so a malformed + # submission is recorded as an errored job (visible via + # datapusher_status / the CKAN UI) instead of raising with no trace. + try: + validate_input(job_input.input) + except utils.JobError as e: + dph.mark_job_as_errored(job_id, str(e)) + raise + + job_timeout_seconds = resolve_int( + "DATAPUSHER_PLUS_FLOW_TIMEOUT_SECONDS", + "ckanext.datapusher_plus.flow_timeout", + 7200, + ) + job_start = time.monotonic() + + def _check_deadline(): + """Raise if the configured deadline has been exceeded. + + Soft enforcement, checked between stages only: a stage hung + mid-execution is RQ's ``timeout`` to kill (set from the same + config value at enqueue time), not this check's. + """ + elapsed = time.monotonic() - job_start + if elapsed > job_timeout_seconds: + raise utils.JobError( + f"Job exceeded configured timeout of " + f"{job_timeout_seconds}s (elapsed: {elapsed:.0f}s)" + ) + + # Announce running state to CKAN. + result_url = job_input.input.get("result_url") + if result_url: + callback_datapusher_hook( + result_url=result_url, + job_dict={ + "metadata": job_input.input.get("metadata", {}), + "status": "running", + }, + ) + + errored = False + with tempfile.TemporaryDirectory() as temp_dir: + # ``runtime`` / ``token`` are built *inside* the try so that a + # failure in build_runtime_context (e.g. get_resource raising) is + # caught: mark_job_as_errored runs and the error callback fires, + # instead of the exception escaping and leaving the job stuck + # "running" (set by the announce callback above). + runtime = None + token = None + try: + runtime = build_runtime_context(job_input, temp_dir) + token = set_runtime_context(runtime) + + if resource_is_datastore_dump(runtime): + runtime.logger.info("Dump files are managed with the Datastore API") + dph.mark_job_as_completed(job_id, {"skipped": "datastore-managed"}) + _log_done(runtime, "(skipped: datastore-managed) ") + return None + + # Read-only / non-destructive stages. Each mutates the one + # live RuntimeContext, so — unlike the Prefect tasks, whose + # bodies may be skipped by a cache hit — no result needs to + # be threaded from stage to stage. + run_stage(DownloadStage()) + _check_deadline() + run_stage(FormatConverterStage()) + run_stage(ValidationStage()) + + # Enforce the quarantine threshold (raises if exceeded). + # No-op when no rows were rejected. + quarantine.apply_quarantine( + resource_id=runtime.resource_id, + clean_csv_path=runtime.tmp, + quarantine_csv_path=runtime.quarantine_csv_path or None, + quarantined_rows=runtime.quarantined_rows, + total_rows=runtime.rows_to_copy + runtime.quarantined_rows, + ) + + run_stage(AnalysisStage()) + _check_deadline() + + # Optional AI suggestions, before the datastore writes so a + # (rare) failure to patch the package can't poison the + # rollback path below. Gated on + # ``ckanext.datapusher_plus.enable_ai_suggestions`` (default + # False) and swallows every internal failure. + run_stage(AISuggestionsStage()) + _check_deadline() + + _guard_pii_review(runtime) + + # Datastore-mutating group. A failure *after* the database + # stage means the table is half-built: drop it (and restore + # any stashed Data Dictionary), which is what the Prefect + # flow's transaction rollback does. A failure *in* the + # database stage is deliberately left alone — see the module + # docstring. + run_stage(DatabaseStage()) + _check_deadline() + try: + run_stage(IndexingStage()) + run_stage(FormulaStage()) + run_stage(MetadataStage()) + except Exception: + # Catches ``StageAbort`` too — one of these stages + # returning ``None`` after the load has committed leaves + # the same half-finished table as a raise, and under + # Prefect it would likewise fail the transaction and + # fire the rollback hooks. The outer handler still marks + # the job complete-with-skip. + rollback_datastore_writes(runtime) + raise + + if job_input.dry_run: + dph.mark_job_as_completed(job_id, {"headers": runtime.headers_dicts}) + _log_done(runtime, "(dry-run) ") + return None + + if runtime.quarantined_rows > 0: + runtime.logger.warning( + f"{runtime.quarantined_rows} row(s) quarantined to " + f"{runtime.quarantine_csv_path}" + ) + + dph.mark_job_as_completed( + job_id, + { + "rows": runtime.copied_count, + "headers": runtime.headers_dicts, + }, + ) + _log_done(runtime) + return None + + except StageAbort as e: + # A stage signalled "nothing to do" by returning None (e.g. + # the Analysis stage on a zero-record file). v2 stopped the + # pipeline here and the job *completed* — there was simply + # nothing to load. Match that: complete-with-skip, not an + # error. ``errored`` stays False so the finally block fires + # the "complete" callback. + if runtime is not None: + runtime.logger.info(str(e)) + log.info(str(e)) + dph.mark_job_as_completed(job_id, {"skipped": e.stage_name}) + _log_done(runtime, f"(skipped: {e.stage_name}) ") + return None + except utils.JobError as e: + errored = True + dph.mark_job_as_errored(job_id, str(e)) + if runtime is not None: + runtime.logger.error(f"DataPusher Plus error: {e}") + log.error(f"DataPusher Plus error: {e}") + raise + except Exception as e: + errored = True + tb = traceback.format_tb(sys.exc_info()[2])[-1] + repr(e) + dph.mark_job_as_errored(job_id, tb) + if runtime is not None: + runtime.logger.error( + f"DataPusher Plus error: {e}, {traceback.format_exc()}" + ) + log.error(f"DataPusher Plus error: {e}") + raise + finally: + if token is not None: + reset_runtime_context(token) + # Issue #265: on any successful exit (including the + # ``StageAbort`` complete-with-skip), drop the Data + # Dictionary stash — the run reached its natural end. On + # failure, leave the stash for the rollback path (and a + # possible resubmission). + if not errored: + try: + dict_stash.clear(job_input.resource_id) + except Exception as e: # noqa: BLE001 — never block teardown on stash cleanup + log.warning( + f"Could not clear dictionary stash for " + f"{job_input.resource_id}: {e}" + ) + if result_url: + status = "error" if errored else "complete" + saved_ok = callback_datapusher_hook( + result_url=result_url, + job_dict={ + "metadata": job_input.input.get("metadata", {}), + "status": status, + }, + ) + if not saved_ok and not errored: + dph.mark_job_as_failed_to_post_result(job_id) + + +def _log_done(runtime, marker: str = "") -> None: + """Emit the "JOB DONE!" capstone (issue #111) on every success path.""" + if runtime is None: + return + total_elapsed = time.time() - runtime.timer_start + runtime.logger.info( + f"DATAPUSHER+ v{_dpp_version} JOB DONE! {marker}" + f"Total elapsed time: {total_elapsed:,.2f} seconds." + ) diff --git a/ckanext/datapusher_plus/jobs/pipeline_core.py b/ckanext/datapusher_plus/jobs/pipeline_core.py new file mode 100644 index 00000000..b5f149e1 --- /dev/null +++ b/ckanext/datapusher_plus/jobs/pipeline_core.py @@ -0,0 +1,326 @@ +# -*- coding: utf-8 -*- +# flake8: noqa: E501 +""" +Prefect-free core shared by both DataPusher+ job runners. + +DP+ v3 can execute the same nine ingestion stages two ways: + +* ``jobs/prefect_flow.py`` — the default, Prefect-orchestrated flow + (per-task retries, result caching, run artifacts, transactional + rollback, PII suspend-for-review). +* ``jobs/local_runner.py`` — the in-process fallback selected by + ``ckanext.datapusher_plus.prefect_enabled = false``. Runs on CKAN's own + RQ background-job worker with no Prefect server, worker, or import + anywhere in the path. + +Everything both runners need — the CKAN status callback, input +validation, ``RuntimeContext`` construction, the stage invoker and its +"nothing to do" signal, and the datastore rollback body — lives here so +there is exactly one implementation of each. + +**Nothing in this module (or anything it imports) may import +``prefect``.** Avoiding that import is the whole point of the disabled +mode: on a deployment whose ``$PREFECT_HOME`` is not writable (a CKAN +process running with ``HOME=/root``, say) even ``import prefect`` raises +``PermissionError`` while it tries to create ``profiles.toml``. +""" + +from __future__ import annotations + +import json +import logging +import os +import time +from pathlib import Path +from typing import Any, Dict, Optional + +import requests + +import ckanext.datapusher_plus.config as conf +import ckanext.datapusher_plus.datastore_utils as dsu +import ckanext.datapusher_plus.dictionary_stash as dict_stash +import ckanext.datapusher_plus.utils as utils +from ckanext.datapusher_plus.jobs.context import ProcessingContext +from ckanext.datapusher_plus.jobs.runtime_context import ( + JobInput, + RuntimeContext, + get_runtime_context, + rehydrate, +) +from ckanext.datapusher_plus.logging_utils import TRACE +from ckanext.datapusher_plus.qsv_utils import QSVCommand + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Tunable resolution +# --------------------------------------------------------------------------- + + +def resolve_int(env_name: str, config_key: str, default: int) -> int: + """Resolve an int tunable from env → ckan.ini → default. + + Env wins so operators and CI can override per-process without + touching ``ckan.ini``. When the env var is unset, fall back to the + CKAN config key. When nothing is set, return ``default``. + + ``prefect_flow._resolve_int`` layers a Prefect Variable lookup on + top of this and then delegates here; the local runner uses it + directly (there is no Prefect server to hold Variables). + """ + env_value = os.environ.get(env_name) + if env_value is not None and env_value != "": + try: + return int(env_value) + except ValueError: + pass + try: + import ckan.plugins.toolkit as tk + + v = tk.config.get(config_key) + if v is not None and v != "": + return int(v) + except Exception: + # CKAN config not loaded (e.g., when running in a bare worker + # process) — fall through to the default. + pass + return default + + +# --------------------------------------------------------------------------- +# Callback helper +# --------------------------------------------------------------------------- + + +def callback_datapusher_hook(result_url: str, job_dict: Dict[str, Any]) -> bool: + """ + POST a status update to CKAN's ``datapusher_hook`` endpoint. + + Preserves the v2 contract: the worker reports running/complete/error + state by POSTing here, which drives default-view creation, plugin + ``IDataPusher.after_upload`` hooks, and auto-resubmit on file change. + """ + api_token = utils.get_dp_plus_user_apitoken() + headers = { + "Content-Type": "application/json", + "Authorization": api_token, + } + try: + response = requests.post( + result_url, + data=json.dumps(job_dict, cls=utils.DatetimeJsonEncoder), + verify=conf.SSL_VERIFY, + headers=headers, + timeout=30, + ) + except requests.ConnectionError: + return False + return response.status_code == requests.codes.ok + + +# --------------------------------------------------------------------------- +# Stage invocation +# --------------------------------------------------------------------------- + + +class StageAbort(Exception): + """A stage returned ``None`` — the BaseStage "nothing to do" signal. + + Per the ``BaseStage`` contract, ``process()`` may return ``None`` to + stop the rest of the pipeline gracefully (e.g. the Analysis stage on + a zero-record file logs "Upload skipped as there are zero records" + and returns ``None``). The v2 pipeline stopped there and the job + *completed* — nothing was wrong, there was simply nothing to load. + + Raised by ``run_stage`` and caught distinctly from ``JobError`` by + both runners so the job is marked complete-with-skip, not errored. + """ + + def __init__(self, stage_name: str): + self.stage_name = stage_name + super().__init__( + f"Stage {stage_name} stopped the pipeline (nothing to do)" + ) + + +def run_stage(stage, prev: Any = None) -> RuntimeContext: + """Invoke a stage on the bound RuntimeContext. + + ``prev`` is the upstream task's result. When given, the bound + ``RuntimeContext`` is rehydrated from it first, so the stage sees + correct ``ctx`` state even if the upstream task's body never ran (a + Prefect cache hit, or a persisted-result replay on a flow re-run) — + that body is what would otherwise have mutated the shared context. + The root task (``download_task``) passes no ``prev``, and neither + does the local runner: there, every stage mutates one live context + in a single process, so there is nothing to rehydrate from. + + A stage returning ``None`` is the BaseStage "skip / nothing to do" + signal (per its docstring) — surfaced here as ``StageAbort`` so the + caller can stop cleanly and mark the job *complete*, not errored. + """ + ctx = get_runtime_context() + if prev is not None: + rehydrate(ctx, prev) + result = stage(ctx) + if result is None: + raise StageAbort(stage.name) + return result + + +# --------------------------------------------------------------------------- +# Pre-flight helpers +# --------------------------------------------------------------------------- + + +def validate_input(input_payload: Dict[str, Any]) -> None: + """Mirror of v2 ``pipeline.validate_input``.""" + if "metadata" not in input_payload: + raise utils.JobError("Metadata missing") + if "resource_id" not in input_payload["metadata"]: + raise utils.JobError("No id provided.") + + +def build_runtime_context(job_input: JobInput, temp_dir: str) -> RuntimeContext: + """ + Construct the per-run ``RuntimeContext`` (== legacy ``ProcessingContext``). + + Sets up the task-scoped logger with both the v2 ``StoringHandler`` (so + the DP+ ``Logs`` table continues to populate, and the CKAN UI's job + detail view keeps working) and a stream handler for the worker's + stdout. + """ + task_id = job_input.task_id + input_payload = job_input.input + + # Task-scoped logger — same approach as v2 ``_push_to_datastore``. + handler = utils.StoringHandler(task_id, input_payload) + logger = logging.getLogger(task_id) + logger.addHandler(handler) + logger.addHandler(logging.StreamHandler()) + try: + log_level = getattr(logging, conf.UPLOAD_LOG_LEVEL.upper()) + except AttributeError: + log_level = TRACE + logger.setLevel(log_level) + logger.info(f"Setting log level to {logging.getLevelName(int(log_level))}") + + if not Path(conf.QSV_BIN).is_file(): + raise utils.JobError(f"{conf.QSV_BIN} not found.") + + qsv = QSVCommand(logger=logger) + + # Fetch the resource (one retry, as in v2). + resource_id = job_input.resource_id + try: + resource = dsu.get_resource(resource_id) + except utils.JobError: + time.sleep(5) + resource = dsu.get_resource(resource_id) + + ctx = ProcessingContext( + task_id=task_id, + input=input_payload, + dry_run=job_input.dry_run, + temp_dir=temp_dir, + logger=logger, + qsv=qsv, + resource=resource, + resource_id=resource_id, + ckan_url=job_input.ckan_url, + # Stamp now so the duration-since-start computed in the success + # event (``time.time() - timer_start``) is meaningful. + timer_start=time.time(), + ) + return ctx + + +def resource_is_datastore_dump(ctx: RuntimeContext) -> bool: + """v2 early-exit: ``url_type == 'datastore'`` resources are not re-ingested.""" + return ctx.resource.get("url_type") == "datastore" + + +# --------------------------------------------------------------------------- +# Datastore rollback +# --------------------------------------------------------------------------- + + +def rollback_datastore_writes(runtime: RuntimeContext) -> None: + """Drop the datastore table after a failed write group, restoring the + stashed Data Dictionary if the analysis stage saved one. + + The Prefect flow calls this from ``database_task.on_rollback`` when + its transaction unwinds; the local runner calls it directly when a + stage after the database stage raises. Both mean the same thing: the + datastore table this run built is not trustworthy. + + The database stage's path is: delete any pre-existing table, create + an empty one, then COPY into it. So by the time a later stage fails, + the original content is already gone in *both* the "created from + empty" and "had pre-existing content" cases — what is on disk is a + half-written *new* table, not recoverable original data. Dropping it + unconditionally is strictly better than leaving polluted contents an + operator may not notice. + """ + resource_id = runtime.resource_id + try: + dsu.delete_datastore_resource(resource_id) + runtime.logger.info( + f"Rollback: dropped datastore resource {resource_id} " + "after transactional failure" + ) + except Exception as e: + runtime.logger.warning( + f"Rollback: could not drop datastore {resource_id}: {e}" + ) + + # Issue #265: if the analysis stage stashed a Data Dictionary + # before the original delete, restore it now by re-creating the + # datastore resource with the stashed per-field ``info`` dicts and + # zero rows. The *data* is unrecoverable (it never landed), but the + # operator's annotations (labels, descriptions, type_overrides) + # are preserved across the failed run. The stash file is left in + # place for inspection if restore itself fails — a future + # successful run will overwrite it. + stashed = dict_stash.load(resource_id) + if not stashed: + return + try: + # Derive each field's Postgres ``type`` from the stashed + # ``info["type_override"]`` (mapped through ``conf.TYPE_MAPPING`` + # values, e.g. ``numeric`` / ``timestamp`` / ``text``). This + # mirrors the analysis stage's ``_build_headers_dicts`` merge: + # otherwise CKAN's ``datastore_create`` falls back to ``text`` + # for every column, and a column the operator originally + # annotated as numeric or timestamp would be restored as text — + # silently inconsistent with the stashed dictionary's intent. + valid_types = set(conf.TYPE_MAPPING.values()) + fields = [] + for fid, info in stashed.items(): + field: Dict[str, Any] = {"id": fid, "info": info} + type_override = (info or {}).get("type_override") + if type_override in valid_types: + field["type"] = type_override + else: + field["type"] = "text" + fields.append(field) + dsu.send_resource_to_datastore( + resource=None, + resource_id=resource_id, + headers=fields, + records=[], + aliases=[], + calculate_record_count=False, + ) + runtime.logger.info( + f"Rollback: restored Data Dictionary for {resource_id} " + f"({len(fields)} field(s)) from stash" + ) + dict_stash.clear(resource_id) + except Exception as e: + runtime.logger.warning( + f"Rollback: could not restore Data Dictionary for " + f"{resource_id}: {e}. Stash file retained at " + f"{dict_stash.stash_path(resource_id)} for inspection." + ) diff --git a/ckanext/datapusher_plus/jobs/prefect_flow.py b/ckanext/datapusher_plus/jobs/prefect_flow.py index 098b80a2..e8d3fc49 100644 --- a/ckanext/datapusher_plus/jobs/prefect_flow.py +++ b/ckanext/datapusher_plus/jobs/prefect_flow.py @@ -26,7 +26,6 @@ from __future__ import annotations -import json import logging import os import sys @@ -34,7 +33,7 @@ import time import traceback from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Optional # --------------------------------------------------------------------------- @@ -172,14 +171,12 @@ def _bootstrap_ckan_app_context() -> None: # --------------------------------------------------------------------------- -import requests import sqlalchemy as sa from prefect import flow, task from prefect.logging import get_run_logger from prefect.transactions import transaction import ckanext.datapusher_plus.config as conf -import ckanext.datapusher_plus.datastore_utils as dsu import ckanext.datapusher_plus.dictionary_stash as dict_stash import ckanext.datapusher_plus.helpers as dph import ckanext.datapusher_plus.job_exceptions as job_exceptions @@ -194,7 +191,21 @@ def _bootstrap_ckan_app_context() -> None: DOWNLOAD_CACHE_POLICY, ) from ckanext.datapusher_plus.jobs.file_persistence import persist_file -from ckanext.datapusher_plus.jobs.context import ProcessingContext +# The Prefect-free half of the pipeline. Shared verbatim with +# ``jobs/local_runner.py`` (the ``prefect_enabled = false`` path), which +# is why it lives in its own module: nothing there may import Prefect. +# Bound to the historical private names so this module's call sites — and +# the tests that monkeypatch them — stay put. +from ckanext.datapusher_plus.jobs.pipeline_core import ( + StageAbort as _StageAbort, + build_runtime_context as _build_runtime_context, + callback_datapusher_hook, + resolve_int as _resolve_int_from_env, + resource_is_datastore_dump as _resource_is_datastore_dump, + rollback_datastore_writes, + run_stage as _stage_run, + validate_input as _validate_input, +) from ckanext.datapusher_plus.jobs.runtime_context import ( AnalyzeResult, ConvertResult, @@ -220,8 +231,6 @@ def _bootstrap_ckan_app_context() -> None: from ckanext.datapusher_plus.jobs.stages.indexing import IndexingStage from ckanext.datapusher_plus.jobs.stages.metadata import MetadataStage from ckanext.datapusher_plus.jobs.stages.validation import ValidationStage -from ckanext.datapusher_plus.logging_utils import TRACE -from ckanext.datapusher_plus.qsv_utils import QSVCommand # --------------------------------------------------------------------------- @@ -258,12 +267,9 @@ def _resolve_int( that Variable wins if set — operators can tune the value from the Prefect UI without shell access to the worker. Lookup failures (server unreachable, name absent, value not int-parseable) silently - fall through to the env / ckan.ini path. - - Env wins next so operators and CI can override per-process without - touching ``ckan.ini``. When the env var is unset, fall back to the - CKAN config key (useful for operators who manage all settings via - ``ckan.ini``). When nothing is set, return ``default``. + fall through to the env / ckan.ini path handled by + ``pipeline_core.resolve_int`` (which the local runner uses directly, + there being no Prefect server to hold Variables). Intentionally NOT a module-import-time call: ``Variable.get()`` triggers a Prefect API call (and an ephemeral server bootstrap when @@ -284,23 +290,7 @@ def _resolve_int( # older Prefect 3.x, Variable name has unexpected type, or # value isn't int-parseable — fall through to env / config. pass - env_value = os.environ.get(env_name) - if env_value is not None and env_value != "": - try: - return int(env_value) - except ValueError: - pass - try: - import ckan.plugins.toolkit as tk - - v = tk.config.get(config_key) - if v is not None and v != "": - return int(v) - except Exception: - # CKAN config not loaded (e.g., when running in a bare - # ``prefect worker`` process) — fall through to the default. - pass - return default + return _resolve_int_from_env(env_name, config_key, default) @@ -364,37 +354,6 @@ def _persist_stage_file( ) -# --------------------------------------------------------------------------- -# Callback helper (moved from pipeline.py) -# --------------------------------------------------------------------------- - - -def callback_datapusher_hook(result_url: str, job_dict: Dict[str, Any]) -> bool: - """ - POST a status update to CKAN's ``datapusher_hook`` endpoint. - - Preserves the v2 contract: the worker reports running/complete/error - state by POSTing here, which drives default-view creation, plugin - ``IDataPusher.after_upload`` hooks, and auto-resubmit on file change. - """ - api_token = utils.get_dp_plus_user_apitoken() - headers = { - "Content-Type": "application/json", - "Authorization": api_token, - } - try: - response = requests.post( - result_url, - data=json.dumps(job_dict, cls=utils.DatetimeJsonEncoder), - verify=conf.SSL_VERIFY, - headers=headers, - timeout=30, - ) - except requests.ConnectionError: - return False - return response.status_code == requests.codes.ok - - # --------------------------------------------------------------------------- # Stage tasks # --------------------------------------------------------------------------- @@ -411,49 +370,6 @@ def callback_datapusher_hook(result_url: str, job_dict: Dict[str, Any]) -> bool: # retry would fail identically. -class _StageAbort(Exception): - """A stage returned ``None`` — the BaseStage "nothing to do" signal. - - Per the ``BaseStage`` contract, ``process()`` may return ``None`` to - stop the rest of the pipeline gracefully (e.g. the Analysis stage on - a zero-record file logs "Upload skipped as there are zero records" - and returns ``None``). The v2 pipeline stopped there and the job - *completed* — nothing was wrong, there was simply nothing to load. - - Raised by ``_stage_run`` and caught distinctly from ``JobError`` in - the flow so the job is marked complete-with-skip, not errored. - """ - - def __init__(self, stage_name: str): - self.stage_name = stage_name - super().__init__( - f"Stage {stage_name} stopped the pipeline (nothing to do)" - ) - - -def _stage_run(stage, prev: Any = None) -> RuntimeContext: - """Invoke a stage on the bound RuntimeContext. - - ``prev`` is the upstream task's result. When given, the bound - ``RuntimeContext`` is rehydrated from it first, so the stage sees - correct ``ctx`` state even if the upstream task's body never ran (a - Prefect cache hit, or a persisted-result replay on a flow re-run) — - that body is what would otherwise have mutated the shared context. - The root task (``download_task``) passes no ``prev``. - - A stage returning ``None`` is the BaseStage "skip / nothing to do" - signal (per its docstring) — surfaced here as ``_StageAbort`` so the - flow can stop cleanly and mark the job *complete*, not errored. - """ - ctx = get_runtime_context() - if prev is not None: - rehydrate(ctx, prev) - result = stage(ctx) - if result is None: - raise _StageAbort(stage.name) - return result - - # Both JobError hierarchies in the codebase (``utils.JobError`` and # ``job_exceptions.JobError``, the latter the parent of ``HTTPError`` / # ``LoaderError``) represent deterministic data failures — a malformed @@ -799,80 +715,15 @@ def _runtime_or_none() -> Optional[RuntimeContext]: def _rollback_database(txn) -> None: """Drop the datastore table on transactional failure. - The database stage's path is: delete any pre-existing table, create - an empty one, then COPY into it. So by the time a later task in the - transaction fails, the original content is already gone in *both* - the "created from empty" and "had pre-existing content" cases — what - is on disk is a half-written *new* table, not recoverable original - data. Dropping it unconditionally is strictly better than leaving - polluted contents an operator may not notice. (The earlier - ``existing_info`` branch claimed to "preserve" the original, but the - delete had already destroyed it.) + The work — dropping the half-written table and restoring the stashed + Data Dictionary — lives in ``pipeline_core.rollback_datastore_writes`` + so the local (no-Prefect) runner performs exactly the same cleanup + when a stage after the database load fails. """ runtime = _runtime_or_none() if runtime is None: return - resource_id = runtime.resource_id - try: - dsu.delete_datastore_resource(resource_id) - runtime.logger.info( - f"Rollback: dropped datastore resource {resource_id} " - "after transactional failure" - ) - except Exception as e: - runtime.logger.warning( - f"Rollback: could not drop datastore {resource_id}: {e}" - ) - - # Issue #265: if the analysis stage stashed a Data Dictionary - # before the original delete, restore it now by re-creating the - # datastore resource with the stashed per-field ``info`` dicts and - # zero rows. The *data* is unrecoverable (it never landed), but the - # operator's annotations (labels, descriptions, type_overrides) - # are preserved across the failed run. The stash file is left in - # place for inspection if restore itself fails — a future - # successful run will overwrite it. - stashed = dict_stash.load(resource_id) - if not stashed: - return - try: - # Derive each field's Postgres ``type`` from the stashed - # ``info["type_override"]`` (mapped through ``conf.TYPE_MAPPING`` - # values, e.g. ``numeric`` / ``timestamp`` / ``text``). This - # mirrors the analysis stage's ``_build_headers_dicts`` merge: - # otherwise CKAN's ``datastore_create`` falls back to ``text`` - # for every column, and a column the operator originally - # annotated as numeric or timestamp would be restored as text — - # silently inconsistent with the stashed dictionary's intent. - valid_types = set(conf.TYPE_MAPPING.values()) - fields = [] - for fid, info in stashed.items(): - field: Dict[str, Any] = {"id": fid, "info": info} - type_override = (info or {}).get("type_override") - if type_override in valid_types: - field["type"] = type_override - else: - field["type"] = "text" - fields.append(field) - dsu.send_resource_to_datastore( - resource=None, - resource_id=resource_id, - headers=fields, - records=[], - aliases=[], - calculate_record_count=False, - ) - runtime.logger.info( - f"Rollback: restored Data Dictionary for {resource_id} " - f"({len(fields)} field(s)) from stash" - ) - dict_stash.clear(resource_id) - except Exception as e: - runtime.logger.warning( - f"Rollback: could not restore Data Dictionary for " - f"{resource_id}: {e}. Stash file retained at " - f"{dict_stash.stash_path(resource_id)} for inspection." - ) + rollback_datastore_writes(runtime) @indexing_task.on_rollback @@ -914,81 +765,6 @@ def _rollback_metadata(txn) -> None: ) -# --------------------------------------------------------------------------- -# Pre-flight helpers -# --------------------------------------------------------------------------- - - -def _validate_input(input_payload: Dict[str, Any]) -> None: - """Mirror of v2 ``pipeline.validate_input``.""" - if "metadata" not in input_payload: - raise utils.JobError("Metadata missing") - if "resource_id" not in input_payload["metadata"]: - raise utils.JobError("No id provided.") - - -def _build_runtime_context( - job_input: JobInput, temp_dir: str -) -> RuntimeContext: - """ - Construct the per-run ``RuntimeContext`` (== legacy ``ProcessingContext``). - - Sets up the task-scoped logger with both the v2 ``StoringHandler`` (so - the DP+ ``Logs`` table continues to populate, and the CKAN UI's job - detail view keeps working) and a stream handler for the worker's - stdout. - """ - task_id = job_input.task_id - input_payload = job_input.input - - # Task-scoped logger — same approach as v2 ``_push_to_datastore``. - handler = utils.StoringHandler(task_id, input_payload) - logger = logging.getLogger(task_id) - logger.addHandler(handler) - logger.addHandler(logging.StreamHandler()) - try: - log_level = getattr(logging, conf.UPLOAD_LOG_LEVEL.upper()) - except AttributeError: - log_level = TRACE - logger.setLevel(log_level) - logger.info(f"Setting log level to {logging.getLevelName(int(log_level))}") - - if not Path(conf.QSV_BIN).is_file(): - raise utils.JobError(f"{conf.QSV_BIN} not found.") - - qsv = QSVCommand(logger=logger) - - # Fetch the resource (one retry, as in v2). - resource_id = job_input.resource_id - try: - resource = dsu.get_resource(resource_id) - except utils.JobError: - time.sleep(5) - resource = dsu.get_resource(resource_id) - - ctx = ProcessingContext( - task_id=task_id, - input=input_payload, - dry_run=job_input.dry_run, - temp_dir=temp_dir, - logger=logger, - qsv=qsv, - resource=resource, - resource_id=resource_id, - ckan_url=job_input.ckan_url, - # Stamp now so the duration-since-start computed in the success - # event (``time.time() - timer_start``) is meaningful. - timer_start=time.time(), - ) - return ctx - - -def _resource_is_datastore_dump(ctx: RuntimeContext) -> bool: - """v2 early-exit: ``url_type == 'datastore'`` resources are not re-ingested.""" - return ctx.resource.get("url_type") == "datastore" - - - # --------------------------------------------------------------------------- # PII review suspension # --------------------------------------------------------------------------- diff --git a/ckanext/datapusher_plus/logic/action.py b/ckanext/datapusher_plus/logic/action.py index 0ccbca9f..7e6dd80d 100644 --- a/ckanext/datapusher_plus/logic/action.py +++ b/ckanext/datapusher_plus/logic/action.py @@ -16,6 +16,7 @@ import ckan.logic as logic import ckan.plugins as p from ckan.common import config +import ckanext.datapusher_plus.config as dpp_config import ckanext.datapusher_plus.logic.schema as dpschema import ckanext.datapusher_plus.interfaces as interfaces import ckanext.datapusher_plus.prefect_client as prefect_client @@ -37,6 +38,46 @@ from typing import Any, cast +def _orchestrator(): + """Return the module that runs ingestion jobs for this deployment. + + ``prefect_client`` by default; ``jobs.local_runner`` when + ``ckanext.datapusher_plus.prefect_enabled`` is false. Both expose + ``get_running_resource_ids()``, and both have a submit entry point + (``submit_flow_run`` / ``enqueue_job``) — see ``_submit_job`` for the + one place the two signatures differ. + + ``local_runner`` is imported lazily (it pulls in every stage class), + and ``prefect_client`` never imports ``prefect`` at module level — + so a CKAN process with Prefect turned off gets through this without + touching Prefect at all. That matters on a host where + ``$PREFECT_HOME`` is not writable: there, the import itself raises. + """ + if dpp_config.prefect_enabled(): + return prefect_client + + import ckanext.datapusher_plus.jobs.local_runner as local_runner + + return local_runner + + +def _submit_job(job_input: JobInput, timeout: int) -> tuple[str, bool]: + """Hand a job to the active orchestrator. + + Returns ``(run_id, via_prefect)`` — the caller records the id on the + CKAN ``task_status`` row under the key that matches the backend, so + ``datapusher_status`` only ever builds a Prefect UI deep-link for a + run that really is a Prefect flow run. + """ + orchestrator = _orchestrator() + if orchestrator is not prefect_client: + return orchestrator.enqueue_job(job_input, timeout=timeout), False + return ( + orchestrator.submit_flow_run(asdict(job_input), timeout=timeout), + True, + ) + + def datapusher_submit(context, data_dict: dict[str, Any]): """Submit a job to the datapusher. The datapusher is a service that imports tabular data into the datastore. @@ -124,9 +165,11 @@ def datapusher_submit(context, data_dict: dict[str, Any]): seconds=tk.asint(config.get("ckan.datapusher.assume_task_stillborn_after", 5)) ) if existing_task.get("state") == "pending": - # Query Prefect for resource_ids currently in non-terminal flow - # runs. Replaces the v2 RQ-queue regex scan. - queued_res_ids = prefect_client.get_running_resource_ids() + # Ask the active orchestrator which resource_ids are still + # in flight: Prefect's non-terminal flow runs, or — with + # Prefect disabled — CKAN's RQ queue. Replaces the v2 + # RQ-queue regex scan. + queued_res_ids = _orchestrator().get_running_resource_ids() # Symmetric with the write sites that serialize via # ``utcnow_naive().isoformat()`` — ``fromisoformat`` round-trips # the same value cleanly and handles the missing-microseconds @@ -223,17 +266,19 @@ def datapusher_submit(context, data_dict: dict[str, Any]): dry_run=False, ) try: - flow_run_id = prefect_client.submit_flow_run( - asdict(job_input), timeout=dp_timeout - ) + run_id, via_prefect = _submit_job(job_input, dp_timeout) except Exception as e: log.error("Error submitting job to DataPusher: %s", e) return False # Public contract: ``job_id`` keeps working for v2 consumers (CKAN UI's # status page reads it). ``flow_run_id`` is the new field for clients - # that want to deep-link into the Prefect UI. - value = json.dumps({"job_id": job_id, "flow_run_id": flow_run_id}) + # that want to deep-link into the Prefect UI — recorded only when the + # run really is a Prefect flow run. With Prefect disabled the RQ job + # id goes in ``rq_job_id`` instead, so nothing tries to build a + # Prefect URL out of it. + run_id_key = "flow_run_id" if via_prefect else "rq_job_id" + value = json.dumps({"job_id": job_id, run_id_key: run_id}) task["value"] = value task["state"] = "pending" task["last_updated"] = utcnow_naive().isoformat() diff --git a/ckanext/datapusher_plus/plugin.py b/ckanext/datapusher_plus/plugin.py index 268c6e5b..7bbf583a 100644 --- a/ckanext/datapusher_plus/plugin.py +++ b/ckanext/datapusher_plus/plugin.py @@ -77,6 +77,19 @@ def configure(self, config): if self.enable_druf: log.info("DRUF functionality enabled for DataPusher Plus") + # Which orchestrator will run ingestions. Logged at INFO when + # Prefect is off so operators can confirm the mode (and the + # worker they need) from the CKAN startup log rather than by + # reading ckan.ini. + import ckanext.datapusher_plus.config as conf + + if not conf.prefect_enabled(): + log.info( + "Prefect orchestration disabled for DataPusher Plus — jobs " + "will run in-process on CKAN's background worker " + "(`ckan jobs worker`)" + ) + def update_config(self, config: CKANConfig): # Always add base templates tk.add_template_directory(config, "templates") diff --git a/tests/test_dictionary_stash.py b/tests/test_dictionary_stash.py index 438e1765..ce06eac9 100644 --- a/tests/test_dictionary_stash.py +++ b/tests/test_dictionary_stash.py @@ -439,7 +439,7 @@ def test_rollback_restore_derives_field_type_from_type_override( from types import SimpleNamespace from unittest import mock - from ckanext.datapusher_plus.jobs import prefect_flow + from ckanext.datapusher_plus.jobs import pipeline_core, prefect_flow resource_id = "res-rollback-types" stashed = { @@ -456,13 +456,13 @@ def test_rollback_restore_derives_field_type_from_type_override( runtime = SimpleNamespace(resource_id=resource_id, logger=mock.Mock()) monkeypatch.setattr(prefect_flow, "_runtime_or_none", lambda: runtime) monkeypatch.setattr( - prefect_flow.dsu, "delete_datastore_resource", lambda rid: None + pipeline_core.dsu, "delete_datastore_resource", lambda rid: None ) captured = {} def _capture(**kwargs): captured.update(kwargs) return {} - monkeypatch.setattr(prefect_flow.dsu, "send_resource_to_datastore", _capture) + monkeypatch.setattr(pipeline_core.dsu, "send_resource_to_datastore", _capture) prefect_flow._rollback_database(txn=None) @@ -488,7 +488,7 @@ def test_rollback_restore_ignores_unknown_type_override( from types import SimpleNamespace from unittest import mock - from ckanext.datapusher_plus.jobs import prefect_flow + from ckanext.datapusher_plus.jobs import pipeline_core, prefect_flow resource_id = "res-rollback-unknown-type" stash_module.save( @@ -499,11 +499,11 @@ def test_rollback_restore_ignores_unknown_type_override( runtime = SimpleNamespace(resource_id=resource_id, logger=mock.Mock()) monkeypatch.setattr(prefect_flow, "_runtime_or_none", lambda: runtime) monkeypatch.setattr( - prefect_flow.dsu, "delete_datastore_resource", lambda rid: None + pipeline_core.dsu, "delete_datastore_resource", lambda rid: None ) captured = {} monkeypatch.setattr( - prefect_flow.dsu, + pipeline_core.dsu, "send_resource_to_datastore", lambda **kwargs: captured.update(kwargs) or {}, ) diff --git a/tests/test_local_runner.py b/tests/test_local_runner.py new file mode 100644 index 00000000..31f53ad8 --- /dev/null +++ b/tests/test_local_runner.py @@ -0,0 +1,584 @@ +# -*- coding: utf-8 -*- +""" +Unit-level coverage for the no-Prefect (``prefect_enabled = false``) path. + +Three things are under test here: + +* ``config.prefect_enabled`` — the switch itself. +* ``jobs/local_runner.py`` — the in-process runner: stage order, the + Jobs-row state machine, complete-with-skip, datastore rollback, the + PII-review guard, and the RQ helpers (``enqueue_job`` / + ``get_running_resource_ids``). +* ``logic/action.datapusher_submit`` routing — that flipping the flag + moves submissions from Prefect to RQ, and that the task_status row + records the run id under a key matching the backend. + +Every test runs without a Prefect server, an RQ worker, CKAN, or +Postgres: stage classes are patched to no-ops that mutate the +``ProcessingContext`` the way the real stages would. + +The critical invariant — that this path never imports Prefect — is +asserted directly in ``test_local_runner_does_not_import_prefect``. +""" + +from __future__ import annotations + +import logging +import sys +from contextlib import ExitStack +from types import SimpleNamespace +from unittest import mock + +import pytest + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def job_input(): + from ckanext.datapusher_plus.jobs.runtime_context import JobInput + + return JobInput( + task_id="test-task-local-1", + resource_id="resource-abc", + ckan_url="http://ckan.test", + input={ + "api_key": "test-token", + "job_type": "push_to_datastore", + "result_url": "http://ckan.test/api/3/action/datapusher_hook", + "metadata": { + "resource_id": "resource-abc", + "ckan_url": "http://ckan.test", + "ignore_hash": False, + }, + }, + dry_run=True, + ) + + +STAGE_NAMES = [ + "DownloadStage", + "FormatConverterStage", + "ValidationStage", + "AnalysisStage", + "AISuggestionsStage", + "DatabaseStage", + "IndexingStage", + "FormulaStage", + "MetadataStage", +] + + +@pytest.fixture +def patched_runner(): + """Patch every external integration so ``run_job`` runs in-process. + + Yields a dict with the stage mocks (keyed by class name), the + ``mark_job_as_*`` mocks, the callback mock, and ``calls`` — the + ordered list of stage names as they execute, which is what the + stage-ordering assertions read. + """ + from ckanext.datapusher_plus.jobs import local_runner + + calls: list[str] = [] + stages = {} + + with ExitStack() as stack: + for name in STAGE_NAMES: + stage_instance = mock.MagicMock() + stage_instance.name = name + + def _run(ctx, _name=name): + calls.append(_name) + return ctx + + stage_instance.side_effect = _run + stack.enter_context( + mock.patch.object( + local_runner, name, return_value=stage_instance + ) + ) + stages[name] = stage_instance + + stack.enter_context( + mock.patch( + "ckanext.datapusher_plus.jobs.pipeline_core.dsu.get_resource", + return_value={ + "url_type": "upload", + "format": "CSV", + "url": "x.csv", + }, + ) + ) + stack.enter_context( + mock.patch("ckanext.datapusher_plus.jobs.pipeline_core.QSVCommand") + ) + stack.enter_context( + mock.patch( + "ckanext.datapusher_plus.jobs.pipeline_core.Path.is_file", + return_value=True, + ) + ) + stack.enter_context( + mock.patch( + "ckanext.datapusher_plus.jobs.pipeline_core.utils.StoringHandler", + return_value=logging.NullHandler(), + ) + ) + callback = stack.enter_context( + mock.patch.object( + local_runner, "callback_datapusher_hook", return_value=True + ) + ) + stack.enter_context(mock.patch.object(local_runner.dph, "add_pending_job")) + stack.enter_context(mock.patch.object(local_runner.dph, "set_aps_job_id")) + mark_completed = stack.enter_context( + mock.patch.object(local_runner.dph, "mark_job_as_completed") + ) + mark_errored = stack.enter_context( + mock.patch.object(local_runner.dph, "mark_job_as_errored") + ) + stack.enter_context( + mock.patch.object(local_runner.dph, "mark_job_as_failed_to_post_result") + ) + stack.enter_context(mock.patch.object(local_runner.dict_stash, "clear")) + + yield { + "stages": stages, + "calls": calls, + "callback": callback, + "mark_completed": mark_completed, + "mark_errored": mark_errored, + } + + +# --------------------------------------------------------------------------- +# The switch +# --------------------------------------------------------------------------- + + +def test_prefect_enabled_defaults_to_true(monkeypatch): + import ckanext.datapusher_plus.config as conf + + monkeypatch.setitem(conf.tk.config, "ckanext.datapusher_plus.prefect_enabled", "") + assert conf.prefect_enabled() is True + + +@pytest.mark.parametrize( + "value,expected", + [("false", False), ("False", False), ("no", False), (False, False), + ("true", True), (True, True), ("1", True)], +) +def test_prefect_enabled_reads_config(monkeypatch, value, expected): + import ckanext.datapusher_plus.config as conf + + monkeypatch.setitem( + conf.tk.config, "ckanext.datapusher_plus.prefect_enabled", value + ) + assert conf.prefect_enabled() is expected + + +# --------------------------------------------------------------------------- +# The no-Prefect invariant +# --------------------------------------------------------------------------- + + +def test_local_runner_does_not_import_prefect(): + """Importing the local path must not pull Prefect in. + + This is the whole point of the mode: on a host whose + ``$PREFECT_HOME`` is unwritable, ``import prefect`` itself raises + ``PermissionError: '/root/.prefect/profiles.toml'``. Anything the + disabled path imports has to stay clear of it. + + Runs in a subprocess because it needs a pristine ``sys.modules``: + other tests in this session have Prefect imported already, and + tearing it back out of ``sys.modules`` in-process would hand every + later Prefect test a second, mismatched copy of the library. + """ + import subprocess + + script = ( + "import sys\n" + "import ckanext.datapusher_plus.jobs.pipeline_core\n" + "import ckanext.datapusher_plus.jobs.local_runner\n" + "leaked = sorted(n for n in sys.modules " + "if n == 'prefect' or n.startswith('prefect.'))\n" + "print(','.join(leaked))\n" + ) + result = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + check=True, + ) + assert result.stdout.strip() == "", ( + f"the no-Prefect path imported Prefect: {result.stdout.strip()}" + ) + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + + +def test_run_job_runs_all_stages_in_order_and_completes(job_input, patched_runner): + from ckanext.datapusher_plus.jobs.local_runner import run_job + + result = run_job(job_input) + + assert result is None # same contract as the flow + assert patched_runner["calls"] == STAGE_NAMES + patched_runner["mark_completed"].assert_called_once() + patched_runner["mark_errored"].assert_not_called() + + +def test_run_job_accepts_the_enqueued_dict_payload(job_input, patched_runner): + """RQ hands back the ``asdict``-ed payload, not the dataclass.""" + from dataclasses import asdict + + from ckanext.datapusher_plus.jobs.local_runner import run_job + + assert run_job(asdict(job_input)) is None + patched_runner["mark_completed"].assert_called_once() + + +def test_run_job_posts_running_then_complete_callbacks(job_input, patched_runner): + from ckanext.datapusher_plus.jobs.local_runner import run_job + + run_job(job_input) + + statuses = [ + call.kwargs["job_dict"]["status"] + for call in patched_runner["callback"].call_args_list + ] + assert statuses == ["running", "complete"] + + +# --------------------------------------------------------------------------- +# Failure propagation +# --------------------------------------------------------------------------- + + +def test_run_job_marks_errored_and_posts_error_callback(job_input, patched_runner): + from ckanext.datapusher_plus import utils + from ckanext.datapusher_plus.jobs.local_runner import run_job + + patched_runner["stages"]["AnalysisStage"].side_effect = utils.JobError( + "fake analysis failure" + ) + + with pytest.raises(utils.JobError): + run_job(job_input) + + patched_runner["mark_errored"].assert_called_once() + args, _ = patched_runner["mark_errored"].call_args + assert "fake analysis failure" in args[1] + + statuses = [ + call.kwargs["job_dict"]["status"] + for call in patched_runner["callback"].call_args_list + ] + assert statuses == ["running", "error"] + + +def test_stage_returning_none_completes_with_skip(job_input, patched_runner): + """A stage returning ``None`` is "nothing to do", not a failure.""" + from ckanext.datapusher_plus.jobs.local_runner import run_job + + patched_runner["stages"]["AnalysisStage"].side_effect = lambda ctx: None + + assert run_job(job_input) is None + + patched_runner["mark_errored"].assert_not_called() + patched_runner["mark_completed"].assert_called_once() + _, kwargs = patched_runner["mark_completed"].call_args + args, _ = patched_runner["mark_completed"].call_args + assert args[1] == {"skipped": "AnalysisStage"} + # Stages after the abort must not have run. + assert "DatabaseStage" not in patched_runner["calls"] + + +def test_datastore_dump_resource_is_skipped(job_input, patched_runner): + """``url_type == 'datastore'`` completes without running any stage.""" + from ckanext.datapusher_plus.jobs.local_runner import run_job + + with mock.patch( + "ckanext.datapusher_plus.jobs.pipeline_core.dsu.get_resource", + return_value={"url_type": "datastore"}, + ): + assert run_job(job_input) is None + + assert patched_runner["calls"] == [] + args, _ = patched_runner["mark_completed"].call_args + assert args[1] == {"skipped": "datastore-managed"} + + +# --------------------------------------------------------------------------- +# Rollback +# --------------------------------------------------------------------------- + + +def test_failure_after_database_stage_drops_the_datastore_table( + job_input, patched_runner +): + """Indexing blowing up must clean up the half-built table. + + Mirrors what ``database_task.on_rollback`` does under Prefect's + ``transaction()``. + """ + from ckanext.datapusher_plus import utils + from ckanext.datapusher_plus.jobs import pipeline_core + from ckanext.datapusher_plus.jobs.local_runner import run_job + + patched_runner["stages"]["IndexingStage"].side_effect = utils.JobError( + "fake indexing failure" + ) + + with mock.patch.object( + pipeline_core.dsu, "delete_datastore_resource" + ) as delete_ds, mock.patch.object( + pipeline_core.dict_stash, "load", return_value=None + ): + with pytest.raises(utils.JobError): + run_job(job_input) + + delete_ds.assert_called_once_with(job_input.resource_id) + + +def test_database_stage_failure_leaves_the_datastore_alone( + job_input, patched_runner +): + """A database-stage failure must NOT drop the table. + + The stage raises before it is known to have touched anything — "could + not connect to the Datastore" is the common case — so dropping here + would destroy data this run never wrote. + """ + from ckanext.datapusher_plus import utils + from ckanext.datapusher_plus.jobs import pipeline_core + from ckanext.datapusher_plus.jobs.local_runner import run_job + + patched_runner["stages"]["DatabaseStage"].side_effect = utils.JobError( + "Could not connect to the Datastore" + ) + + with mock.patch.object( + pipeline_core.dsu, "delete_datastore_resource" + ) as delete_ds: + with pytest.raises(utils.JobError): + run_job(job_input) + + delete_ds.assert_not_called() + + +# --------------------------------------------------------------------------- +# PII review guard +# --------------------------------------------------------------------------- + + +def test_pii_review_gate_aborts_before_datastore_writes(job_input, patched_runner): + """With no Prefect there is nothing to suspend on, so a flagged run + must stop before the database stage rather than load unreviewed PII.""" + from ckanext.datapusher_plus import utils + from ckanext.datapusher_plus.jobs import local_runner + from ckanext.datapusher_plus.jobs.local_runner import run_job + + def _flag_pii(ctx): + patched_runner["calls"].append("AnalysisStage") + ctx.pii_found = True + ctx.pii_candidate_count = 5 + return ctx + + patched_runner["stages"]["AnalysisStage"].side_effect = _flag_pii + + with mock.patch.object(local_runner, "resolve_int") as resolve_int: + # ``resolve_int`` also resolves the job timeout — only the PII + # threshold should be non-zero here. + resolve_int.side_effect = lambda env, key, default: ( + 1 if "pii_review_threshold" in key else default + ) + with pytest.raises(utils.JobError, match="PII review gate crossed"): + run_job(job_input) + + assert "DatabaseStage" not in patched_runner["calls"] + patched_runner["mark_errored"].assert_called_once() + + +def test_pii_review_threshold_zero_does_not_gate(job_input, patched_runner): + """The feature is off by default: PII alone must not stop a job.""" + from ckanext.datapusher_plus.jobs.local_runner import run_job + + def _flag_pii(ctx): + patched_runner["calls"].append("AnalysisStage") + ctx.pii_found = True + ctx.pii_candidate_count = 5 + return ctx + + patched_runner["stages"]["AnalysisStage"].side_effect = _flag_pii + + assert run_job(job_input) is None + assert "DatabaseStage" in patched_runner["calls"] + + +# --------------------------------------------------------------------------- +# RQ helpers +# --------------------------------------------------------------------------- + + +def test_enqueue_job_hands_the_payload_and_timeout_to_rq(job_input): + from dataclasses import asdict + + from ckanext.datapusher_plus.jobs import local_runner + + with mock.patch.object( + local_runner.tk, "enqueue_job", return_value=SimpleNamespace(id="rq-1") + ) as enqueue: + assert local_runner.enqueue_job(job_input, timeout=900) == "rq-1" + + args, kwargs = enqueue.call_args + assert args[0] is local_runner.run_job + assert args[1] == [asdict(job_input)] + assert kwargs["rq_kwargs"] == {"timeout": 900} + + +def test_get_running_resource_ids_reads_queued_and_started_jobs(): + from ckanext.datapusher_plus.jobs import local_runner + + queued = SimpleNamespace(args=[{"resource_id": "res-queued"}]) + started = SimpleNamespace(args=[{"resource_id": "res-started"}]) + unrelated = SimpleNamespace(args=["not-a-payload"]) + + queue = mock.MagicMock() + queue.get_jobs.return_value = [queued, unrelated] + queue.fetch_job.return_value = started + + fake_registry = mock.MagicMock() + fake_registry.get_job_ids.return_value = ["started-id"] + + with mock.patch("ckan.lib.jobs.get_queue", return_value=queue), \ + mock.patch("rq.registry.StartedJobRegistry", return_value=fake_registry): + ids = local_runner.get_running_resource_ids() + + assert ids == {"res-queued", "res-started"} + + +def test_get_running_resource_ids_fails_open(): + """An unreachable Redis must not block submissions.""" + from ckanext.datapusher_plus.jobs import local_runner + + with mock.patch( + "ckan.lib.jobs.get_queue", side_effect=RuntimeError("redis down") + ): + assert local_runner.get_running_resource_ids() == set() + + +# --------------------------------------------------------------------------- +# datapusher_submit routing +# --------------------------------------------------------------------------- + + +@pytest.fixture +def prefect_off(monkeypatch): + import ckanext.datapusher_plus.config as conf + + monkeypatch.setitem( + conf.tk.config, "ckanext.datapusher_plus.prefect_enabled", "false" + ) + + +def test_orchestrator_is_prefect_by_default(): + import ckanext.datapusher_plus.prefect_client as prefect_client + from ckanext.datapusher_plus.logic import action + + assert action._orchestrator() is prefect_client + + +def test_orchestrator_is_local_runner_when_disabled(prefect_off): + from ckanext.datapusher_plus.jobs import local_runner + from ckanext.datapusher_plus.logic import action + + assert action._orchestrator() is local_runner + + +def test_submit_job_goes_to_prefect_by_default(job_input): + from dataclasses import asdict + + from ckanext.datapusher_plus.logic import action + + with mock.patch.object( + action.prefect_client, "submit_flow_run", return_value="flow-1" + ) as submit: + run_id, via_prefect = action._submit_job(job_input, 900) + + assert (run_id, via_prefect) == ("flow-1", True) + submit.assert_called_once_with(asdict(job_input), timeout=900) + + +def test_submit_job_goes_to_rq_when_disabled(job_input, prefect_off): + from ckanext.datapusher_plus.jobs import local_runner + from ckanext.datapusher_plus.logic import action + + with mock.patch.object( + local_runner, "enqueue_job", return_value="rq-1" + ) as enqueue: + run_id, via_prefect = action._submit_job(job_input, 900) + + assert (run_id, via_prefect) == ("rq-1", False) + enqueue.assert_called_once_with(job_input, timeout=900) + + +def _run_datapusher_submit(monkeypatch, submit_return): + """Drive ``datapusher_submit`` with CKAN's I/O stubbed out. + + Returns the ``task_status_update`` dict the action wrote, which is + what records the orchestrator's run id. + """ + import json + + from ckanext.datapusher_plus.logic import action + + updates: list[dict] = [] + + def _fake_get_action(name): + if name == "resource_show": + return lambda ctx, data: {"id": data["id"], "package_id": "pkg-1"} + if name == "task_status_show": + def _raise(ctx, data): + raise action.tk.ObjectNotFound("no task") + + return _raise + if name == "task_status_update": + return lambda ctx, task: updates.append(task) or task + raise AssertionError(f"unexpected action: {name}") + + monkeypatch.setattr(action.p.toolkit, "get_action", _fake_get_action) + monkeypatch.setattr(action.p.toolkit, "check_access", lambda *a, **kw: True) + monkeypatch.setattr(action.h, "url_for", lambda *a, **kw: "http://ckan.test/") + monkeypatch.setattr(action.utils, "get_dp_plus_user_apitoken", lambda: "tok") + monkeypatch.setattr(action, "_submit_job", lambda job_input, t: submit_return) + + context = {"model": mock.MagicMock(), "user": "tester"} + assert action.datapusher_submit(context, {"resource_id": "resource-abc"}) is True + assert updates, "the action never wrote a task_status row" + return json.loads(updates[-1]["value"]) + + +def test_submit_records_flow_run_id_under_prefect(monkeypatch): + value = _run_datapusher_submit(monkeypatch, ("flow-1", True)) + + assert value["flow_run_id"] == "flow-1" + assert "rq_job_id" not in value + assert value["job_id"] + + +def test_submit_records_rq_job_id_when_prefect_disabled(monkeypatch): + """A Prefect-UI deep-link must never be built from an RQ job id.""" + value = _run_datapusher_submit(monkeypatch, ("rq-1", False)) + + assert value["rq_job_id"] == "rq-1" + assert "flow_run_id" not in value + assert value["job_id"] diff --git a/tests/test_prefect_flow.py b/tests/test_prefect_flow.py index 7bf90a46..19752131 100644 --- a/tests/test_prefect_flow.py +++ b/tests/test_prefect_flow.py @@ -95,7 +95,7 @@ def patched_dependencies(): return_value=mock.MagicMock(side_effect=lambda ctx: ctx), ), mock.patch( - "ckanext.datapusher_plus.jobs.prefect_flow.dsu.get_resource", + "ckanext.datapusher_plus.jobs.pipeline_core.dsu.get_resource", return_value={"url_type": "upload", "format": "CSV", "url": "x.csv"}, ), mock.patch( @@ -117,11 +117,14 @@ def patched_dependencies(): "ckanext.datapusher_plus.jobs.prefect_flow.utils.StoringHandler", return_value=logging.NullHandler(), ), + # ``build_runtime_context`` (and so the QSVCommand construction + # and the QSV_BIN existence check) lives in ``pipeline_core``, + # shared with the local runner — patch it there. mock.patch( - "ckanext.datapusher_plus.jobs.prefect_flow.QSVCommand" + "ckanext.datapusher_plus.jobs.pipeline_core.QSVCommand" ), mock.patch( - "ckanext.datapusher_plus.jobs.prefect_flow.Path.is_file", + "ckanext.datapusher_plus.jobs.pipeline_core.Path.is_file", return_value=True, ), mock.patch( @@ -205,7 +208,7 @@ def test_rollback_drops_datastore_when_indexing_fails(job_input, patched_depende from unittest import mock from ckanext.datapusher_plus import utils - from ckanext.datapusher_plus.jobs import prefect_flow + from ckanext.datapusher_plus.jobs import pipeline_core, prefect_flow # Make indexing raise *after* database has committed within the # transaction. @@ -222,7 +225,7 @@ def test_rollback_drops_datastore_when_indexing_fails(job_input, patched_depende ) with mock.patch.object( - prefect_flow.dsu, "delete_datastore_resource" + pipeline_core.dsu, "delete_datastore_resource" ) as delete_ds: with pytest.raises(utils.JobError): prefect_flow.datapusher_plus_flow(job_input_real) @@ -246,7 +249,7 @@ def test_pii_review_rejection_raises_before_database_writes( from unittest import mock from ckanext.datapusher_plus import utils - from ckanext.datapusher_plus.jobs import prefect_flow + from ckanext.datapusher_plus.jobs import pipeline_core, prefect_flow # Configure the AnalysisStage mock to report two PII candidate # matches on the ProcessingContext that the task wrapper reads. @@ -277,7 +280,7 @@ def _set_pii(ctx): ), mock.patch( "prefect.flow_runs.suspend_flow_run", return_value=rejection ), mock.patch.object( - prefect_flow.dsu, "delete_datastore_resource" + pipeline_core.dsu, "delete_datastore_resource" ) as delete_ds: with pytest.raises(utils.JobError, match="PII review rejected"): prefect_flow.datapusher_plus_flow(job_input_real) @@ -382,12 +385,12 @@ def test_flow_short_circuits_for_datastore_dumps(job_input): """``url_type == 'datastore'`` resources are completed without running stages.""" from contextlib import ExitStack - from ckanext.datapusher_plus.jobs import prefect_flow + from ckanext.datapusher_plus.jobs import pipeline_core, prefect_flow with ExitStack() as stack: stack.enter_context( mock.patch.object( - prefect_flow.dsu, + pipeline_core.dsu, "get_resource", return_value={"url_type": "datastore"}, ) @@ -405,10 +408,10 @@ def test_flow_short_circuits_for_datastore_dumps(job_input): ) ) stack.enter_context( - mock.patch.object(prefect_flow.QSVCommand, "__init__", return_value=None) + mock.patch.object(pipeline_core.QSVCommand, "__init__", return_value=None) ) stack.enter_context( - mock.patch.object(prefect_flow.Path, "is_file", return_value=True) + mock.patch.object(pipeline_core.Path, "is_file", return_value=True) ) stack.enter_context( mock.patch.object(