From 302d9a62d6b90f658c238cf570d488ca433c5dfd Mon Sep 17 00:00:00 2001 From: Casey de la Vega-Rawson Date: Fri, 28 Aug 2026 18:59:42 -0600 Subject: [PATCH 1/4] refactor: update deploy scripts for new naming --- deploy/00_config.sh | 4 ++-- deploy/30_dagster_gcp_auth.sh | 10 +++++----- deploy/31_verify_dagster_auth.py | 2 +- deploy/README.md | 2 +- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/deploy/00_config.sh b/deploy/00_config.sh index ebda7a1..10749db 100755 --- a/deploy/00_config.sh +++ b/deploy/00_config.sh @@ -50,8 +50,8 @@ export SQL_DB_VERSION="POSTGRES_17" # --- Secret Manager --------------------------------------------------------- export SECRET_FROST_DB_PW="frost-db-password" # HydroVu OAuth client id/secret, read at ingest time via ADC. Matches -# [sources.hydrovu] gcp_secret in .dlt/config.toml. -export SECRET_HYDROVU="hydrovu_pvacd" +# [sources.pvacd_hydrovu] gcp_secret in .dlt/config.toml. +export SECRET_PVACD_HYDROVU="hydrovu_pvacd" # --- GCS buckets ------------------------------------------------------------ # Where Dagster+ writes raw parquet. BUCKET_PROD is the deployment target; diff --git a/deploy/30_dagster_gcp_auth.sh b/deploy/30_dagster_gcp_auth.sh index 283c816..722e2f3 100755 --- a/deploy/30_dagster_gcp_auth.sh +++ b/deploy/30_dagster_gcp_auth.sh @@ -19,7 +19,7 @@ # # Requires: roles/storage.admin (bucket create + IAM), roles/iam.serviceAccountAdmin # (only if the SA does not exist yet), roles/iam.serviceAccountKeyAdmin (only with -# --emit-key), and permission to bind secretAccessor on ${SECRET_HYDROVU}. +# --emit-key), and permission to bind secretAccessor on ${SECRET_PVACD_HYDROVU}. set -euo pipefail source "$(dirname "${BASH_SOURCE[0]}")/00_config.sh" @@ -116,8 +116,8 @@ done # The HydroVu OAuth credentials are themselves fetched through ADC at ingest time, # so bucket access alone is not enough to run the pipeline. -echo "== Grant secretAccessor on ${SECRET_HYDROVU} ==" -gcloud secrets add-iam-policy-binding "${SECRET_HYDROVU}" \ +echo "== Grant secretAccessor on ${SECRET_PVACD_HYDROVU} ==" +gcloud secrets add-iam-policy-binding "${SECRET_PVACD_HYDROVU}" \ --project="${PROJECT_ID}" \ --member="serviceAccount:${DAGSTER_SA}" \ --role="roles/secretmanager.secretAccessor" >/dev/null @@ -184,7 +184,7 @@ cat <} Dagster+ → Deployment → Environment variables (code location aqueduct_dagster_defs_definitions): @@ -206,6 +206,6 @@ Verify without minting a key (needs roles/iam.serviceAccountTokenCreator on the Confirm the bindings landed: gcloud storage buckets get-iam-policy gs://${BUCKET_PROD} --format=json - gcloud secrets get-iam-policy ${SECRET_HYDROVU} --project=${PROJECT_ID} + gcloud secrets get-iam-policy ${SECRET_PVACD_HYDROVU} --project=${PROJECT_ID} gcloud run services get-iam-policy ${FROST_SERVICE} --project=${PROJECT_ID} --region=${REGION} EOF diff --git a/deploy/31_verify_dagster_auth.py b/deploy/31_verify_dagster_auth.py index 5d662e0..ec3552f 100755 --- a/deploy/31_verify_dagster_auth.py +++ b/deploy/31_verify_dagster_auth.py @@ -104,7 +104,7 @@ def check_secret_manager() -> str: config = load_config() project_number = config["destination"]["filesystem"]["gcp_project_number"] - secret_id = config["sources"]["hydrovu"]["gcp_secret"] + secret_id = config["sources"]["pvacd_hydrovu"]["gcp_secret"] client = secretmanager.SecretManagerServiceClient() name = client.secret_version_path(project_number, secret_id, "latest") diff --git a/deploy/README.md b/deploy/README.md index da05efe..cdc1837 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -250,7 +250,7 @@ uv run python deploy/31_verify_dagster_auth.py This checks four layers independently: ADC bootstrap, GCS write/read/delete, Secret Manager access, and an ID-token GET against FROST. `VERBOSE=1` adds tracebacks. For a full end-to-end run instead, `uv run dagster dev` with the same two -variables set and materialize `hydrovu_pipeline`. +variables set and materialize `pvacd_hydrovu_pipeline`. ## How Dagster+ connects to FROST From 55aa62298e84681fb9c8d00156dce0bc790f7e74 Mon Sep 17 00:00:00 2001 From: Casey de la Vega-Rawson Date: Fri, 28 Aug 2026 19:13:04 -0600 Subject: [PATCH 2/4] docs: update docs to reflect new naming convention --- AGENTS.md | 10 +- README.md | 32 ++--- docs/BACKFILL_STRATEGY.md | 10 +- docs/PIPELINE.md | 56 +++++---- docs/STORAGE_CONVENTIONS.md | 112 +++++++++++------- docs/sources/_mapping_template.md | 3 + docs/sources/bernco_hydrovu.md | 39 ++---- .../canonical/CANONICAL_MODEL.md | 2 +- 8 files changed, 137 insertions(+), 127 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 4cc2da9..97de69c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,11 +20,11 @@ Each source is an independent pipeline with three stages: API → dlt → GCS (parquet) → Adapter → CanonicalBundle → FROST loader → FROST ``` -| Stage | Asset (HydroVu) | Asset (CABQ) | +| Stage | Asset (PVACD HydroVu) | Asset (CABQ) | |---|---|---| -| Ingest (dlt → GCS) | `raw_hydrovu_readings` | `raw_cabq_readings` | -| Transform (GCS → CanonicalBundles) | `canonical_bundles_hydrovu` | `canonical_bundles_cabq` | -| Load (CanonicalBundles → FROST) | `frost_load_hydrovu` | `frost_load_cabq` | +| Ingest (dlt → GCS) | `raw_pvacd_hydrovu_readings` | `raw_cabq_readings` | +| Transform (GCS → CanonicalBundles) | `canonical_bundles_pvacd_hydrovu` | `canonical_bundles_cabq` | +| Load (CanonicalBundles → FROST) | `frost_load_pvacd_hydrovu` | `frost_load_cabq` | HydroVu is live; CABQ is scaffolded (`cabq_*` raise `NotImplementedError`). Use HydroVu as the reference implementation when wiring up a new source. @@ -48,7 +48,7 @@ src/aqueduct_dagster/ │ ├── pipeline.py # build_source_pipeline() — shared dlt pipeline factory │ ├── http.py # retry_transient(), TokenManager, BearerAuth, build_authenticated_client() │ └── source_registry.py # SOURCE_REGISTRY — single per-source config, read by definitions.py and load.py -├── sources/ # one folder per agency source (vertical slice) — see hydrovu/ as the reference +├── sources/ # one folder per source key (vertical slice) — see pvacd_hydrovu/ as the reference │ └── / │ ├── adapter.py # raw rows → CanonicalBundle (source-specific) │ ├── dlt_pipeline.py # dlt source/resource/pipeline factory diff --git a/README.md b/README.md index 259c1c5..b030f62 100644 --- a/README.md +++ b/README.md @@ -13,17 +13,17 @@ Dagster + dlt + GCS + FROST SensorThings Two independent source pipelines, each running on its own schedule: ``` -HydroVu API → dlt → GCS (parquet) → HydroVuAdapter → CanonicalBundle → frost_load_hydrovu → FROST -CABQ API → dlt → GCS (parquet) → CabqAdapter → CanonicalBundle → frost_load_cabq → FROST +HydroVu API → dlt → GCS (parquet) → HydroVuAdapter → CanonicalBundle → frost_load_pvacd_hydrovu → FROST +CABQ API → dlt → GCS (parquet) → CabqAdapter → CanonicalBundle → frost_load_cabq → FROST ``` Orchestrated by Dagster. Each pipeline has three assets: -| Asset | HydroVu | CABQ | -|-------|---------|------| -| Ingest (dlt → GCS) | `raw_hydrovu_readings` | `raw_cabq_readings` | -| Transform (GCS → CanonicalBundles) | `canonical_bundles_hydrovu` | `canonical_bundles_cabq` | -| Load (CanonicalBundles → FROST) | `frost_load_hydrovu` | `frost_load_cabq` | +| Asset | PVACD HydroVu | CABQ | +|-------|---------------|------| +| Ingest (dlt → GCS) | `raw_pvacd_hydrovu_readings` | `raw_cabq_readings` | +| Transform (GCS → CanonicalBundles) | `canonical_bundles_pvacd_hydrovu` | `canonical_bundles_cabq` | +| Load (CanonicalBundles → FROST) | `frost_load_pvacd_hydrovu` | `frost_load_cabq` | @@ -60,21 +60,21 @@ Aqueduct/ │ │ ├── http.py # retry_transient(), TokenManager, BearerAuth, build_authenticated_client() │ │ ├── backfill.py # month_chunks(), BackfillCheckpointStore, ChunkResult — Mode A refetch infra │ │ └── source_registry.py # SOURCE_REGISTRY — single per-source config for definitions.py + load.py -│ ├── sources/ # one folder per agency source (vertical slice) -│ │ ├── hydrovu/ +│ ├── sources/ # one folder per source key (vertical slice) +│ │ ├── pvacd_hydrovu/ # PVACD's HydroVu tenant; BernCo's will sit beside it │ │ │ ├── adapter.py # HydroVu → CanonicalBundle mapping │ │ │ ├── dlt_pipeline.py # dlt source + resource + pipeline factory -│ │ │ ├── ingest.py # Dagster asset: raw_hydrovu_readings -│ │ │ ├── transform.py # Dagster asset: canonical_bundles_hydrovu +│ │ │ ├── ingest.py # Dagster asset: raw_pvacd_hydrovu_readings +│ │ │ ├── transform.py # Dagster asset: canonical_bundles_pvacd_hydrovu │ │ │ └── backfill.py # Mode A refetch: isolated ingest + transform + load per chunk -│ │ └── cabq/ # same shape as hydrovu/ — currently a stub +│ │ └── cabq/ # same shape as pvacd_hydrovu/ — currently a stub │ │ ├── adapter.py │ │ ├── dlt_pipeline.py │ │ ├── ingest.py │ │ └── transform.py │ ├── defs/ │ │ ├── assets/ -│ │ │ └── load.py # Dagster assets: frost_load_hydrovu, frost_load_cabq (shared factory) +│ │ │ └── load.py # Dagster assets: frost_load_pvacd_hydrovu, frost_load_cabq (shared factory) │ │ ├── jobs/ │ │ │ └── backfill.py # _backfill_refetch job factory (BackfillRefetchConfig, chunk loop) │ │ ├── definitions.py # Dagster entry point — jobs, schedules, asset registry @@ -84,7 +84,7 @@ Aqueduct/ │ └── watermark_store.py # FrostWatermarkStore — per-run dedup via Dagster context └── tests/ # mirrors src/aqueduct_dagster/'s layout above ├── conftest.py # cross-file test helpers (e.g. httpx.MockTransport/BearerAuth builders) - ├── sources/{hydrovu,cabq}/ + ├── sources/{pvacd_hydrovu,cabq}/ ├── shared/ ├── defs/assets/ ├── defs/jobs/ @@ -190,7 +190,7 @@ Open the Dagster UI at `http://localhost:3000`. To run the full HydroVu pipeline end-to-end: 1. Click **Assets** in the left nav -2. Select all three `hydrovu` group assets (`raw_hydrovu_readings`, `canonical_bundles_hydrovu`, `frost_load_hydrovu`) +2. Select all three `pvacd_hydrovu` group assets (`raw_pvacd_hydrovu_readings`, `canonical_bundles_pvacd_hydrovu`, `frost_load_pvacd_hydrovu`) 3. Click **Materialize selected** On first run, dlt fetches from `initial_start_date` in `.dlt/config.toml` (currently `2026-05-01`). Subsequent runs are incremental. @@ -225,6 +225,6 @@ dlt tracks a cursor (`timestamp` field) per source. On first run it fetches from `FrostWatermarkStore` tracks the last observation timestamp successfully loaded into FROST per datastream. Each run skips any observation at or before the watermark — FROST has no built-in deduplication. **Independent pipelines** -`hydrovu_pipeline` and `cabq_pipeline` are completely independent Dagster jobs. Each has its own schedule and its own terminal load asset (`frost_load_hydrovu` / `frost_load_cabq`). Running one never triggers or blocks the other. +`pvacd_hydrovu_pipeline` and `cabq_pipeline` are completely independent Dagster jobs. Each has its own schedule and its own terminal load asset (`frost_load_pvacd_hydrovu` / `frost_load_cabq`). Running one never triggers or blocks the other. --- diff --git a/docs/BACKFILL_STRATEGY.md b/docs/BACKFILL_STRATEGY.md index c3f11e6..4b0b1b1 100644 --- a/docs/BACKFILL_STRATEGY.md +++ b/docs/BACKFILL_STRATEGY.md @@ -147,7 +147,7 @@ regardless of count. **Option 3 — Per-source generated jobs, two per source (recommended).** This follows the pattern already used in this codebase for -`frost_load_hydrovu` / `frost_load_cabq` and `hydrovu_pipeline` / +`frost_load_pvacd_hydrovu` / `frost_load_cabq` and `pvacd_hydrovu_pipeline` / `cabq_pipeline`: one shared factory function per job type, looped over the source registry, so that adding a third source requires one registry entry and no new job-wiring code. "Which source" is determined by which job is @@ -182,7 +182,7 @@ load together: (DTW filtering, location join, grouping, adapter mapping), factored into a plain function callable from both the normal transform asset and this job, mirroring how `_frost_load()` is already factored out separately - from the `frost_load_hydrovu` asset wrapper. + from the `frost_load_pvacd_hydrovu` asset wrapper. 3. **Load** — always uses the window-scoped, delete-then-repost mechanism described in §4.4, rather than the normal `frost_load` asset's single-watermark filter. This mechanism is safe for every situation in @@ -316,8 +316,8 @@ or list specific ids to scope the run to just those. Each source's existing schedule fires its existing pipeline job automatically, with no operator involvement: -- `hydrovu_schedule` (`0 6 * * *`) triggers `hydrovu_pipeline`, which runs - `raw_hydrovu_readings` → `canonical_bundles_hydrovu` → `frost_load_hydrovu` +- `pvacd_hydrovu_schedule` (`0 6 * * *`) triggers `pvacd_hydrovu_pipeline`, which runs + `raw_pvacd_hydrovu_readings` → `canonical_bundles_pvacd_hydrovu` → `frost_load_pvacd_hydrovu` in sequence. - `cabq_schedule` (`0 8 * * *`) triggers `cabq_pipeline` independently, with its own three assets. Running one source's pipeline never triggers or @@ -331,7 +331,7 @@ and `_backfill_replay`. Neither has a schedule attached — both are launched manually, on demand: 1. Open the Dagster UI and select **Jobs**. -2. Select the relevant job (for example, `hydrovu_backfill_refetch`). +2. Select the relevant job (for example, `pvacd_hydrovu_backfill_refetch`). 3. Open the **Launchpad** and provide the run configuration — entity list, start and end dates, and (for Mode A) an isolated pipeline name — as YAML or JSON directly in the browser. No command-line access is required. diff --git a/docs/PIPELINE.md b/docs/PIPELINE.md index 535a04c..fca990e 100644 --- a/docs/PIPELINE.md +++ b/docs/PIPELINE.md @@ -15,33 +15,33 @@ HydroVu API → dlt → GCS (parquet) → Adapter → CanonicalBundle → FROST | Stage | Dagster asset | Code | |---|---|---| -| 1. Ingest | `raw_hydrovu_readings` | [sources/hydrovu/ingest.py](../src/aqueduct_dagster/sources/hydrovu/ingest.py), [dlt_pipeline.py](../src/aqueduct_dagster/sources/hydrovu/dlt_pipeline.py) | -| 2. Transform | `canonical_bundles_hydrovu` | [sources/hydrovu/transform.py](../src/aqueduct_dagster/sources/hydrovu/transform.py), [adapter.py](../src/aqueduct_dagster/sources/hydrovu/adapter.py) | -| 3. Load | `frost_load_hydrovu` | [defs/assets/load.py](../src/aqueduct_dagster/defs/assets/load.py), [loader/frost_loader.py](../src/aqueduct_dagster/loader/frost_loader.py) | +| 1. Ingest | `raw_pvacd_hydrovu_readings` | [sources/pvacd_hydrovu/ingest.py](../src/aqueduct_dagster/sources/pvacd_hydrovu/ingest.py), [dlt_pipeline.py](../src/aqueduct_dagster/sources/pvacd_hydrovu/dlt_pipeline.py) | +| 2. Transform | `canonical_bundles_pvacd_hydrovu` | [sources/pvacd_hydrovu/transform.py](../src/aqueduct_dagster/sources/pvacd_hydrovu/transform.py), [adapter.py](../src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py) | +| 3. Load | `frost_load_pvacd_hydrovu` | [defs/assets/load.py](../src/aqueduct_dagster/defs/assets/load.py), [loader/frost_loader.py](../src/aqueduct_dagster/loader/frost_loader.py) | -One job (`hydrovu_pipeline`), one daily schedule (`hydrovu_schedule`, cron +One job (`pvacd_hydrovu_pipeline`), one daily schedule (`pvacd_hydrovu_schedule`, cron `0 6 * * *`), no Dagster partitioning — each run just processes everything new since the last watermark. Both the job and schedule are generated from a single config entry, not hand-written (see [Wiring](#wiring)). --- -## Stage 1 — Ingest: `raw_hydrovu_readings` +## Stage 1 — Ingest: `raw_pvacd_hydrovu_readings` No upstream deps — this is the entry point. -[`ingest.py`](../src/aqueduct_dagster/sources/hydrovu/ingest.py) wraps a dlt +[`ingest.py`](../src/aqueduct_dagster/sources/pvacd_hydrovu/ingest.py) wraps a dlt pipeline run in `forward_python_logs_to_dagster` (so dlt's stdlib logging -shows up in the Dagster UI) and calls `pipeline.run(hydrovu_source(...), +shows up in the Dagster UI) and calls `pipeline.run(pvacd_hydrovu_source(...), loader_file_format="parquet")`. It raises `dagster.Failure` only if *every* location errored; otherwise it returns a `MaterializeResult` with metadata (`rows_yielded`, `locations_fetched/skipped/no_data/errored`, `failed_location_ids`). -[`dlt_pipeline.py`](../src/aqueduct_dagster/sources/hydrovu/dlt_pipeline.py) +[`dlt_pipeline.py`](../src/aqueduct_dagster/sources/pvacd_hydrovu/dlt_pipeline.py) defines the actual dlt source: -- `hydrovu_source()` — fetches OAuth creds from GCP Secret Manager (secret +- `pvacd_hydrovu_source()` — fetches OAuth creds from GCP Secret Manager (secret `hydrovu_pvacd`, see [Config](#config)), builds one shared `httpx.Client` via `build_authenticated_client()` + `BearerAuth` (shared infra, see [shared/http.py](../src/aqueduct_dagster/shared/http.py)), fetches the @@ -60,14 +60,14 @@ defines the actual dlt source: `X-ISI-Next-Page` cursor headers, with `retry_transient` handling 429/5xx/transient failures (429 respects `Retry-After`, falling back to a 60s backoff, capped at 3 retries). -- Output lands as parquet in GCS (`raw_pvacd` dataset — see +- Output lands as parquet in GCS (`raw_pvacd_hydrovu` dataset — see [STORAGE_CONVENTIONS.md](STORAGE_CONVENTIONS.md)). -## Stage 2 — Transform: `canonical_bundles_hydrovu` +## Stage 2 — Transform: `canonical_bundles_pvacd_hydrovu` -Deps: `raw_hydrovu_readings`. +Deps: `raw_pvacd_hydrovu_readings`. -[`transform.py`](../src/aqueduct_dagster/sources/hydrovu/transform.py): +[`transform.py`](../src/aqueduct_dagster/sources/pvacd_hydrovu/transform.py): 1. Reads the **transform watermark** (the highest dlt `load_id` already processed, stored as a GCS sidecar file — see @@ -88,7 +88,7 @@ Deps: `raw_hydrovu_readings`. not committed here** — it only advances after Stage 3 confirms FROST accepted the data (see [Idempotency](#idempotency--watermarks)). -[`adapter.py`](../src/aqueduct_dagster/sources/hydrovu/adapter.py) +[`adapter.py`](../src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py) (`HydroVuAdapter`, subclass of the shared `BaseAdapter`) converts one grouped record into canonical entities: @@ -114,9 +114,9 @@ real API evidence), see the canonical model itself, see [canonical/CANONICAL_MODEL.md](../src/aqueduct_dagster/canonical/CANONICAL_MODEL.md). -## Stage 3 — Load: `frost_load_hydrovu` +## Stage 3 — Load: `frost_load_pvacd_hydrovu` -Deps: `canonical_bundles_hydrovu` (via `AssetIn`). +Deps: `canonical_bundles_pvacd_hydrovu` (via `AssetIn`). This stage has **no HydroVu-specific code** — it's generated once, generically, for every source in `SOURCE_REGISTRY` (see [Wiring](#wiring)). @@ -149,7 +149,7 @@ is the single source of truth per source: ```python SOURCE_REGISTRY: list[SourceConfig] = [ - {"name": "hydrovu", "dataset": "raw_pvacd", "cron": "0 6 * * *"}, + {"name": "pvacd_hydrovu", "dataset": "raw_pvacd_hydrovu", "cron": "0 6 * * *"}, {"name": "cabq", "dataset": "raw_cabq", "cron": "0 8 * * *"}, ] ``` @@ -174,8 +174,8 @@ Every stage is safe to re-run: ## Config -All HydroVu-specific settings live in `.dlt/config.toml` under -`[sources.hydrovu]`: `api_base_url`, `token_url`, `gcp_secret` (Secret Manager +All of this source's settings live in `.dlt/config.toml` under +`[sources.pvacd_hydrovu]`: `api_base_url`, `token_url`, `gcp_secret` (Secret Manager secret name — real credentials never touch git), `initial_start_date`, and an explicit `location_ids` allowlist. FROST's target URL is under `[destination.frost]`. GCS bucket/layout is under `[destination.filesystem]` @@ -194,9 +194,9 @@ Google ID token; localhost is called unauthenticated. Mirrors `src/` layout, unit-only (no live GCS/FROST/API calls — see [AGENTS.md](../AGENTS.md)): -- `tests/sources/hydrovu/test_adapter.py` — `HydroVuAdapter` against mock +- `tests/sources/pvacd_hydrovu/test_adapter.py` — `HydroVuAdapter` against mock grouped records. -- `tests/sources/hydrovu/test_dlt_pipeline.py` — pagination, auth-retry, +- `tests/sources/pvacd_hydrovu/test_dlt_pipeline.py` — pagination, auth-retry, 404/429/5xx handling, cursor behavior, via `httpx.MockTransport`. - `tests/shared/test_http.py`, `tests/shared/test_gcs.py` — shared infra (`TokenManager`/`BearerAuth`/`retry_transient`, `read_new_parquet_rows`). @@ -206,16 +206,19 @@ Mirrors `src/` layout, unit-only (no live GCS/FROST/API calls — see ## Adding a new source checklist +1. Pick the source key ``. It is `_` when the data + arrives through a named third-party platform (`pvacd_hydrovu`, `bernco_hydrovu`) + and just `` when the agency serves its own data (`cabq`). 1. Write `docs/sources/.md` from [`_mapping_template.md`](sources/_mapping_template.md), filled against a real sample response -2. Create `sources//` with `adapter.py`, `dlt_pipeline.py`, - `ingest.py`, `transform.py` — mirror HydroVu's structure. -3. Add one entry to `SOURCE_REGISTRY` — this alone generates the job, +1. Create `sources//` with `adapter.py`, `dlt_pipeline.py`, + `ingest.py`, `transform.py` — mirror `pvacd_hydrovu`'s structure. +1. Add one entry to `SOURCE_REGISTRY` — this alone generates the job, schedule, and `frost_load_` asset. -4. Follow [STORAGE_CONVENTIONS.md](STORAGE_CONVENTIONS.md) for the GCS +1. Follow [STORAGE_CONVENTIONS.md](STORAGE_CONVENTIONS.md) for the GCS dataset/table names. -5. Do not touch `loader/` or `canonical/` unless the canonical model itself +1. Do not touch `loader/` or `canonical/` unless the canonical model itself is missing a field — those stay source-agnostic (see [AGENTS.md](../AGENTS.md#the-one-rule-that-explains-the-design)). @@ -226,3 +229,4 @@ Mirrors `src/` layout, unit-only (no live GCS/FROST/API calls — see | Date | Change | |---|---| | 2026-07-20 | Initial version, based on the live PVACD HydroVu pipeline. | +| 2026-08-28 | Renamed the `hydrovu` source to `pvacd_hydrovu` throughout and moved its dataset to `raw_pvacd_hydrovu` (ST2DAT-241), so a second HydroVu tenant can be added without touching this one. Added the source-key step to the new-source checklist. | diff --git a/docs/STORAGE_CONVENTIONS.md b/docs/STORAGE_CONVENTIONS.md index 2ec962a..a600334 100644 --- a/docs/STORAGE_CONVENTIONS.md +++ b/docs/STORAGE_CONVENTIONS.md @@ -8,7 +8,7 @@ grows. When you add a source, a zone, or a partitioning scheme, update the [Changelog](#changelog) at the bottom. - **Status:** raw zone only, date-partitioned, 2 agencies (PVACD via HydroVu live; CABQ scaffolded) -- **Last updated:** 2026-08-13 +- **Last updated:** 2026-08-28 --- @@ -18,16 +18,16 @@ grows. When you add a source, a zone, or a partitioning scheme, update the |---|---|---| | Bucket | lowercase, hyphen-delimited, `nmwdi-aqueduct-` | `nmwdi-aqueduct-production` | | Zone prefix | `raw_` today; reserve `staging_` / `curated_` for later | `raw_` | -| Dataset (top folder) | `raw_`, lowercase `snake_case` | `raw_pvacd` | +| Dataset (top folder) | `raw_`, lowercase `snake_case` | `raw_pvacd_hydrovu` | | Table (sub-folder) | `_`, lowercase `snake_case` | `hydrovu_readings` | | Partition path | date-partitioned, Hive-style `key=value/` | `year=2024/month=06/day=18/` | | Data file | **dlt-managed — never hand-name** | `1781192390.555875.0.parquet` | -| Control / sidecar file | leading underscore, not a data table | `_hydrovu_transform_watermark.json` | +| Control / sidecar file | leading underscore, not a data table | `_pvacd_hydrovu_transform_watermark.json` | Three guiding rules that cover almost everything: 1. **Lowercase + hyphens for buckets, lowercase + `snake_case` for everything inside.** -2. **One folder = one logical thing.** A dataset is one agency; a table folder holds one source's entity; nothing else lives in it. +2. **One folder = one logical thing.** A dataset is one agency's feed from one source system; a table folder holds one entity from it; nothing else lives in it. 3. **Don't invent file names.** dlt owns the file names inside table folders. The only files you name by hand are underscore-prefixed control files. --- @@ -39,7 +39,7 @@ and the pipeline factories: ``` gs://nmwdi-aqueduct-production/ # the raw-zone bucket (one per environment) -├── raw_pvacd/ # agency PVACD — can hold several source tables +├── raw_pvacd_hydrovu/ # PVACD's HydroVu feed — source key `pvacd_hydrovu` │ ├── hydrovu_locations/ # HydroVu source: reference table (write_disposition="replace") │ │ └── year=2024/month=06/day=18/ │ │ └── ..parquet @@ -49,17 +49,22 @@ gs://nmwdi-aqueduct-production/ # the raw-zone bucket (one per environm │ ├── hydrovu_backfill_readings/ # HydroVu Mode A backfill: separate table, own dlt │ │ └── year=2024/month=06/day=18/ # pipeline_name — never read by the normal scheduled │ │ └── ..parquet # transform, so it can't interfere with production -│ ├── metermanager_readings/ # ← example: a 2nd PVACD source (not built yet) -│ │ └── year=2024/month=06/day=18/ -│ │ └── ..parquet -│ ├── _hydrovu_transform_watermark.json # app sidecar: highest load_id transformed +│ ├── _pvacd_hydrovu_transform_watermark.json # app sidecar: highest load_id transformed │ ├── _backfill_checkpoints/ # one file per backfill run_key — completed chunks -│ │ └── hydrovu-jan2026-repair.json # e.g. {"completed_chunks": ["_", ...]} +│ │ └── pvacd_hydrovu-jan2026-repair.json # e.g. {"completed_chunks": ["_", ...]} │ └── _dlt_* # dlt control tables (state, loads, version) -├── raw_pvacd_backfill/ # NOT a real dlt dataset — just the isolated FROST -│ └── _frost_watermarks.json # watermark file backfill jobs read/write, kept -│ # fully separate from raw_pvacd/_frost_watermarks.json -└── raw_cabq/ # agency CABQ (scaffolded) +├── raw_pvacd_hydrovu_backfill/ # NOT a real dlt dataset — just the isolated FROST +│ └── _frost_watermarks.json # watermark file backfill jobs read/write, kept fully +│ # separate from raw_pvacd_hydrovu/_frost_watermarks.json +├── raw_bernco_hydrovu/ # ← BernCo's HydroVu feed (not built yet). Same table +│ └── hydrovu_readings/ # names as PVACD's — the dataset is what separates them. +│ └── year=2024/month=06/day=18/ +│ └── ..parquet +├── raw_pvacd_metermanager/ # ← example: a 2nd PVACD source system (not built yet) +│ └── metermanager_readings/ +│ └── year=2024/month=06/day=18/ +│ └── ..parquet +└── raw_cabq/ # CABQ, which exposes its data directly (scaffolded) └── cabq_readings/ └── year=2024/month=06/day=18/ └── ..parquet @@ -70,7 +75,7 @@ dlt builds these paths from two settings: - `bucket_url` and the date-partitioned `layout` (see [Date partitioning](#date-partitioning)) in `.dlt/config.toml`. `bucket_url` can be overridden by setting a `GCS_BUCKET_URL` env var -- `dataset_name=` in each `build_pipeline()` (`raw_pvacd`, `raw_cabq`), which dlt +- `dataset_name=` in each `build_pipeline()` (`raw_pvacd_hydrovu`, `raw_cabq`), which dlt prepends as the top-level folder. So every object lands at: @@ -100,7 +105,7 @@ across tools: start rather than discovering the collision at provisioning time. - `env` — deployment context: `production`, `stage`, `dev`. - Agency scope is **not** in the bucket name — it lives in the dataset prefix - (`raw_pvacd`, `raw_cabq`), so one production bucket holds every agency's data. + (`raw_pvacd_hydrovu`, `raw_cabq`), so one production bucket holds every agency's data. One bucket per environment keeps IAM and lifecycle rules simple. Don't split a single logical dataset across multiple buckets. @@ -122,23 +127,36 @@ it; when you do, reuse the same dataset/table rules below with the new prefix. ### Datasets (top-level folders) -A dataset = one **agency** — the organization that owns the data (PVACD, CABQ). -Name it `raw_`: `raw_pvacd`, `raw_cabq`, … +A dataset = one **source key** — one agency's feed from one source system. Name it +`raw_`, using the same key as the `sources//` folder, the +`[sources.]` config block, and the `SOURCE_REGISTRY` entry: +`raw_pvacd_hydrovu`, `raw_bernco_hydrovu`, `raw_cabq`, … + +The source key itself is `_` when an agency's data arrives +through a named third-party platform, and just `` when the agency exposes its +data directly (hence `cabq`, not `cabq_cabq`). + +**Why the key and not the agency.** Two things are true at once, and only a +source-keyed dataset covers both: -An agency can expose data through **more than one source system** (HydroVu, -MeterManager, a CKAN portal, …). All of an agency's sources live under that one -agency dataset; the source name lives in the *table* prefix, not the dataset. So -PVACD's HydroVu and (future) MeterManager feeds both sit under `raw_pvacd`: +- One agency, several source systems — PVACD via HydroVu and (future) MeterManager. +- One source system, several agencies — HydroVu serves both PVACD and BernCo, on + separate tenants with separate credentials. + +Keying on the agency alone handles the first and collides on the second: both +tenants' HydroVu feeds would want `hydrovu_readings` inside the same dataset. Keying +on the source pair handles both, and gives each feed its own `_dlt_*` state, its own +`_frost_watermarks.json`, and its own transform watermark with nothing shared: ``` -raw_pvacd/hydrovu_readings/ -raw_pvacd/metermanager_readings/ +raw_pvacd_hydrovu/hydrovu_readings/ # PVACD's HydroVu tenant +raw_bernco_hydrovu/hydrovu_readings/ # BernCo's — same table name, different dataset +raw_pvacd_metermanager/metermanager_readings/ ``` -Each Dagster pipeline sets a `dataset_name`. When an agency gains a second -source, that source's pipeline sets the **same** `dataset_name` as the agency -(`raw_pvacd`) and just uses a distinct table name — multiple pipelines can share -one agency dataset as long as their table names don't collide. +Each Dagster pipeline sets its own `dataset_name`, and no two pipelines share one. +That is the rule that makes a new tenant purely additive: nothing existing has to +move to make room for it. ### Tables (sub-folders) @@ -175,7 +193,7 @@ and lexically separated from real data: - `_dlt_loads`, `_dlt_pipeline_state`, `_dlt_version` — dlt's own bookkeeping. Treat as read-only; never edit or delete. -- `_hydrovu_transform_watermark.json` — our sidecar tracking the highest +- `_pvacd_hydrovu_transform_watermark.json` — our sidecar tracking the highest `load_id` already transformed. New app-managed state files follow the same pattern: `_.json`, written @@ -203,16 +221,16 @@ layout = "{table_name}/year={YYYY}/month={MM}/day={DD}/{load_id}.{file_id}.{ext} Which produces paths like: ``` -raw_pvacd/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet +raw_pvacd_hydrovu/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet ``` instead of the old flat form: ``` -raw_pvacd/hydrovu_readings/1781192390.555875.0.parquet +raw_pvacd_hydrovu/hydrovu_readings/1781192390.555875.0.parquet ``` -**Watermark is unaffected.** `transform_hydrovu.py` derives its incremental +**Watermark is unaffected.** `sources/pvacd_hydrovu/transform.py` derives its incremental watermark from the `load_id` embedded in the *filename*, not from the path — adding `year=`/`month=`/`day=` folders does not change which files it picks up, so no transform code change is needed. @@ -232,20 +250,21 @@ so no transform code change is needed. ## Adding data — checklist -**New source under an existing agency** (e.g. MeterManager for PVACD): - -1. Name each dlt resource `_`; the GCS table folder inherits it. -2. Set the pipeline's `dataset_name` to the **existing** agency dataset - (`raw_`). Pipelines can share it — just don't reuse a table name. -3. Leave `bucket_url` and `layout` alone — they're shared. -4. Any new state file → `_.json` inside the agency dataset folder. -5. Update [Current layout](#current-layout) and add a [Changelog](#changelog) line. - -**New agency** (a brand-new data provider): - -1. Pick the dataset name `raw_` and set `dataset_name="raw_"` in - that pipeline's `build_pipeline()`. -2. Then follow steps 1 and 3–5 above for its first source. +Every new feed is the same checklist, whether it's a new agency, a new source system +for an existing agency, or a second tenant on a source system already in use: + +1. Pick the source key — `_`, or just `` if the agency + exposes its data directly. It names the `sources//` folder, the + `[sources.]` config block, and the `SOURCE_REGISTRY` entry. +2. Set `dataset_name="raw_"` in that pipeline's `build_pipeline()`, and + the matching `dataset` in its `SOURCE_REGISTRY` entry. Never reuse another feed's + dataset. +3. Name each dlt resource `_`; the GCS table folder inherits it. Two + tenants on one platform use the **same** table names — their datasets already + separate them. +4. Leave `bucket_url` and `layout` alone — they're shared. +5. Any new state file → `_.json` inside that feed's dataset folder. +6. Update [Current layout](#current-layout) and add a [Changelog](#changelog) line. --- @@ -259,3 +278,4 @@ so no transform code change is needed. | 2026-07-27 | Isolated the backfill job's FROST watermark from production's — added `raw_pvacd_backfill/_frost_watermarks.json`, a distinct file from `raw_pvacd/_frost_watermarks.json`, so a backfill run can no longer race with or silently advance the daily scheduled pipeline's own watermark. | | 2026-08-11 | Adopted the `nmwdi-` bucket prefix: the pattern is now `nmwdi-aqueduct-` and the production bucket is `gs://nmwdi-aqueduct-production`. The unprefixed `aqueduct-production` referenced in earlier entries was never created — the name is held by another organization, and GCS bucket names are globally unique. `bucket_url` in `.dlt/config.toml` still points at `gs://aqueduct-poc-bravo-pvacd`; moving it is a separate ticket. | | 2026-08-13 | Production moved onto `gs://nmwdi-aqueduct-production` via `GCS_BUCKET_URL` on the Dagster+ full deployment; the committed `bucket_url` stays on `gs://aqueduct-poc-bravo-pvacd` so local runs cannot default to production. Production started **empty** — no data was copied — so dlt cursors restarted from `initial_start_date` and raw parquet from before this date exists only in the POC bucket. | +| 2026-08-28 | Datasets are now keyed on the **source key**, `raw_`, not on the agency (ST2DAT-241). `raw_pvacd` became `raw_pvacd_hydrovu`, and BernCo's HydroVu tenant will land at `raw_bernco_hydrovu`. The old agency rule could not express two agencies on one source system: both tenants' HydroVu feeds would have wanted `hydrovu_readings` in one dataset. Table names are unchanged — `hydrovu_readings` and `hydrovu_locations` stay as they are under both tenants. Nothing was copied: `raw_pvacd/` is left in place, orphaned, and the renamed dataset starts empty, so PVACD's dlt cursors restarted from `initial_start_date`. | diff --git a/docs/sources/_mapping_template.md b/docs/sources/_mapping_template.md index a223fae..c9a609c 100644 --- a/docs/sources/_mapping_template.md +++ b/docs/sources/_mapping_template.md @@ -1,6 +1,9 @@ # Source Mapping: {AGENCY} > Copy this template to `docs/sources/{source_key}.md` and fill it out. +> The source key is `{agency}_{source system}` when the data arrives through a named +> third-party platform (`pvacd_hydrovu`, `bernco_hydrovu`) and just `{agency}` when the +> agency serves its own data (`cabq`) — see `docs/STORAGE_CONVENTIONS.md`. > Fixed fields (same for every source) are pre-filled. Fill in the **Source Field** column for everything else. **Source key:** `{source_key}` diff --git a/docs/sources/bernco_hydrovu.md b/docs/sources/bernco_hydrovu.md index 9208db2..452ff91 100644 --- a/docs/sources/bernco_hydrovu.md +++ b/docs/sources/bernco_hydrovu.md @@ -9,7 +9,7 @@ **Credentials:** GCP Secret Manager, project `waterdatainitiative-271000` (project number `95715287188`), secret name `hydrovu_bernco`, confirmed present 2026-08-24 and separate from PVACD's `hydrovu_pvacd`. The payload is a JSON object with keys `id` and `secret`, -which is what `_resolve_hydrovu_credentials()` in `sources/hydrovu/dlt_pipeline.py` +which is what `_resolve_hydrovu_credentials()` in `sources/pvacd_hydrovu/dlt_pipeline.py` already expects, so no credential-handling code has to change. `deploy/30_dagster_gcp_auth.sh` grants `roles/secretmanager.secretAccessor` on @@ -228,7 +228,7 @@ SIS lookup. > the sensor below the water surface. Parameter 4 is the depth from the reference point > down to water. They share `unitId="35"` (metres) and are easy to conflate, but they > measure in opposite directions. Filtering on `parameter_id == "4"`, as -> `sources/hydrovu/adapter.py` already does, is correct. +> `sources/pvacd_hydrovu/adapter.py` already does, is correct. **A location's parameter set changes over time.** `SP4VuLink-636814` reports parameter 3 in its oldest data (from 2019-11-05) and parameter 4 in recent data. @@ -325,7 +325,7 @@ not catch both. |---|---|---|---|---| | `result` (DTW reading) | metres | `unitId="35"` > `"m"` via `GET /v1/sispec/friendlynames` (2026-08-24). Verified as the only unit used for `parameterId="4"` across all 53 locations | feet | `× 3.28084` | -Identical to PVACD, so `METRES_TO_FEET = 3.28084` in `sources/hydrovu/adapter.py` carries +Identical to PVACD, so `METRES_TO_FEET = 3.28084` in `sources/pvacd_hydrovu/adapter.py` carries over unchanged. Reference factors for the other level units HydroVu can return, none of which appear in this tenant today: @@ -340,25 +340,8 @@ which appear in this tenant today: ## Open Questions -1. **Multi-tenant HydroVu is unmodeled.** BernCo is the first case of two agencies on one - source system, and nothing in the repo anticipates it: - - `.dlt/config.toml` has a single flat `[sources.hydrovu]` block with one `gcp_secret` - (`hydrovu_pvacd`) and one flat `location_ids` list. - - `shared/source_registry.py` has one entry, `{"name": "hydrovu", "dataset": "raw_pvacd"}`. - - `deploy/00_config.sh:54` hardcodes `SECRET_HYDROVU="hydrovu_pvacd"`, and the - `secretmanager.secretAccessor` grant in `deploy/30_dagster_gcp_auth.sh` is scoped to - that one secret. - - `docs/STORAGE_CONVENTIONS.md:128-134` covers only the inverse case (one agency, several - source systems). Following that convention BernCo would land at - `raw_bernco/hydrovu_readings/` with the vertical slice at `sources/bernco_hydrovu/`. - One more thing to settle is whether the existing `hydrovu` source should become - `pvacd_hydrovu` for symmetry. Its dlt pipeline name already is `pvacd_hydrovu` - (`dlt_pipeline.py:503`) while the folder and registry entry are the bare `hydrovu`. - This ticket changes no code and no config. - -2. **HTTP 404 means "no data at or after `startTime`", not "no data endpoint."** - `sources/hydrovu/dlt_pipeline.py:191-193` logs a 404 as "no data endpoint" and treats it +1. **HTTP 404 means "no data at or after `startTime`", not "no data endpoint."** + `sources/pvacd_hydrovu/dlt_pipeline.py:191-193` logs a 404 as "no data endpoint" and treats it as terminal for the run. I retried the 14 BernCo locations that 404 on a recent `startTime`, and 13 of them return full history at `startTime=0`. They are dormant, not endpoint-less. Only `SerenityMesa` (id `4562953333243904`) 404s at `startTime=0` and has @@ -366,25 +349,25 @@ which appear in this tenant today: the log message points a debugger in the wrong direction. Worth a docstring fix in the implementation ticket. -3. **The API has an `endTime` parameter that the code says it does not.** - `sources/hydrovu/dlt_pipeline.py:130-137` states "The real API has no server-side +1. **The API has an `endTime` parameter that the code says it does not.** + `sources/pvacd_hydrovu/dlt_pipeline.py:130-137` states "The real API has no server-side end-time parameter (only startTime)" and implements a client-side cutoff for windowed backfill. The published OpenAPI spec lists `endTime` as a query parameter on `/locations/{id}/data` alongside `startTime`. If it works server-side it would cut the windowing code out of the backfill path. I did not test it. Flagging it for whoever picks up the backfill work. -4. **Which locations are wells?** The Carlito Springs cluster (flume, baro, lower pool) is +1. **Which locations are wells?** The Carlito Springs cluster (flume, baro, lower pool) is surface water and barometric instrumentation, not wells, so the fixed `Thing.name = "Water Well"` would be wrong for them. All three are dormant today and the DTW-only ingest never touches them, but BernCo should confirm the intended scope before anyone draws up the allowlist. -5. **The three factory-default locations.** `default-1191022`, `default-817181` and +1. **The three factory-default locations.** `default-1191022`, `default-817181` and `default-969659` report In-Situ's Fort Collins coordinates. `default-969659` has live DTW data. Does BernCo have real coordinates for these, or should they be excluded? -6. **`initial_start_date` for BernCo.** History runs far deeper than PVACD's. The earliest +1. **`initial_start_date` for BernCo.** History runs far deeper than PVACD's. The earliest readings go back to 2009-05-18 (`BCFDWildlandSub-1091579`), with onsets spread across 2009, 2011, 2013, 2014 (×4), 2015, 2016, 2018, 2023 (×2), 2024 (×8), 2025 (×6) and 2026. A full load from 2009 across 35 DTW locations is a big backfill: at roughly 2-day pages, @@ -392,7 +375,7 @@ which appear in this tenant today: BernCo wants before `initial_start_date` gets set. Build that backfill allowlist from historical parameter coverage, not current (see ObservedProperty above). -7. **Where do `topic`, `is_provisional`, `is_continuous`, `measurement_method` and +1. **Where do `topic`, `is_provisional`, `is_continuous`, `measurement_method` and `data_source` come from?** None appears in any source API. `ebid.md`, `san_acacia.md` and `pvacd_hydrovu.md` all raise versions of this; `is_continuous` is used as a `source_specific` key by two of them but is not documented in diff --git a/src/aqueduct_dagster/canonical/CANONICAL_MODEL.md b/src/aqueduct_dagster/canonical/CANONICAL_MODEL.md index eda0607..0a7cc1c 100644 --- a/src/aqueduct_dagster/canonical/CANONICAL_MODEL.md +++ b/src/aqueduct_dagster/canonical/CANONICAL_MODEL.md @@ -87,7 +87,7 @@ base_adapter.py ← defines the interface every adapter must follow # Each source adapter imports all three: sources/cabq/adapter.py → inherits BaseAdapter → produces CanonicalBundle -sources/hydrovu/adapter.py → inherits BaseAdapter → produces CanonicalBundle +sources/pvacd_hydrovu/adapter.py → inherits BaseAdapter → produces CanonicalBundle sources/source_c/adapter.py → inherits BaseAdapter → produces CanonicalBundle (future) # The loader imports canonical_model.py only: From 730bccc5e2b6b30694326d58b8fd035ea5099178 Mon Sep 17 00:00:00 2001 From: Casey de la Vega-Rawson Date: Fri, 28 Aug 2026 19:28:55 -0600 Subject: [PATCH 3/4] refactor: migrate src/tests to new naming convention --- .dlt/config.toml | 2 +- src/aqueduct_dagster/defs/assets/load.py | 2 +- src/aqueduct_dagster/defs/definitions.py | 8 +- src/aqueduct_dagster/defs/jobs/backfill.py | 26 +++--- .../loader/watermark_store.py | 2 +- src/aqueduct_dagster/shared/backfill.py | 2 +- src/aqueduct_dagster/shared/config.py | 2 +- src/aqueduct_dagster/shared/gcs.py | 4 +- src/aqueduct_dagster/shared/http.py | 6 +- .../shared/source_registry.py | 6 +- .../sources/cabq/dlt_pipeline.py | 4 +- .../sources/cabq/transform.py | 2 +- .../{hydrovu => pvacd_hydrovu}/__init__.py | 0 .../{hydrovu => pvacd_hydrovu}/adapter.py | 2 +- .../{hydrovu => pvacd_hydrovu}/backfill.py | 18 ++-- .../dlt_pipeline.py | 30 +++---- .../{hydrovu => pvacd_hydrovu}/ingest.py | 20 ++--- .../{hydrovu => pvacd_hydrovu}/transform.py | 38 ++++---- tests/conftest.py | 2 +- tests/defs/jobs/test_backfill.py | 8 +- tests/defs/test_definitions.py | 86 +++++++++++++++++++ tests/loader/test_watermark_store.py | 10 ++- tests/shared/test_backfill.py | 32 ++++--- tests/shared/test_config.py | 2 +- tests/shared/test_gcs.py | 2 +- tests/shared/test_http.py | 2 +- .../{hydrovu => pvacd_hydrovu}/__init__.py | 0 .../test_adapter.py | 4 +- .../test_backfill.py | 54 ++++++------ .../test_dlt_pipeline.py | 28 +++--- 30 files changed, 249 insertions(+), 155 deletions(-) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/__init__.py (100%) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/adapter.py (99%) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/backfill.py (94%) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/dlt_pipeline.py (94%) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/ingest.py (82%) rename src/aqueduct_dagster/sources/{hydrovu => pvacd_hydrovu}/transform.py (86%) create mode 100644 tests/defs/test_definitions.py rename tests/sources/{hydrovu => pvacd_hydrovu}/__init__.py (100%) rename tests/sources/{hydrovu => pvacd_hydrovu}/test_adapter.py (99%) rename tests/sources/{hydrovu => pvacd_hydrovu}/test_backfill.py (87%) rename tests/sources/{hydrovu => pvacd_hydrovu}/test_dlt_pipeline.py (94%) diff --git a/.dlt/config.toml b/.dlt/config.toml index 8bc3b71..b093551 100644 --- a/.dlt/config.toml +++ b/.dlt/config.toml @@ -1,4 +1,4 @@ -[sources.hydrovu] +[sources.pvacd_hydrovu] api_base_url = "https://www.hydrovu.com/public-api/v1" token_url = "https://hydrovu.com/public-api/oauth/token" gcp_secret = "hydrovu_pvacd" diff --git a/src/aqueduct_dagster/defs/assets/load.py b/src/aqueduct_dagster/defs/assets/load.py index 0b49e40..d997145 100644 --- a/src/aqueduct_dagster/defs/assets/load.py +++ b/src/aqueduct_dagster/defs/assets/load.py @@ -3,7 +3,7 @@ Terminal load assets — one per source — backed by a shared private helper. - frost_load_hydrovu upstream: canonical_bundles_hydrovu + frost_load_pvacd_hydrovu upstream: canonical_bundles_pvacd_hydrovu frost_load_cabq upstream: canonical_bundles_cabq All sources share _frost_load() logic. Per-source assets are generated by diff --git a/src/aqueduct_dagster/defs/definitions.py b/src/aqueduct_dagster/defs/definitions.py index 59d5c21..a9c0c8f 100644 --- a/src/aqueduct_dagster/defs/definitions.py +++ b/src/aqueduct_dagster/defs/definitions.py @@ -4,7 +4,7 @@ Dagster entry point — all assets, jobs, and schedules registered here. Two independent pipelines — each can be run and scheduled separately: - hydrovu_pipeline: raw_hydrovu_readings → canonical_bundles_hydrovu → frost_load_hydrovu + pvacd_hydrovu_pipeline: raw_pvacd_hydrovu_readings → canonical_bundles_pvacd_hydrovu → frost_load_pvacd_hydrovu cabq_pipeline: raw_cabq_readings → canonical_bundles_cabq → frost_load_cabq Adding source 3: add one entry to shared/source_registry.py's SOURCE_REGISTRY. @@ -23,7 +23,7 @@ from aqueduct_dagster import sources as sources_pkg from aqueduct_dagster.defs import assets as shared_assets_pkg -from aqueduct_dagster.defs.jobs.backfill import hydrovu_backfill_refetch +from aqueduct_dagster.defs.jobs.backfill import pvacd_hydrovu_backfill_refetch from aqueduct_dagster.shared.source_registry import SOURCE_REGISTRY # ── Load all assets ─────────────────────────────────────────────────────────── @@ -58,10 +58,10 @@ # ── Backfill jobs (Mode A refetch) ───────────────────────────────────────────── # Launched manually via run configuration — no schedule attached, see -# docs/BACKFILL_STRATEGY.md §5.2. Only HydroVu is wired so far +# docs/BACKFILL_STRATEGY.md §5.2. Only pvacd_hydrovu is wired so far # (defs/jobs/backfill.py); a future source adds one more factory call there. -_jobs.append(hydrovu_backfill_refetch) +_jobs.append(pvacd_hydrovu_backfill_refetch) # ── Definitions ─────────────────────────────────────────────────────────────── diff --git a/src/aqueduct_dagster/defs/jobs/backfill.py b/src/aqueduct_dagster/defs/jobs/backfill.py index e173bec..319ea76 100644 --- a/src/aqueduct_dagster/defs/jobs/backfill.py +++ b/src/aqueduct_dagster/defs/jobs/backfill.py @@ -3,12 +3,12 @@ Mode A (refetch) backfill jobs — see docs/BACKFILL_STRATEGY.md §4.2, §4.5. -hydrovu_backfill_refetch is the first of these (ST2DAT-202). The job is built +pvacd_hydrovu_backfill_refetch is the first of these (ST2DAT-202). The job is built by a small factory (_make_backfill_refetch_job) parameterized by the same per-source prepare_backfill()/run_backfill_chunk() shape every source's -backfill.py exposes (see sources/hydrovu/backfill.py) — so a future source +backfill.py exposes (see sources/pvacd_hydrovu/backfill.py) — so a future source only needs its own sources//backfill.py plus one more call to the -factory here, no other job-wiring changes. Only HydroVu is wired for now. +factory here, no other job-wiring changes. Only pvacd_hydrovu is wired for now. Not a Dagster asset job — this is a plain @job of one @op, driven entirely by run configuration (BackfillRefetchConfig), matching how an operator actually @@ -46,7 +46,7 @@ ) from aqueduct_dagster.shared.gcs import _gcs_bucket_url, _gcs_filesystem from aqueduct_dagster.shared.source_registry import SOURCE_REGISTRY -from aqueduct_dagster.sources.hydrovu.backfill import ( +from aqueduct_dagster.sources.pvacd_hydrovu.backfill import ( default_backfill_location_ids, prepare_backfill, run_backfill_chunk, @@ -126,7 +126,7 @@ def _make_backfill_refetch_op( 3. dry_run short-circuits here — logs the resolved plan, no GCS/FROST calls. 4. Otherwise processes chunks sequentially, skipping ones already checkpointed for this run_key, checkpointing each only after its - ingest + transform + load succeed (sources/hydrovu/backfill.py's + ingest + transform + load succeed (sources/pvacd_hydrovu/backfill.py's run_backfill_chunk). """ @@ -141,7 +141,7 @@ def _op(context: OpExecutionContext, config: config_cls) -> None: # type: ignor chunks = month_chunks(start, end) # Forwards prepare_fn()/run_chunk_fn()'s stdlib logging (see - # sources/hydrovu/dlt_pipeline.py) plus BackfillCheckpointStore's own + # sources/pvacd_hydrovu/dlt_pipeline.py) plus BackfillCheckpointStore's own # logger ("aqueduct_dagster.shared.backfill", not a descendant of # "aqueduct_dagster.sources.{name}") into this run's log stream. # prepare_fn() runs even during dry_run, so this wraps it unconditionally. @@ -194,7 +194,7 @@ def _op(context: OpExecutionContext, config: config_cls) -> None: # type: ignor fs = _gcs_filesystem() checkpoints = BackfillCheckpointStore(fs, bucket, dataset, run_key=cfg.run_key) # Separate FROST watermark file from production's - # (raw_pvacd/_frost_watermarks.json), same isolation principle as the + # (raw_pvacd_hydrovu/_frost_watermarks.json), same isolation principle as the # separate GCS raw table (hydrovu_backfill_readings vs hydrovu_readings) — # so a backfill run can never race with, or clobber, the daily scheduled # pipeline's own watermark state. @@ -290,7 +290,7 @@ def _job() -> None: class HydroVuBackfillRefetchConfig(BackfillRefetchConfig): """ - hydrovu_backfill_refetch's run configuration. Only overrides + pvacd_hydrovu_backfill_refetch's run configuration. Only overrides location_ids' default (HydroVu's own known-good allowlist, read at import time via default_backfill_location_ids()) — every other field is inherited unchanged from BackfillRefetchConfig. @@ -300,15 +300,15 @@ class HydroVuBackfillRefetchConfig(BackfillRefetchConfig): default=default_backfill_location_ids(), description="HydroVu location IDs to backfill. Defaults to the " "daily pipeline's own allowlist (.dlt/config.toml " - "[sources.hydrovu].location_ids). Leave empty to backfill every " + "[sources.pvacd_hydrovu].location_ids). Leave empty to backfill every " "location the API returns instead.", ) -_hydrovu_registry_cfg = next(cfg for cfg in SOURCE_REGISTRY if cfg["name"] == "hydrovu") -hydrovu_backfill_refetch = _make_backfill_refetch_job( - _hydrovu_registry_cfg["name"], - _hydrovu_registry_cfg["dataset"], +_pvacd_hydrovu_registry_cfg = next(cfg for cfg in SOURCE_REGISTRY if cfg["name"] == "pvacd_hydrovu") +pvacd_hydrovu_backfill_refetch = _make_backfill_refetch_job( + _pvacd_hydrovu_registry_cfg["name"], + _pvacd_hydrovu_registry_cfg["dataset"], prepare_backfill, run_backfill_chunk, HydroVuBackfillRefetchConfig, diff --git a/src/aqueduct_dagster/loader/watermark_store.py b/src/aqueduct_dagster/loader/watermark_store.py index 280b995..e344e54 100644 --- a/src/aqueduct_dagster/loader/watermark_store.py +++ b/src/aqueduct_dagster/loader/watermark_store.py @@ -18,7 +18,7 @@ FrostWatermarkStore — GCS-backed, durable across Dagster restarts InMemoryWatermarkStore — dev/test only, not durable across runs -GCS watermark file: raw_pvacd/_frost_watermarks.json +GCS watermark file: raw_pvacd_hydrovu/_frost_watermarks.json {"pvacd-4745648669458432-dtw": "2026-06-16T18:00:00+00:00", ...} One key per datastream. Written after every successful chunk so a partial failure resumes from the last successful chunk on the next run. diff --git a/src/aqueduct_dagster/shared/backfill.py b/src/aqueduct_dagster/shared/backfill.py index d3aeaad..3d9db48 100644 --- a/src/aqueduct_dagster/shared/backfill.py +++ b/src/aqueduct_dagster/shared/backfill.py @@ -273,7 +273,7 @@ class BackfillCheckpointStore: """ GCS-backed record of which chunks a specific backfill run has completed. - Keyed by an operator-supplied `run_key` (e.g. "hydrovu-jan2026-repair") — + Keyed by an operator-supplied `run_key` (e.g. "pvacd_hydrovu-jan2026-repair") — re-launching the job with the same run_key resumes from the last completed chunk; a different run_key starts fresh and can be run independently (e.g. two unrelated backfills for the same source). diff --git a/src/aqueduct_dagster/shared/config.py b/src/aqueduct_dagster/shared/config.py index 4405bb3..31126bd 100644 --- a/src/aqueduct_dagster/shared/config.py +++ b/src/aqueduct_dagster/shared/config.py @@ -70,7 +70,7 @@ def settings_dir() -> Path: to match when it isn't already set. Exporting the variable is the point of doing this eagerly: dlt resolves - `dlt.config.value` defaults (the whole `[sources.hydrovu]` block) through its own + `dlt.config.value` defaults (the whole `[sources.pvacd_hydrovu]` block) through its own provider chain, not through this module, so it has to be pointed at the same file. Raises FileNotFoundError naming every path tried, since a wrong answer here diff --git a/src/aqueduct_dagster/shared/gcs.py b/src/aqueduct_dagster/shared/gcs.py index 064f776..d5bcbf5 100644 --- a/src/aqueduct_dagster/shared/gcs.py +++ b/src/aqueduct_dagster/shared/gcs.py @@ -132,7 +132,7 @@ def _load_id_from_filename(path: str) -> float | None: """ Extracts the dlt load_id from a parquet filename dlt itself writes. Expected format: .../year={YYYY}/month={MM}/day={DD}/{load_id}.{file_id}.parquet - e.g. raw_pvacd/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet → 1781192390.555875 + e.g. raw_pvacd_hydrovu/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet → 1781192390.555875 """ name = path.split("/")[-1] m = re.match(r"^(\d+\.\d+)\.", name) @@ -170,7 +170,7 @@ def read_new_parquet_rows( keeping only rows where row_filter(row) is True (all rows if row_filter is None). Shared by every source's transform asset for incremental reads — see - hydrovu/transform.py for the reference usage. + pvacd_hydrovu/transform.py for the reference usage. Returns (rows, max_load_id_seen_this_run) — max_load_id is None if no new files. """ diff --git a/src/aqueduct_dagster/shared/http.py b/src/aqueduct_dagster/shared/http.py index c3486e0..d6e0eaa 100644 --- a/src/aqueduct_dagster/shared/http.py +++ b/src/aqueduct_dagster/shared/http.py @@ -8,17 +8,17 @@ client-credentials token that refreshes itself, and a way to attach that token to every request without rebuilding headers/timeout/base-url by hand at each call site. Without shared helpers, all three get hand-rolled per -source — see hydrovu/dlt_pipeline.py's git history before this module +source — see pvacd_hydrovu/dlt_pipeline.py's git history before this module existed, which had the retry loop alone copy-pasted three times. retry_transient() re-raises the final exception once retries are exhausted. Call sites that want a non-raising fallback (e.g. returning an (None, reason) tuple instead of propagating) wrap the call in their own -`except transient_errors:` — see hydrovu/dlt_pipeline.py for the pattern. +`except transient_errors:` — see pvacd_hydrovu/dlt_pipeline.py for the pattern. TokenManager + BearerAuth + build_authenticated_client() together give a source an httpx.Client that: sends a Bearer token on every request, and -transparently refreshes and retries once on a 401 — see hydrovu/dlt_pipeline.py +transparently refreshes and retries once on a 401 — see pvacd_hydrovu/dlt_pipeline.py for the reference usage. Only the OAuth2 client-credentials flow is implemented; a source using a different auth scheme (API key, etc.) would need its own Auth subclass, but can still reuse retry_transient and diff --git a/src/aqueduct_dagster/shared/source_registry.py b/src/aqueduct_dagster/shared/source_registry.py index 5d45332..bba75d7 100644 --- a/src/aqueduct_dagster/shared/source_registry.py +++ b/src/aqueduct_dagster/shared/source_registry.py @@ -17,13 +17,11 @@ class SourceConfig(TypedDict): name: str # source key — must match the sources// folder and asset naming convention - dataset: ( - str # GCS dataset name (raw_) — FROST watermark store + transform watermark path - ) + dataset: str # GCS dataset name, raw_ — FROST watermark store + transform watermark path cron: str # cron schedule for this source's daily pipeline job SOURCE_REGISTRY: list[SourceConfig] = [ - {"name": "hydrovu", "dataset": "raw_pvacd", "cron": "0 6 * * *"}, + {"name": "pvacd_hydrovu", "dataset": "raw_pvacd_hydrovu", "cron": "0 6 * * *"}, {"name": "cabq", "dataset": "raw_cabq", "cron": "0 8 * * *"}, ] diff --git a/src/aqueduct_dagster/sources/cabq/dlt_pipeline.py b/src/aqueduct_dagster/sources/cabq/dlt_pipeline.py index e139f20..5a204aa 100644 --- a/src/aqueduct_dagster/sources/cabq/dlt_pipeline.py +++ b/src/aqueduct_dagster/sources/cabq/dlt_pipeline.py @@ -3,10 +3,10 @@ dlt pipeline for CABQ raw ingestion. -Follows the same pattern as hydrovu_dlt_pipeline.py. +Follows the same pattern as pvacd_hydrovu/dlt_pipeline.py. - @dlt.source: reads config from dlt.config under [cabq] - @dlt.resource: per-location incremental cursor via dlt.current.resource_state() - build_pipeline(): filesystem destination → GCS under raw_cabq/ - - run_pipeline(): convenience entry point (mirrors hydrovu_dlt_pipeline.run_pipeline) + - run_pipeline(): convenience entry point (mirrors pvacd_hydrovu/dlt_pipeline.py run_pipeline) Add CABQ config block to .dlt/config.toml when wiring up: [cabq] diff --git a/src/aqueduct_dagster/sources/cabq/transform.py b/src/aqueduct_dagster/sources/cabq/transform.py index 29cee19..5c5374e 100644 --- a/src/aqueduct_dagster/sources/cabq/transform.py +++ b/src/aqueduct_dagster/sources/cabq/transform.py @@ -8,7 +8,7 @@ - Returns bundles downstream to frost_load_cabq Incremental reads: - Follow the same load_id watermark pattern as hydrovu/transform.py, using the + Follow the same load_id watermark pattern as pvacd_hydrovu/transform.py, using the shared helpers in shared/gcs.py — no need to duplicate this logic: - read_transform_watermark(fs, bucket, WATERMARK_PATH) for since_load_id - read_new_parquet_rows(bucket, glob_suffix, since_load_id, fs, row_filter=...) diff --git a/src/aqueduct_dagster/sources/hydrovu/__init__.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/__init__.py similarity index 100% rename from src/aqueduct_dagster/sources/hydrovu/__init__.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/__init__.py diff --git a/src/aqueduct_dagster/sources/hydrovu/adapter.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py similarity index 99% rename from src/aqueduct_dagster/sources/hydrovu/adapter.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py index eb55f34..3bb640b 100644 --- a/src/aqueduct_dagster/sources/hydrovu/adapter.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py @@ -1,5 +1,5 @@ """ -sources/hydrovu/adapter.py +sources/pvacd_hydrovu/adapter.py Transforms grouped HydroVu parquet rows into CanonicalBundles for FROST. diff --git a/src/aqueduct_dagster/sources/hydrovu/backfill.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/backfill.py similarity index 94% rename from src/aqueduct_dagster/sources/hydrovu/backfill.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/backfill.py index 646e075..ac7c809 100644 --- a/src/aqueduct_dagster/sources/hydrovu/backfill.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/backfill.py @@ -1,5 +1,5 @@ """ -sources/hydrovu/backfill.py +sources/pvacd_hydrovu/backfill.py Mode A (refetch) backfill for PVACD HydroVu — see docs/BACKFILL_STRATEGY.md §4.2. @@ -8,7 +8,7 @@ and its own GCS table (hydrovu_backfill_readings, not hydrovu_readings — see BACKFILL_TABLE_NAME). Because it's a different table, not just a different pipeline_name, its files never match production transform.py's -"raw_pvacd/hydrovu_readings/**/*.parquet" glob — the normal scheduled pipeline +"raw_pvacd_hydrovu/hydrovu_readings/**/*.parquet" glob — the normal scheduled pipeline cannot see this data at all, so there's nothing to coordinate or interfere with. Not a Dagster asset or op itself — no Dagster imports here. Called per-chunk @@ -46,13 +46,13 @@ run_backfill_ingest, ) from aqueduct_dagster.shared.gcs import read_parquet_rows_for_load_id -from aqueduct_dagster.sources.hydrovu.adapter import HydroVuAdapter -from aqueduct_dagster.sources.hydrovu.dlt_pipeline import ( +from aqueduct_dagster.sources.pvacd_hydrovu.adapter import HydroVuAdapter +from aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline import ( _fetch_location_data, _fetch_locations, build_hydrovu_client, ) -from aqueduct_dagster.sources.hydrovu.transform import ( +from aqueduct_dagster.sources.pvacd_hydrovu.transform import ( DTW_PARAMETER_ID, GCS_DATASET, _group_by_location, @@ -144,7 +144,7 @@ def _locations_by_id(locations: list[dict]) -> dict[int, dict]: def default_backfill_location_ids() -> list[int]: """ Same allowlist the daily pipeline reads from .dlt/config.toml - ([sources.hydrovu].location_ids). Called once, eagerly, at + ([sources.pvacd_hydrovu].location_ids). Called once, eagerly, at defs/jobs/backfill.py import time, since Dagster's Launchpad only shows a plain, already-computed default — not a lazily-resolved one. @@ -155,7 +155,7 @@ def default_backfill_location_ids() -> list[int]: failing Dagster's definitions load loudly beats silently defaulting to "backfill everything." """ - return list(load_source_config("hydrovu").get("location_ids", [])) + return list(load_source_config("pvacd_hydrovu").get("location_ids", [])) def prepare_backfill() -> tuple[httpx.Client, list[dict], dict[int, dict]]: @@ -167,12 +167,12 @@ def prepare_backfill() -> tuple[httpx.Client, list[dict], dict[int, dict]]: Returns (client, locations, locations_by_id). """ - cfg = load_source_config("hydrovu") + cfg = load_source_config("pvacd_hydrovu") client = build_hydrovu_client("", "", cfg["gcp_secret"], cfg["api_base_url"], cfg["token_url"]) try: locations = _fetch_locations(client) except Exception: - # Mirrors hydrovu_source() in dlt_pipeline.py: nothing else holds a + # Mirrors pvacd_hydrovu_source() in dlt_pipeline.py: nothing else holds a # reference to this client yet if this raises, so it must close itself. client.close() raise diff --git a/src/aqueduct_dagster/sources/hydrovu/dlt_pipeline.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/dlt_pipeline.py similarity index 94% rename from src/aqueduct_dagster/sources/hydrovu/dlt_pipeline.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/dlt_pipeline.py index 83149ae..7f17ef3 100644 --- a/src/aqueduct_dagster/sources/hydrovu/dlt_pipeline.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/dlt_pipeline.py @@ -1,15 +1,15 @@ """ -sources/hydrovu/dlt_pipeline.py +sources/pvacd_hydrovu/dlt_pipeline.py dlt pipeline for HydroVu raw ingestion. -Two resources returned from hydrovu_source(): +Two resources returned from pvacd_hydrovu_source(): hydrovu_locations (write_disposition="replace") Fetches GET /locations/list on every run and fully replaces the parquet. One row per location: id, name, description, latitude, longitude. Acts as a reference table — rename in HydroVu → latest name in GCS. - Written to: gs:///raw_pvacd/hydrovu_locations/year={YYYY}/month={MM}/day={DD}/ + Written to: gs:///raw_pvacd_hydrovu/hydrovu_locations/year={YYYY}/month={MM}/day={DD}/ hydrovu_readings (write_disposition="append", per-location incremental cursor) Fetches readings per location since that location's last successful fetch. @@ -17,14 +17,14 @@ from the same point next run rather than being skipped permanently. One row per (location, parameter, reading) — location metadata is NOT embedded; join to hydrovu_locations on location_id at transform time. - Written to: gs:///raw_pvacd/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/ + Written to: gs:///raw_pvacd_hydrovu/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/ A TokenManager is created (via build_hydrovu_client(), below) once per run and wrapped in an authenticated httpx.Client (base_url + BearerAuth) shared by both resources, so a single token and a single client cover the full run. -This module is NOT a Dagster asset — it is called by sources/hydrovu/ingest.py -(normal daily pipeline) and sources/hydrovu/backfill.py (Mode A refetch), which +This module is NOT a Dagster asset — it is called by sources/pvacd_hydrovu/ingest.py +(normal daily pipeline) and sources/pvacd_hydrovu/backfill.py (Mode A refetch), which both reuse build_hydrovu_client()/_fetch_locations()/_fetch_location_data() from here rather than duplicating the OAuth/pagination logic. @@ -132,7 +132,7 @@ def _fetch_location_data( are chronological, so a later page would only contain data further beyond the window. The real API has no server-side end-time parameter (only startTime); this is a client-side cutoff. Used by backfill's - windowed chunk fetch (sources/hydrovu/backfill.py). Production's normal + windowed chunk fetch (sources/pvacd_hydrovu/backfill.py). Production's normal ingest (hydrovu_readings, below) always calls with end_time=None — unbounded, fetch-to-present, unchanged from before this parameter existed. @@ -288,7 +288,7 @@ def build_hydrovu_client( ) -> httpx.Client: """ Resolves credentials (Secret Manager if client_id is empty) and returns an - authenticated httpx.Client for the HydroVu API. Shared by hydrovu_source() + authenticated httpx.Client for the HydroVu API. Shared by pvacd_hydrovu_source() (normal ingest) and hydrovu_backfill_source() (backfill.py) so the OAuth/Secret-Manager logic is written once. """ @@ -297,8 +297,8 @@ def build_hydrovu_client( return build_authenticated_client(api_base_url, tm, timeout=_LOCATION_TIMEOUT) -@dlt.source(name="hydrovu") -def hydrovu_source( +@dlt.source(name="pvacd_hydrovu") +def pvacd_hydrovu_source( client_id: str = "", client_secret: str = "", gcp_secret: str = dlt.config.value, @@ -309,7 +309,7 @@ def hydrovu_source( _stats: dict | None = None, ) -> Any: """ - Reads config from dlt.config under [hydrovu]. + Reads config from dlt.config under [sources.pvacd_hydrovu]. Creates a single authenticated httpx.Client shared by both resources, so the token is fetched once and both requests and auth-retries go through one client for the full run. @@ -317,7 +317,7 @@ def hydrovu_source( a redundant second API call. location_ids: allowlist of HydroVu location integer IDs to fetch. - Read from [sources.hydrovu] location_ids in .dlt/config.toml. + Read from [sources.pvacd_hydrovu] location_ids in .dlt/config.toml. Add or remove IDs there without any code change. _stats: optional mutable dict populated with extraction counts after pipeline.run(). @@ -394,7 +394,7 @@ def hydrovu_readings( location_ids: allowlist of HydroVu location integer IDs to fetch. Locations absent from this list are skipped to avoid slow 404s on /locations/{id}/data. - Managed via [sources.hydrovu] location_ids in .dlt/config.toml. + Managed via [sources.pvacd_hydrovu] location_ids in .dlt/config.toml. Incremental: each location has its own cursor stored in dlt.current.resource_state() under "location_cursors". A location's cursor only advances after a successful fetch, @@ -500,11 +500,11 @@ def hydrovu_readings( def build_pipeline() -> dlt.Pipeline: - return build_source_pipeline("pvacd_hydrovu", "raw_pvacd") + return build_source_pipeline("pvacd_hydrovu", "raw_pvacd_hydrovu") def run_pipeline() -> None: """Convenience entry point: builds and runs the pipeline with parquet output.""" pipeline = build_pipeline() - load_info = pipeline.run(hydrovu_source(), loader_file_format="parquet") + load_info = pipeline.run(pvacd_hydrovu_source(), loader_file_format="parquet") logger.info("Load complete: %s", load_info) diff --git a/src/aqueduct_dagster/sources/hydrovu/ingest.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/ingest.py similarity index 82% rename from src/aqueduct_dagster/sources/hydrovu/ingest.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/ingest.py index a6f1613..f426d55 100644 --- a/src/aqueduct_dagster/sources/hydrovu/ingest.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/ingest.py @@ -1,13 +1,13 @@ """ -sources/hydrovu/ingest.py +sources/pvacd_hydrovu/ingest.py -Dagster asset: raw_hydrovu_readings +Dagster asset: raw_pvacd_hydrovu_readings Runs the HydroVu dlt source which writes two resources to GCS: - hydrovu_locations (replace) gs:///raw_pvacd/hydrovu_locations/year={YYYY}/month={MM}/day={DD}/ + hydrovu_locations (replace) gs:///raw_pvacd_hydrovu/hydrovu_locations/year={YYYY}/month={MM}/day={DD}/ Full location list on every run — one row per location. - hydrovu_readings (append, per-location incremental) gs:///raw_pvacd/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/ + hydrovu_readings (append, per-location incremental) gs:///raw_pvacd_hydrovu/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/ New readings since each location's last successful fetch — one row per (location, parameter, reading). Location metadata is omitted; join to hydrovu_locations on location_id at transform time. @@ -18,16 +18,16 @@ from dagster import AssetExecutionContext, Failure, MaterializeResult, MetadataValue, asset from aqueduct_dagster.defs.dagster_logging import forward_python_logs_to_dagster -from aqueduct_dagster.sources.hydrovu.dlt_pipeline import build_pipeline, hydrovu_source +from aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline import build_pipeline, pvacd_hydrovu_source @asset( - name="raw_hydrovu_readings", - group_name="hydrovu", + name="raw_pvacd_hydrovu_readings", + group_name="pvacd_hydrovu", description="Raw HydroVu readings landed in GCS via dlt.", compute_kind="dlt", ) -def raw_hydrovu_readings(context: AssetExecutionContext) -> MaterializeResult: +def raw_pvacd_hydrovu_readings(context: AssetExecutionContext) -> MaterializeResult: """ Runs the dlt pipeline to incrementally fetch HydroVu readings and write them as parquet to the GCS raw zone. @@ -49,8 +49,8 @@ def raw_hydrovu_readings(context: AssetExecutionContext) -> MaterializeResult: pipeline.dataset_name, ) stats: dict = {} - with forward_python_logs_to_dagster(context, "aqueduct_dagster.sources.hydrovu", "dlt"): - load_info = pipeline.run(hydrovu_source(_stats=stats), loader_file_format="parquet") + with forward_python_logs_to_dagster(context, "aqueduct_dagster.sources.pvacd_hydrovu", "dlt"): + load_info = pipeline.run(pvacd_hydrovu_source(_stats=stats), loader_file_format="parquet") context.log.info("HydroVu dlt load complete: %s", load_info) diff --git a/src/aqueduct_dagster/sources/hydrovu/transform.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/transform.py similarity index 86% rename from src/aqueduct_dagster/sources/hydrovu/transform.py rename to src/aqueduct_dagster/sources/pvacd_hydrovu/transform.py index 2f29fc3..8857203 100644 --- a/src/aqueduct_dagster/sources/hydrovu/transform.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/transform.py @@ -1,29 +1,29 @@ """ -sources/hydrovu/transform.py +sources/pvacd_hydrovu/transform.py -Dagster asset: canonical_bundles_hydrovu +Dagster asset: canonical_bundles_pvacd_hydrovu - Reads only NEW hydrovu_readings parquet from GCS since the last successful run - Always reads the latest hydrovu_locations parquet (replace resource — one file) - Filters readings to DTW rows only (parameter_id="4") - Joins readings to locations on location_id to restore name/lat/lon metadata - Groups joined rows by location_id into one record per location - Runs HydroVuAdapter to produce CanonicalBundles (one per DTW location) - - Returns bundles downstream to frost_load_hydrovu + - Returns bundles downstream to frost_load_pvacd_hydrovu Incremental reads (readings only): - A watermark file (raw_pvacd/_hydrovu_transform_watermark.json) in GCS tracks + A watermark file (raw_pvacd_hydrovu/_pvacd_hydrovu_transform_watermark.json) in GCS tracks the highest dlt load_id processed so far. On each run only readings parquet files with a newer load_id are read. The watermark is updated after a successful run. load_id is the float Unix timestamp dlt embeds in every parquet filename: - raw_pvacd/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/{load_id}.{file_id}.parquet - e.g. raw_pvacd/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet + raw_pvacd_hydrovu/hydrovu_readings/year={YYYY}/month={MM}/day={DD}/{load_id}.{file_id}.parquet + e.g. raw_pvacd_hydrovu/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet Locations parquet (hydrovu_locations/) uses write_disposition="replace" so it is always a single up-to-date file — read fresh on every run, no watermark needed. -Upstream: raw_hydrovu_readings -Downstream: frost_load_hydrovu +Upstream: raw_pvacd_hydrovu_readings +Downstream: frost_load_pvacd_hydrovu """ import logging @@ -43,7 +43,7 @@ read_transform_watermark, transform_watermark_path, ) -from aqueduct_dagster.sources.hydrovu.adapter import HydroVuAdapter +from aqueduct_dagster.sources.pvacd_hydrovu.adapter import HydroVuAdapter @dataclass @@ -61,9 +61,9 @@ class HydroVuTransformResult: logger = logging.getLogger(__name__) -GCS_DATASET = "raw_pvacd" +GCS_DATASET = "raw_pvacd_hydrovu" DTW_PARAMETER_ID = "4" -WATERMARK_PATH = transform_watermark_path(GCS_DATASET, "hydrovu") +WATERMARK_PATH = transform_watermark_path(GCS_DATASET, "pvacd_hydrovu") def _read_locations_from_gcs(bucket_url: str, fs: gcsfs.GCSFileSystem) -> dict[int, dict]: @@ -77,7 +77,7 @@ def _read_locations_from_gcs(bucket_url: str, fs: gcsfs.GCSFileSystem) -> dict[i if not files: raise FileNotFoundError( f"No locations parquet found at {pattern}. " - "Ensure raw_hydrovu_readings has run at least once." + "Ensure raw_pvacd_hydrovu_readings has run at least once." ) locations: dict[int, dict] = {} @@ -135,7 +135,7 @@ def _transform_metadata( since_load_id: float | None, max_load_id: float | None, ) -> dict[str, MetadataValue]: - """Shared shape for canonical_bundles_hydrovu's output metadata — used by + """Shared shape for canonical_bundles_pvacd_hydrovu's output metadata — used by both the no-new-rows early return and the normal path, so the two can't drift out of sync on key names.""" return { @@ -149,20 +149,20 @@ def _transform_metadata( @asset( - name="canonical_bundles_hydrovu", - group_name="hydrovu", + name="canonical_bundles_pvacd_hydrovu", + group_name="pvacd_hydrovu", description="CanonicalBundles produced by HydroVuAdapter from GCS raw parquet.", compute_kind="python", - deps=["raw_hydrovu_readings"], + deps=["raw_pvacd_hydrovu_readings"], ) -def canonical_bundles_hydrovu( +def canonical_bundles_pvacd_hydrovu( context: AssetExecutionContext, ) -> HydroVuTransformResult: """ Reads only new HydroVu parquet from GCS (since last run), filters to DTW readings, groups by location, and runs HydroVuAdapter to produce CanonicalBundles. - Does NOT write the watermark — that happens in frost_load_hydrovu after FROST + Does NOT write the watermark — that happens in frost_load_pvacd_hydrovu after FROST confirms success, so a FROST failure leaves the watermark unadvanced and the next run retries the same data. """ @@ -206,7 +206,7 @@ def canonical_bundles_hydrovu( adapter = HydroVuAdapter(records) with forward_python_logs_to_dagster( - context, "aqueduct_dagster.sources.hydrovu", "aqueduct_dagster.canonical" + context, "aqueduct_dagster.sources.pvacd_hydrovu", "aqueduct_dagster.canonical" ): bundles = list(adapter.run()) context.log.info("Produced %d CanonicalBundles", len(bundles)) diff --git a/tests/conftest.py b/tests/conftest.py index 5dc809e..2dd0e00 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,7 @@ tests/conftest.py Shared test helpers for httpx.Client/BearerAuth-based code, used by both -tests/shared/test_http.py and tests/sources/hydrovu/test_dlt_pipeline.py. +tests/shared/test_http.py and tests/sources/pvacd_hydrovu/test_dlt_pipeline.py. Consolidated here after the two files independently grew near-identical make_tm()/client_with_responses() helpers. """ diff --git a/tests/defs/jobs/test_backfill.py b/tests/defs/jobs/test_backfill.py index bc98f40..013a34d 100644 --- a/tests/defs/jobs/test_backfill.py +++ b/tests/defs/jobs/test_backfill.py @@ -237,13 +237,13 @@ def test_frost_watermark_dataset_is_isolated_from_production( prepare_fn = _prepare_fn() run_chunk_fn = MagicMock(return_value=_stub_chunk_result()) - job = _make_backfill_refetch_job("test", "raw_pvacd", prepare_fn, run_chunk_fn) + job = _make_backfill_refetch_job("test", "raw_pvacd_hydrovu", prepare_fn, run_chunk_fn) job.execute_in_process(run_config=_run_config(dry_run=False)) mock_build_loader.assert_called_once() called_dataset = mock_build_loader.call_args[0][1] - assert called_dataset == "raw_pvacd_backfill" - assert called_dataset != "raw_pvacd" # never the same dataset production uses + assert called_dataset == "raw_pvacd_hydrovu_backfill" + assert called_dataset != "raw_pvacd_hydrovu" # never the same dataset production uses @patch("aqueduct_dagster.defs.jobs.backfill.forward_python_logs_to_dagster") @@ -256,7 +256,7 @@ def test_real_run_forwards_python_logs_with_source_specific_prefix( ): """ Regression test: prepare_fn()/run_chunk_fn() emit per-location/per-page - progress via stdlib logging (see sources/hydrovu/dlt_pipeline.py), which + progress via stdlib logging (see sources/pvacd_hydrovu/dlt_pipeline.py), which only reaches the Dagster run log if forward_python_logs_to_dagster wraps the call — this was originally missing, leaving a silent multi-minute gap in real backfill runs. Also covers BackfillCheckpointStore's own logger diff --git a/tests/defs/test_definitions.py b/tests/defs/test_definitions.py new file mode 100644 index 0000000..f479b76 --- /dev/null +++ b/tests/defs/test_definitions.py @@ -0,0 +1,86 @@ +""" +tests/defs/test_definitions.py + +Guards the two places where SOURCE_REGISTRY's strings have to agree with something +written independently somewhere else. Both failure modes are silent — nothing raises, +the run just does the wrong thing — so they are worth a test rather than a comment. + +1. Asset/job/schedule generation. defs/definitions.py and defs/assets/load.py build + names by f-string from the registry `name`, while sources// hard-codes the + same names in its @asset decorators. A mismatch produces a job whose selection + names assets that do not exist; it only surfaces when someone launches a run. + +2. The dataset string. Each source writes it three times — SOURCE_REGISTRY, the + source's build_pipeline(), and its transform module's GCS_DATASET — with nothing + cross-checking them. If the registry and transform disagree, the load asset commits + the transform watermark to a path the transform never reads, and every run + reprocesses from zero. shared/gcs.py:transform_watermark_path() unifies the + filename but not the dataset it sits in. + +Offline: no GCS, FROST, or dlt destination is touched — build_source_pipeline is +patched out so build_pipeline() can be inspected without credentials. +""" + +from __future__ import annotations + +import importlib +from unittest.mock import patch + +import pytest + +from aqueduct_dagster.defs.definitions import defs +from aqueduct_dagster.shared.gcs import transform_watermark_path +from aqueduct_dagster.shared.source_registry import SOURCE_REGISTRY + +_NAMES = [cfg["name"] for cfg in SOURCE_REGISTRY] + + +def _asset_keys() -> set[str]: + graph = defs.resolve_asset_graph() + return {key.to_user_string() for key in graph.get_all_asset_keys()} + + +@pytest.mark.parametrize("name", _NAMES) +def test_registry_entry_has_its_three_assets(name): + """raw_{name}_readings → canonical_bundles_{name} → frost_load_{name} all resolve.""" + expected = { + f"raw_{name}_readings", + f"canonical_bundles_{name}", + f"frost_load_{name}", + } + assert expected <= _asset_keys() + + +@pytest.mark.parametrize("name", _NAMES) +def test_registry_entry_has_its_job_and_schedule(name): + assert f"{name}_pipeline" in {job.name for job in defs.resolve_all_job_defs()} + assert f"{name}_schedule" in {schedule.name for schedule in defs.schedules} + + +@pytest.mark.parametrize("cfg", SOURCE_REGISTRY, ids=_NAMES) +def test_transform_module_agrees_with_registry_dataset(cfg): + """ + The transform's GCS_DATASET and WATERMARK_PATH must match what defs/assets/load.py + derives from the registry — the read side and the write side of the same file. + """ + transform = importlib.import_module(f"aqueduct_dagster.sources.{cfg['name']}.transform") + + assert transform.GCS_DATASET == cfg["dataset"] + assert transform.WATERMARK_PATH == transform_watermark_path(cfg["dataset"], cfg["name"]) + + +@pytest.mark.parametrize("cfg", SOURCE_REGISTRY, ids=_NAMES) +def test_dlt_pipeline_writes_to_the_registry_dataset(cfg): + """ + build_pipeline() passes dataset_name positionally to build_source_pipeline(); it is + the third independent copy of the dataset string and the one that decides where + parquet actually lands. + """ + module = f"aqueduct_dagster.sources.{cfg['name']}.dlt_pipeline" + dlt_pipeline = importlib.import_module(module) + + with patch(f"{module}.build_source_pipeline") as mock_build: + dlt_pipeline.build_pipeline() + + _pipeline_name, dataset_name = mock_build.call_args.args + assert dataset_name == cfg["dataset"] diff --git a/tests/loader/test_watermark_store.py b/tests/loader/test_watermark_store.py index 4e51c92..f0071bf 100644 --- a/tests/loader/test_watermark_store.py +++ b/tests/loader/test_watermark_store.py @@ -39,7 +39,9 @@ def _make_store(gcs_content: dict[str, str] | None = None) -> FrostWatermarkStor mock_fs.open.return_value.__enter__ = lambda _: io.StringIO(raw) mock_fs.open.return_value.__exit__ = MagicMock(return_value=False) - return FrostWatermarkStore(mock_context, mock_fs, "my-bucket", dataset="raw_pvacd"), mock_fs + return FrostWatermarkStore( + mock_context, mock_fs, "my-bucket", dataset="raw_pvacd_hydrovu" + ), mock_fs # ── InMemoryWatermarkStore ─────────────────────────────────────────────────── @@ -72,7 +74,7 @@ def test_corrupt_gcs_file_treated_as_first_run(): mock_fs.open.return_value.__enter__ = lambda _: io.StringIO("not valid json{{{") mock_fs.open.return_value.__exit__ = MagicMock(return_value=False) - store = FrostWatermarkStore(mock_context, mock_fs, "my-bucket", dataset="raw_pvacd") + store = FrostWatermarkStore(mock_context, mock_fs, "my-bucket", dataset="raw_pvacd_hydrovu") assert store.get("any-key") is None assert store._loaded is True @@ -152,8 +154,8 @@ def test_set_writes_to_tmp_then_renames(): ts = datetime(2026, 6, 20, tzinfo=UTC) store.set("ds-1", ts) - tmp_path = f"my-bucket/raw_pvacd/{_FROST_WATERMARKS_FILENAME}.tmp" - final_path = f"my-bucket/raw_pvacd/{_FROST_WATERMARKS_FILENAME}" + tmp_path = f"my-bucket/raw_pvacd_hydrovu/{_FROST_WATERMARKS_FILENAME}.tmp" + final_path = f"my-bucket/raw_pvacd_hydrovu/{_FROST_WATERMARKS_FILENAME}" mock_fs.open.assert_called_with(tmp_path, "w") mock_fs.rename.assert_called_once_with(tmp_path, final_path) diff --git a/tests/shared/test_backfill.py b/tests/shared/test_backfill.py index b6afcf2..778735d 100644 --- a/tests/shared/test_backfill.py +++ b/tests/shared/test_backfill.py @@ -185,7 +185,9 @@ def _make_store(gcs_content: dict | None = None) -> tuple[BackfillCheckpointStor mock_fs.open.return_value.__enter__ = lambda _: io.StringIO(raw) mock_fs.open.return_value.__exit__ = MagicMock(return_value=False) - store = BackfillCheckpointStore(mock_fs, "my-bucket", "raw_pvacd", run_key="hydrovu-jan2026") + store = BackfillCheckpointStore( + mock_fs, "my-bucket", "raw_pvacd_hydrovu", run_key="hydrovu-jan2026" + ) return store, mock_fs @@ -200,9 +202,9 @@ def test_run_key_is_sanitized_in_checkpoint_path(): characters can't split the two identifiers for the same run apart. """ store = BackfillCheckpointStore( - MagicMock(), "my-bucket", "raw_pvacd", run_key="team/jan-2026 fix" + MagicMock(), "my-bucket", "raw_pvacd_hydrovu", run_key="team/jan-2026 fix" ) - assert store._path == "my-bucket/raw_pvacd/_backfill_checkpoints/team_jan-2026_fix.json" + assert store._path == "my-bucket/raw_pvacd_hydrovu/_backfill_checkpoints/team_jan-2026_fix.json" def test_is_complete_false_when_no_checkpoint_file(): @@ -220,7 +222,7 @@ def test_corrupt_checkpoint_file_treated_as_fresh_start(): mock_fs = MagicMock() mock_fs.open.return_value.__enter__ = lambda _: io.StringIO("not valid json{{{") mock_fs.open.return_value.__exit__ = MagicMock(return_value=False) - store = BackfillCheckpointStore(mock_fs, "my-bucket", "raw_pvacd", run_key="r") + store = BackfillCheckpointStore(mock_fs, "my-bucket", "raw_pvacd_hydrovu", run_key="r") assert store.is_complete(*CHUNK_1) is False @@ -240,8 +242,8 @@ def test_mark_complete_writes_tmp_then_renames(): store.mark_complete(*CHUNK_1) - tmp_path = "my-bucket/raw_pvacd/_backfill_checkpoints/hydrovu-jan2026.json.tmp" - final_path = "my-bucket/raw_pvacd/_backfill_checkpoints/hydrovu-jan2026.json" + tmp_path = "my-bucket/raw_pvacd_hydrovu/_backfill_checkpoints/hydrovu-jan2026.json.tmp" + final_path = "my-bucket/raw_pvacd_hydrovu/_backfill_checkpoints/hydrovu-jan2026.json" mock_fs.open.assert_called_with(tmp_path, "w") mock_fs.rename.assert_called_once_with(tmp_path, final_path) @@ -325,8 +327,8 @@ def test_save_raises_after_all_retries_exhausted(): @patch("aqueduct_dagster.shared.backfill.load_config") def test_load_source_config_reads_named_section(mock_load_config): - mock_load_config.return_value = {"sources": {"hydrovu": {"gcp_secret": "x"}}} - assert load_source_config("hydrovu") == {"gcp_secret": "x"} + mock_load_config.return_value = {"sources": {"pvacd_hydrovu": {"gcp_secret": "x"}}} + assert load_source_config("pvacd_hydrovu") == {"gcp_secret": "x"} # ── build_backfill_pipeline ──────────────────────────────────────────────────── @@ -335,17 +337,19 @@ def test_load_source_config_reads_named_section(mock_load_config): @patch("aqueduct_dagster.shared.backfill.build_source_pipeline") def test_build_backfill_pipeline_includes_run_key_in_pipeline_name(mock_build_source_pipeline): build_backfill_pipeline( - pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd", run_key="jan-repair" + pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd_hydrovu", run_key="jan-repair" ) args, _kwargs = mock_build_source_pipeline.call_args - assert args == ("hydrovu_backfill_jan-repair", "raw_pvacd") + assert args == ("hydrovu_backfill_jan-repair", "raw_pvacd_hydrovu") @patch("aqueduct_dagster.shared.backfill.build_source_pipeline") def test_build_backfill_pipeline_sanitizes_run_key(mock_build_source_pipeline): """Two different run_keys must never collide into the same pipeline_name.""" build_backfill_pipeline( - pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd", run_key="jan repair/v2" + pipeline_name_prefix="hydrovu_backfill", + dataset="raw_pvacd_hydrovu", + run_key="jan repair/v2", ) args, _kwargs = mock_build_source_pipeline.call_args assert args[0] == "hydrovu_backfill_jan_repair_v2" @@ -401,10 +405,12 @@ def test_run_backfill_ingest_forwards_prefix_dataset_and_run_key(mock_build_pipe mock_pipeline.run.return_value = MagicMock(loads_ids=["100.0"]) mock_build_pipeline.return_value = mock_pipeline - _run_ingest(pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd", run_key="jan-repair") + _run_ingest( + pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd_hydrovu", run_key="jan-repair" + ) mock_build_pipeline.assert_called_once_with( - pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd", run_key="jan-repair" + pipeline_name_prefix="hydrovu_backfill", dataset="raw_pvacd_hydrovu", run_key="jan-repair" ) diff --git a/tests/shared/test_config.py b/tests/shared/test_config.py index 8fa43df..1d05e49 100644 --- a/tests/shared/test_config.py +++ b/tests/shared/test_config.py @@ -88,7 +88,7 @@ def test_falls_back_to_cwd(self, monkeypatch, tmp_path): class TestDltProjectDirExport: def test_exports_env_for_dlt(self, monkeypatch, tmp_path): """ - dlt resolves the [sources.hydrovu] block through its own provider chain, so + dlt resolves the [sources.pvacd_hydrovu] block through its own provider chain, so it has to be pointed at the same file we resolved. """ root = _make_settings(tmp_path / "repo") diff --git a/tests/shared/test_gcs.py b/tests/shared/test_gcs.py index 5d681e7..360fe80 100644 --- a/tests/shared/test_gcs.py +++ b/tests/shared/test_gcs.py @@ -57,7 +57,7 @@ def test_empty_env_var_falls_back_to_config_toml(self, monkeypatch: pytest.Monke class TestLoadIdFromFilename: def test_extracts_load_id(self): - path = "bucket/raw_pvacd/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet" + path = "bucket/raw_pvacd_hydrovu/hydrovu_readings/year=2024/month=06/day=18/1781192390.555875.0.parquet" assert _load_id_from_filename(path) == 1781192390.555875 def test_returns_none_for_unrecognized_name(self): diff --git a/tests/shared/test_http.py b/tests/shared/test_http.py index c3cee80..8a0cafe 100644 --- a/tests/shared/test_http.py +++ b/tests/shared/test_http.py @@ -9,7 +9,7 @@ Covers: TokenManager — caching, expiry, force-refresh (moved verbatim - from tests/sources/hydrovu/test_dlt_pipeline.py + from tests/sources/pvacd_hydrovu/test_dlt_pipeline.py when TokenManager moved to shared/http.py) BearerAuth — attaches token, refreshes + retries once on 401 build_authenticated_client — wires base_url, default headers, auth, timeout diff --git a/tests/sources/hydrovu/__init__.py b/tests/sources/pvacd_hydrovu/__init__.py similarity index 100% rename from tests/sources/hydrovu/__init__.py rename to tests/sources/pvacd_hydrovu/__init__.py diff --git a/tests/sources/hydrovu/test_adapter.py b/tests/sources/pvacd_hydrovu/test_adapter.py similarity index 99% rename from tests/sources/hydrovu/test_adapter.py rename to tests/sources/pvacd_hydrovu/test_adapter.py index 5183c44..f2827cb 100644 --- a/tests/sources/hydrovu/test_adapter.py +++ b/tests/sources/pvacd_hydrovu/test_adapter.py @@ -1,5 +1,5 @@ """ -tests/sources/hydrovu/test_adapter.py +tests/sources/pvacd_hydrovu/test_adapter.py Unit tests for HydroVuAdapter. No real API calls — uses mock records matching the grouped record shape @@ -25,7 +25,7 @@ from datetime import UTC -from aqueduct_dagster.sources.hydrovu.adapter import ( +from aqueduct_dagster.sources.pvacd_hydrovu.adapter import ( METRES_TO_FEET, HydroVuAdapter, ) diff --git a/tests/sources/hydrovu/test_backfill.py b/tests/sources/pvacd_hydrovu/test_backfill.py similarity index 87% rename from tests/sources/hydrovu/test_backfill.py rename to tests/sources/pvacd_hydrovu/test_backfill.py index eaf92bd..a3632aa 100644 --- a/tests/sources/hydrovu/test_backfill.py +++ b/tests/sources/pvacd_hydrovu/test_backfill.py @@ -1,7 +1,7 @@ """ -tests/sources/hydrovu/test_backfill.py +tests/sources/pvacd_hydrovu/test_backfill.py -Unit tests for sources/hydrovu/backfill.py (Mode A refetch). +Unit tests for sources/pvacd_hydrovu/backfill.py (Mode A refetch). No live API/GCS/FROST — all I/O is mocked. """ @@ -14,7 +14,7 @@ import pytest from aqueduct_dagster.loader.frost_loader import LoadResult -from aqueduct_dagster.sources.hydrovu.backfill import ( +from aqueduct_dagster.sources.pvacd_hydrovu.backfill import ( BACKFILL_PIPELINE_NAME, BACKFILL_TABLE_NAME, GCS_DATASET, @@ -56,7 +56,7 @@ class TestHydroVuBackfillReadings: - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_only_fetches_allowlisted_locations(self, mock_fetch): mock_fetch.return_value = (_READINGS_DATA, None) list( @@ -71,7 +71,7 @@ def test_only_fetches_allowlisted_locations(self, mock_fetch): called_ids = {call[0][1] for call in mock_fetch.call_args_list} assert called_ids == {111} - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_passes_start_and_end_ts_through(self, mock_fetch): mock_fetch.return_value = (_READINGS_DATA, None) list( @@ -87,7 +87,7 @@ def test_passes_start_and_end_ts_through(self, mock_fetch): assert start_time == 123 assert mock_fetch.call_args[1]["end_time"] == 456 - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_yields_flat_rows(self, mock_fetch): mock_fetch.return_value = (_READINGS_DATA, None) rows = list( @@ -110,7 +110,7 @@ def test_yields_flat_rows(self, mock_fetch): } ] - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_skips_location_on_404(self, mock_fetch): mock_fetch.return_value = (None, None) rows = list( @@ -124,7 +124,7 @@ def test_skips_location_on_404(self, mock_fetch): ) assert rows == [] - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_raises_on_real_fetch_error(self, mock_fetch): # dlt wraps the generator's exception in its own ResourceExtractionError # when iterated directly (as it would be inside pipeline.run()) — match @@ -141,7 +141,7 @@ def test_raises_on_real_fetch_error(self, mock_fetch): ) ) - @patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_location_data") def test_fetch_error_message_includes_the_chunk_window(self, mock_fetch): """ An operator glancing at a failed run should immediately see which @@ -181,16 +181,16 @@ def test_locations_by_id_shape(): # ── prepare_backfill ───────────────────────────────────────────────────────── -@patch("aqueduct_dagster.sources.hydrovu.backfill.load_source_config") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.load_source_config") def test_default_backfill_location_ids_reads_the_configured_allowlist(mock_cfg): mock_cfg.return_value = {"location_ids": [111, 222]} assert default_backfill_location_ids() == [111, 222] -@patch("aqueduct_dagster.sources.hydrovu.backfill.load_source_config") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.load_source_config") def test_default_backfill_location_ids_is_empty_when_key_not_configured(mock_cfg): """ - A [sources.hydrovu] section with no location_ids key at all is not an + A [sources.pvacd_hydrovu] section with no location_ids key at all is not an error — some sources may deliberately not curate an allowlist — so this returns [] (meaning "every location", see resolve_location_ids), not a KeyError. @@ -199,7 +199,7 @@ def test_default_backfill_location_ids_is_empty_when_key_not_configured(mock_cfg assert default_backfill_location_ids() == [] -@patch("aqueduct_dagster.sources.hydrovu.backfill.load_source_config") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.load_source_config") def test_default_backfill_location_ids_raises_on_missing_config(mock_cfg): """ Raises when .dlt/config.toml itself can't be read at all — a broken @@ -213,9 +213,9 @@ def test_default_backfill_location_ids_raises_on_missing_config(mock_cfg): default_backfill_location_ids() -@patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_locations") -@patch("aqueduct_dagster.sources.hydrovu.backfill.build_hydrovu_client") -@patch("aqueduct_dagster.sources.hydrovu.backfill.load_source_config") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_locations") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.build_hydrovu_client") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.load_source_config") def test_prepare_backfill_fetches_locations_once(mock_cfg, mock_build_client, mock_fetch_locations): mock_cfg.return_value = { "gcp_secret": "hydrovu_pvacd", @@ -233,9 +233,9 @@ def test_prepare_backfill_fetches_locations_once(mock_cfg, mock_build_client, mo mock_fetch_locations.assert_called_once_with(_DUMMY_CLIENT) -@patch("aqueduct_dagster.sources.hydrovu.backfill._fetch_locations") -@patch("aqueduct_dagster.sources.hydrovu.backfill.build_hydrovu_client") -@patch("aqueduct_dagster.sources.hydrovu.backfill.load_source_config") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill._fetch_locations") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.build_hydrovu_client") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.load_source_config") def test_prepare_backfill_closes_client_if_fetch_locations_fails( mock_cfg, mock_build_client, mock_fetch_locations ): @@ -286,8 +286,8 @@ def load_window(self, datastream_key, datastream_id, records, window_start, wind CHUNK_END = datetime(2026, 2, 1, tzinfo=UTC) -@patch("aqueduct_dagster.sources.hydrovu.backfill.read_parquet_rows_for_load_id") -@patch("aqueduct_dagster.sources.hydrovu.backfill.run_backfill_ingest") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.read_parquet_rows_for_load_id") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.run_backfill_ingest") def test_run_backfill_chunk_reads_by_exact_load_id_and_loads_bundles( mock_run_ingest, mock_read_rows ): @@ -339,8 +339,8 @@ def test_run_backfill_chunk_reads_by_exact_load_id_and_loads_bundles( assert len(records) == 1 -@patch("aqueduct_dagster.sources.hydrovu.backfill.read_parquet_rows_for_load_id") -@patch("aqueduct_dagster.sources.hydrovu.backfill.run_backfill_ingest") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.read_parquet_rows_for_load_id") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.run_backfill_ingest") def test_run_backfill_chunk_reports_adapter_failures_without_dropping_good_locations( mock_run_ingest, mock_read_rows ): @@ -387,8 +387,8 @@ def test_run_backfill_chunk_reports_adapter_failures_without_dropping_good_locat assert len(loader.load_window_calls) == 1 -@patch("aqueduct_dagster.sources.hydrovu.backfill.read_parquet_rows_for_load_id") -@patch("aqueduct_dagster.sources.hydrovu.backfill.run_backfill_ingest") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.read_parquet_rows_for_load_id") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.run_backfill_ingest") def test_run_backfill_chunk_with_no_rows_loads_nothing(mock_run_ingest, mock_read_rows): mock_run_ingest.return_value = 100.0 mock_read_rows.return_value = [] @@ -414,8 +414,8 @@ def test_run_backfill_chunk_with_no_rows_loads_nothing(mock_run_ingest, mock_rea assert loader.ensure_calls == [] -@patch("aqueduct_dagster.sources.hydrovu.backfill.read_parquet_rows_for_load_id") -@patch("aqueduct_dagster.sources.hydrovu.backfill.run_backfill_ingest") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.read_parquet_rows_for_load_id") +@patch("aqueduct_dagster.sources.pvacd_hydrovu.backfill.run_backfill_ingest") def test_run_backfill_chunk_handles_empty_loads_ids_without_crashing( mock_run_ingest, mock_read_rows ): diff --git a/tests/sources/hydrovu/test_dlt_pipeline.py b/tests/sources/pvacd_hydrovu/test_dlt_pipeline.py similarity index 94% rename from tests/sources/hydrovu/test_dlt_pipeline.py rename to tests/sources/pvacd_hydrovu/test_dlt_pipeline.py index dfd006f..ed821af 100644 --- a/tests/sources/hydrovu/test_dlt_pipeline.py +++ b/tests/sources/pvacd_hydrovu/test_dlt_pipeline.py @@ -1,5 +1,5 @@ """ -tests/sources/hydrovu/test_dlt_pipeline.py +tests/sources/pvacd_hydrovu/test_dlt_pipeline.py Unit tests for the HydroVu dlt pipeline private helpers. No real API calls — HTTP interactions are simulated via httpx.MockTransport, @@ -26,7 +26,7 @@ import pytest from aqueduct_dagster.shared.http import BearerAuth, TokenManager -from aqueduct_dagster.sources.hydrovu.dlt_pipeline import ( +from aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline import ( _fetch_location_data, _fetch_locations, _resolve_hydrovu_credentials, @@ -221,7 +221,7 @@ def test_429_respects_retry_after_header(self): assert mock_sleep.call_args_list[0][0][0] == 30.0 def test_429_uses_default_backoff_when_no_retry_after(self): - from aqueduct_dagster.sources.hydrovu.dlt_pipeline import _429_BACKOFF + from aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline import _429_BACKOFF client, _ = _client_with_responses([httpx.Response(429)] * 4) with patch("time.sleep") as mock_sleep: @@ -313,9 +313,11 @@ def test_returns_immediately_when_client_id_already_given(self): result = _resolve_hydrovu_credentials("cid", "csecret", "ignored-secret-name") assert result == ("cid", "csecret") - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline.secretmanager.SecretManagerServiceClient") - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline.ensure_adc") - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline.load_config") + @patch( + "aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline.secretmanager.SecretManagerServiceClient" + ) + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline.ensure_adc") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline.load_config") def test_fetches_from_secret_manager_when_client_id_empty( self, mock_load_config, mock_ensure_adc, mock_sm_cls ): @@ -373,7 +375,7 @@ def test_returns_authenticated_client_without_secret_manager(self): class TestHydroVuReadingsFilter: @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_only_fetches_allowlisted_locations(self, mock_fetch, _mock_state): mock_fetch.return_value = (_READINGS_DATA, None) list( @@ -388,7 +390,7 @@ def test_only_fetches_allowlisted_locations(self, mock_fetch, _mock_state): assert called_ids == {111, 222} @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_skips_locations_not_in_allowlist(self, mock_fetch, _mock_state): mock_fetch.return_value = (_READINGS_DATA, None) list( @@ -404,7 +406,7 @@ def test_skips_locations_not_in_allowlist(self, mock_fetch, _mock_state): assert 333 not in called_ids @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_empty_allowlist_skips_all_locations(self, mock_fetch, _mock_state): list( hydrovu_readings( @@ -422,7 +424,7 @@ def test_empty_allowlist_skips_all_locations(self, mock_fetch, _mock_state): class TestHydroVuReadingsErrorStats: @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_real_error_increments_errored_count(self, mock_fetch, _mock_state): mock_fetch.return_value = (None, "HTTP 500") stats: dict = {} @@ -439,7 +441,7 @@ def test_real_error_increments_errored_count(self, mock_fetch, _mock_state): assert 111 in stats["failed_location_ids"] @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_404_does_not_increment_errored_count(self, mock_fetch, _mock_state): mock_fetch.return_value = (None, None) stats: dict = {} @@ -457,7 +459,7 @@ def test_404_does_not_increment_errored_count(self, mock_fetch, _mock_state): assert stats["failed_location_ids"] == [] @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_error_does_not_advance_cursor(self, mock_fetch, _mock_state): state = {"location_cursors": {"111": 999}} with patch("dlt.current.resource_state", return_value=state): @@ -473,7 +475,7 @@ def test_error_does_not_advance_cursor(self, mock_fetch, _mock_state): assert state["location_cursors"]["111"] == 999 # unchanged @patch("dlt.current.resource_state", return_value={"location_cursors": {}}) - @patch("aqueduct_dagster.sources.hydrovu.dlt_pipeline._fetch_location_data") + @patch("aqueduct_dagster.sources.pvacd_hydrovu.dlt_pipeline._fetch_location_data") def test_partial_failure_stats(self, mock_fetch, _mock_state): # location 111 succeeds, 222 errors mock_fetch.side_effect = [(_READINGS_DATA, None), (None, "HTTP 503")] From b654599ff6b7c38b456a2928a47414a079e49010 Mon Sep 17 00:00:00 2001 From: Casey de la Vega-Rawson Date: Tue, 1 Sep 2026 18:46:31 -0600 Subject: [PATCH 4/4] refactor: rename HydroVuBackfillRefetchConfig to PvacdHydroVuBackfillRefetchConfig --- src/aqueduct_dagster/defs/jobs/backfill.py | 16 +++++++++------- .../sources/pvacd_hydrovu/adapter.py | 2 +- tests/defs/jobs/test_backfill.py | 4 ++-- tests/sources/pvacd_hydrovu/test_backfill.py | 2 +- 4 files changed, 13 insertions(+), 11 deletions(-) diff --git a/src/aqueduct_dagster/defs/jobs/backfill.py b/src/aqueduct_dagster/defs/jobs/backfill.py index 7297d3f..418744f 100644 --- a/src/aqueduct_dagster/defs/jobs/backfill.py +++ b/src/aqueduct_dagster/defs/jobs/backfill.py @@ -20,7 +20,7 @@ end_date, run_key, dry_run), prefilled with example values, plus validation (date format/order, and an auto-attached run_key timestamp). location_ids is also shared, but its default differs per source — a per-source subclass (e.g. -HydroVuBackfillRefetchConfig) only overrides that one field's default, since +PvacdHydroVuBackfillRefetchConfig) only overrides that one field's default, since that's the only thing a new source needs to customize. """ @@ -79,7 +79,7 @@ class BackfillRefetchConfig[LocationId](Config): as-is (dry_run: true). Fields here are common to every source. Per-source subclasses (e.g. - HydroVuBackfillRefetchConfig below) parameterize LocationId with their own + PvacdHydroVuBackfillRefetchConfig below) parameterize LocationId with their own concrete id type and override location_ids' default; everything else is inherited. Validation lives in shared/backfill.py as plain, Dagster-free functions so Mode B (replay) can reuse it too. @@ -313,12 +313,14 @@ def _job() -> None: return _job -class HydroVuBackfillRefetchConfig(BackfillRefetchConfig[int]): +class PvacdHydroVuBackfillRefetchConfig(BackfillRefetchConfig[int]): """ pvacd_hydrovu_backfill_refetch's run configuration. Only overrides - location_ids' default (HydroVu's own known-good allowlist, read at - import time via default_backfill_location_ids()) — every other field is - inherited unchanged from BackfillRefetchConfig. + location_ids' default (PVACD's own known-good allowlist, read at import + time via default_backfill_location_ids()) — every other field is inherited + unchanged from BackfillRefetchConfig. Tenant-scoped, not vendor-scoped: a + second HydroVu tenant gets its own subclass reading its own allowlist, it + does not reuse this one. """ location_ids: list[int] = Field( @@ -336,7 +338,7 @@ class HydroVuBackfillRefetchConfig(BackfillRefetchConfig[int]): _pvacd_hydrovu_registry_cfg["dataset"], pvacd_hydrovu_prepare_backfill, pvacd_hydrovu_run_backfill_chunk, - HydroVuBackfillRefetchConfig, + PvacdHydroVuBackfillRefetchConfig, ) diff --git a/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py b/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py index 3bb640b..5add837 100644 --- a/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py +++ b/src/aqueduct_dagster/sources/pvacd_hydrovu/adapter.py @@ -3,7 +3,7 @@ Transforms grouped HydroVu parquet rows into CanonicalBundles for FROST. -Called by hydrovu/transform.py which: +Called by pvacd_hydrovu/transform.py which: 1. Reads raw parquet from GCS 2. Filters to DTW rows (parameter_id="4") before grouping 3. Groups filtered rows by location_id into one record per location diff --git a/tests/defs/jobs/test_backfill.py b/tests/defs/jobs/test_backfill.py index 1a47a7d..962c598 100644 --- a/tests/defs/jobs/test_backfill.py +++ b/tests/defs/jobs/test_backfill.py @@ -25,7 +25,7 @@ from aqueduct_dagster.defs.jobs.backfill import ( BackfillRefetchConfig, CabqBackfillRefetchConfig, - HydroVuBackfillRefetchConfig, + PvacdHydroVuBackfillRefetchConfig, _make_backfill_refetch_job, ) @@ -148,7 +148,7 @@ def test_already_timestamped_run_key_is_left_unchanged(): # shared, non-generic location_ids type). _LOCATION_ID_CASES = [ - pytest.param(HydroVuBackfillRefetchConfig, 111, id="hydrovu"), + pytest.param(PvacdHydroVuBackfillRefetchConfig, 111, id="pvacd_hydrovu"), pytest.param(CabqBackfillRefetchConfig, "IW4", id="cabq"), ] diff --git a/tests/sources/pvacd_hydrovu/test_backfill.py b/tests/sources/pvacd_hydrovu/test_backfill.py index a3632aa..fe99cd5 100644 --- a/tests/sources/pvacd_hydrovu/test_backfill.py +++ b/tests/sources/pvacd_hydrovu/test_backfill.py @@ -206,7 +206,7 @@ def test_default_backfill_location_ids_raises_on_missing_config(mock_cfg): environment, not an intentional "backfill everything" choice — instead of silently falling back to [] and widening a reviewed allowlist into "backfill everything" at Dagster's definitions-load time (see - HydroVuBackfillRefetchConfig). + PvacdHydroVuBackfillRefetchConfig). """ mock_cfg.side_effect = FileNotFoundError("no .dlt/config.toml") with pytest.raises(FileNotFoundError):