From b375aed6dd8ec030245e63734a77f8c1f578e772 Mon Sep 17 00:00:00 2001 From: cka-y Date: Thu, 23 Jul 2026 09:55:13 -0400 Subject: [PATCH 1/3] feat: rebuilding validation report between dates for a version --- .../src/tasks/validation_reports/README.md | 60 ++++++++++++ .../rebuild_missing_validation_reports.py | 92 +++++++++++++++++-- ...test_rebuild_missing_validation_reports.py | 85 +++++++++++++++++ 3 files changed, 229 insertions(+), 8 deletions(-) diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/README.md b/functions-python/tasks_executor/src/tasks/validation_reports/README.md index 00b1b1a7e..be53d51a1 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/README.md +++ b/functions-python/tasks_executor/src/tasks/validation_reports/README.md @@ -63,6 +63,12 @@ sequenceDiagram Finds GTFS datasets that are missing a validation report **or** have a report from an older validator version, then triggers a GCP Workflow for each one. +By default "older version" means "not the exact version returned by the validator +endpoint". Set `filter_validator_version_prefix` (e.g. `"8."`) to instead target +datasets that lack **any** report matching that major-version prefix — useful when you +want to re-validate everything missing an `8.*` report regardless of the exact patch +version the endpoint reports. + The task is **resumable**: if it times out mid-loop, calling it again skips datasets that were already triggered (tracked in `task_execution_log`). @@ -74,8 +80,11 @@ that were already triggered (tracked in `task_execution_log`). "validator_endpoint": "https://stg-gtfs-validator-web-mbzoxaljzq-ue.a.run.app", "bypass_db_update": false, "filter_after_in_days": 30, + "filter_downloaded_after": "2026-03-01", + "filter_downloaded_before": "2026-06-01", "filter_statuses": ["active"], "filter_op_statuses": ["published"], + "filter_validator_version_prefix": "8.", "force_update": false, "limit": 10, "reports_bucket_name": "stg-gtfs-validator-results" @@ -88,8 +97,11 @@ that were already triggered (tracked in `task_execution_log`). | `validator_endpoint` | string | env-derived | Validator service URL to use and fetch version from | | `bypass_db_update` | bool | `false` | When `true`, results are NOT written to DB/API (use for pre-release runs) | | `filter_after_in_days` | int | `null` | Restrict to datasets downloaded within the last N days. Omit to include all datasets | +| `filter_downloaded_after` | string (ISO date) | `null` | Restrict to datasets downloaded **at or after** this date (inclusive), e.g. `"2026-03-01"`. Omit for no lower bound | +| `filter_downloaded_before` | string (ISO date) | `null` | Restrict to datasets downloaded **strictly before** this date (exclusive), e.g. `"2026-06-01"`. Omit for no upper bound | | `filter_statuses` | list[str] | `null` | Filter feeds by status (e.g. `["active", "inactive"]`). Omit for all statuses | | `filter_op_statuses` | list[str] | `["published"]` | Filter feeds by operational status. Accepted values: `"published"`, `"unpublished"`, `"wip"` | +| `filter_validator_version_prefix` | string | `null` | When set (e.g. `"8."`), only include datasets that do **not** already have a report whose `validator_version` starts with this prefix. Overrides the default exact-version comparison | | `force_update` | bool | `false` | Re-trigger even when a current report already exists | | `limit` | int | `null` | Cap the number of workflows triggered per call — useful for end-to-end testing | | `reports_bucket_name` | string | env-derived | Override the GCS bucket where validator results are stored. Use when running in prod but pointing to the staging validator (e.g. `"stg-gtfs-validator-results"`) | @@ -273,6 +285,54 @@ curl -X POST "https://ingest-data-to-big-query-gtfs-563580583640.northamerica-no --- +## Re-validate a date window missing a major version (prod) + +Use this when a validator fix means a range of historical datasets needs a fresh report +from the **production** validator — e.g. re-running validator `8.x` on datasets published +between March and May 2026 that never got an `8.*` report. + +Run this from the **prod-deployed** function so `ENVIRONMENT=prod` makes the prod validator +URL and results bucket the defaults. + +### Step 1 — Dry run (estimate scope) + +```json +{ + "task": "rebuild_missing_validation_reports", + "payload": { + "dry_run": true, + "filter_downloaded_after": "2026-03-01", + "filter_downloaded_before": "2026-06-01", + "filter_validator_version_prefix": "8." + } +} +``` + +Check `total_candidates` in the response. `filter_validator_version_prefix: "8."` excludes +datasets that already have any `8.*` report, so re-running is idempotent. + +### Step 2 — Full run + +Flip `dry_run` to `false` (optionally add `"limit": 10` first for an end-to-end test): + +```json +{ + "task": "rebuild_missing_validation_reports", + "payload": { + "dry_run": false, + "filter_downloaded_after": "2026-03-01", + "filter_downloaded_before": "2026-06-01", + "filter_validator_version_prefix": "8." + } +} +``` + +`sync_task_run_status` then polls in the background until `ready_for_bigquery: true` (see +above). Because `bypass_db_update` is omitted (defaults to `false`), the new reports are +written to the DB/API. + +--- + ## GCP Environment Variables | Variable | Default | Description | diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py index 4d0991276..53e0e44dd 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py @@ -59,10 +59,18 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: "dry_run": bool, # [optional] If True, count only — do not trigger workflows. Default: True "filter_after_in_days": int, # [optional] Restrict to datasets downloaded within the last N days. # If omitted, all datasets are considered regardless of age. + "filter_downloaded_after": str, # [optional] ISO date (e.g. "2026-03-01"). Restrict to datasets + # downloaded at or after this date (inclusive). + "filter_downloaded_before": str, # [optional] ISO date (e.g. "2026-06-01"). Restrict to datasets + # downloaded strictly before this date (exclusive). "filter_statuses": list[str],# [optional] Filter feeds by status "filter_op_statuses": list[str],# [optional] Filter feeds by operational status. # Default: ["published"] # Accepted values: "published", "unpublished", "wip" + "filter_validator_version_prefix": str, # [optional] Only include datasets that do NOT already + # have a validation report whose validator_version starts with this + # prefix (e.g. "8." → datasets missing any 8.* report). When omitted, + # datasets are selected by comparing against the exact target version. "validator_endpoint": str, # [optional] Override validator URL (e.g. staging). Default: env-derived URL. "bypass_db_update": bool, # [optional] If True, results are NOT written to the DB/API (pre-release runs). Default: False @@ -76,8 +84,11 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: ( dry_run, filter_after_in_days, + filter_downloaded_after, + filter_downloaded_before, filter_statuses, filter_op_statuses, + filter_validator_version_prefix, prod_env, validator_endpoint, bypass_db_update, @@ -91,8 +102,11 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: bypass_db_update=bypass_db_update, dry_run=dry_run, filter_after_in_days=filter_after_in_days, + filter_downloaded_after=filter_downloaded_after, + filter_downloaded_before=filter_downloaded_before, filter_statuses=filter_statuses, filter_op_statuses=filter_op_statuses, + filter_validator_version_prefix=filter_validator_version_prefix, prod_env=prod_env, force_update=force_update, limit=limit, @@ -106,8 +120,11 @@ def rebuild_missing_validation_reports( bypass_db_update: bool = False, dry_run: bool = True, filter_after_in_days: Optional[int] = None, + filter_downloaded_after: Optional[str] = None, + filter_downloaded_before: Optional[str] = None, filter_statuses: List[str] | None = None, filter_op_statuses: List[str] | None = None, + filter_validator_version_prefix: Optional[str] = None, prod_env: bool = False, force_update: bool = False, limit: Optional[int] = None, @@ -124,9 +141,17 @@ def rebuild_missing_validation_reports( dry_run: If True, count only — do not trigger workflows. Default: True filter_after_in_days: Restrict to datasets downloaded within the last N days. If None (default), all datasets are considered regardless of age. + filter_downloaded_after: ISO date (e.g. "2026-03-01"). Restrict to datasets + downloaded at or after this date (inclusive). If None (default), no lower bound. + filter_downloaded_before: ISO date (e.g. "2026-06-01"). Restrict to datasets + downloaded strictly before this date (exclusive). If None (default), no upper bound. filter_statuses: Filter feeds by status. Default: None (all) filter_op_statuses: Filter feeds by operational status. Default: ["published"]. Accepted: "published", "unpublished", "wip". + filter_validator_version_prefix: If set (e.g. "8."), only include datasets that do + NOT already have a validation report whose validator_version starts with this + prefix. When None (default), datasets are selected by comparing against the exact + target validator version. prod_env: True if targeting the production environment. Default: False force_update: Re-trigger even if a report already exists. Default: False limit: Max datasets to trigger per call (for end-to-end testing). Default: unlimited @@ -146,10 +171,13 @@ def rebuild_missing_validation_reports( validator_version=validator_version, force_update=force_update, filter_after_in_days=filter_after_in_days, + filter_downloaded_after=filter_downloaded_after, + filter_downloaded_before=filter_downloaded_before, filter_statuses=filter_statuses, filter_op_statuses=( filter_op_statuses if filter_op_statuses is not None else ["published"] ), + filter_validator_version_prefix=filter_validator_version_prefix, ) total_candidates = len(datasets) logging.info("Found %s candidate datasets", total_candidates) @@ -227,7 +255,10 @@ def rebuild_missing_validation_reports( "validator_endpoint": validator_endpoint, "bypass_db_update": bypass_db_update, "filter_after_in_days": filter_after_in_days, + "filter_downloaded_after": filter_downloaded_after, + "filter_downloaded_before": filter_downloaded_before, "filter_statuses": filter_statuses, + "filter_validator_version_prefix": filter_validator_version_prefix, "prod_env": prod_env, "force_update": force_update, "limit": limit, @@ -253,8 +284,11 @@ def _get_datasets_for_validation( validator_version: str, force_update: bool, filter_after_in_days: Optional[int], + filter_downloaded_after: Optional[str], + filter_downloaded_before: Optional[str], filter_statuses: Optional[List[str]], filter_op_statuses: Optional[List[str]], + filter_validator_version_prefix: Optional[str], ) -> List[tuple]: """ Query datasets that need a (re)validation. @@ -264,28 +298,59 @@ def _get_datasets_for_validation( - Have a report from a different (older) validator version, OR - force_update is True + When filter_validator_version_prefix is set (e.g. "8."), the exact-version + comparison is replaced by "the dataset does not already have any validation + report whose validator_version starts with the prefix". This is what lets us + re-validate datasets missing an 8.* report regardless of the exact patch + version returned by the (prod) validator endpoint. + filter_after_in_days restricts to datasets downloaded within the last N days. - When None, all datasets are included regardless of age. + filter_downloaded_after / filter_downloaded_before restrict to a fixed date + window (inclusive lower bound, exclusive upper bound) on downloaded_at. + When all age filters are None, all datasets are included regardless of age. filter_op_statuses filters by Feed.operational_status (e.g. ["published"]). """ query = ( db_session.query(Gtfsfeed.stable_id, Gtfsdataset.stable_id) .select_from(Gtfsfeed) .join(Gtfsdataset, Gtfsfeed.latest_dataset_id == Gtfsdataset.id) - .outerjoin(Validationreport, Gtfsdataset.validation_reports) - .filter( + ) + + if filter_validator_version_prefix: + # Include datasets that do NOT already have a report matching the version + # prefix (using a correlated EXISTS over the dataset's reports). + has_matching_report = Gtfsdataset.validation_reports.any( + Validationreport.validator_version.like( + f"{filter_validator_version_prefix}%" + ) + ) + query = query.filter(or_(~has_matching_report, force_update)) + else: + query = query.outerjoin( + Validationreport, Gtfsdataset.validation_reports + ).filter( or_( Validationreport.id.is_(None), Validationreport.validator_version != validator_version, force_update, ) ) - .distinct(Gtfsfeed.stable_id, Gtfsdataset.stable_id) - .order_by(Gtfsdataset.stable_id, Gtfsfeed.stable_id) + + query = query.distinct(Gtfsfeed.stable_id, Gtfsdataset.stable_id).order_by( + Gtfsdataset.stable_id, Gtfsfeed.stable_id ) + if filter_after_in_days is not None: filter_after = datetime.today() - timedelta(days=filter_after_in_days) query = query.filter(Gtfsdataset.downloaded_at >= filter_after) + if filter_downloaded_after is not None: + query = query.filter( + Gtfsdataset.downloaded_at >= datetime.fromisoformat(filter_downloaded_after) + ) + if filter_downloaded_before is not None: + query = query.filter( + Gtfsdataset.downloaded_at < datetime.fromisoformat(filter_downloaded_before) + ) if filter_statuses: query = query.filter(Gtfsfeed.status.in_(filter_statuses)) if filter_op_statuses: @@ -333,9 +398,10 @@ def get_parameters(payload): Args: payload (dict): Task payload dict. Returns: - Tuple of (dry_run, filter_after_in_days, filter_statuses, filter_op_statuses, - prod_env, validator_endpoint, bypass_db_update, force_update, limit, - reports_bucket_name) + Tuple of (dry_run, filter_after_in_days, filter_downloaded_after, + filter_downloaded_before, filter_statuses, filter_op_statuses, + filter_validator_version_prefix, prod_env, validator_endpoint, + bypass_db_update, force_update, limit, reports_bucket_name) """ prod_env = os.getenv("ENVIRONMENT", "").lower() == "prod" default_endpoint = get_gtfs_validator_url(prod_env) @@ -351,10 +417,17 @@ def get_parameters(payload): else int(filter_after_in_days) ) + filter_downloaded_after = payload.get("filter_downloaded_after", None) + filter_downloaded_before = payload.get("filter_downloaded_before", None) + filter_statuses = payload.get("filter_statuses", None) filter_op_statuses = payload.get("filter_op_statuses", None) + filter_validator_version_prefix = payload.get( + "filter_validator_version_prefix", None + ) + validator_endpoint = payload.get("validator_endpoint", default_endpoint) bypass_db_update = payload.get("bypass_db_update", False) @@ -380,8 +453,11 @@ def get_parameters(payload): return ( dry_run, filter_after_in_days, + filter_downloaded_after, + filter_downloaded_before, filter_statuses, filter_op_statuses, + filter_validator_version_prefix, prod_env, validator_endpoint, bypass_db_update, diff --git a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py index 3440a923e..df83fda71 100644 --- a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py @@ -32,8 +32,11 @@ def test_defaults(self): ( dry_run, filter_after_in_days, + filter_downloaded_after, + filter_downloaded_before, filter_statuses, filter_op_statuses, + filter_validator_version_prefix, prod_env, validator_endpoint, bypass_db_update, @@ -43,8 +46,11 @@ def test_defaults(self): ) = get_parameters({}) self.assertTrue(dry_run) self.assertIsNone(filter_after_in_days) + self.assertIsNone(filter_downloaded_after) + self.assertIsNone(filter_downloaded_before) self.assertIsNone(filter_statuses) self.assertIsNone(filter_op_statuses) + self.assertIsNone(filter_validator_version_prefix) self.assertFalse(prod_env) self.assertEqual(validator_endpoint, GTFS_VALIDATOR_URL_STAGING) self.assertFalse(bypass_db_update) @@ -56,8 +62,11 @@ def test_all_params(self): payload = { "dry_run": False, "filter_after_in_days": 30, + "filter_downloaded_after": "2026-03-01", + "filter_downloaded_before": "2026-06-01", "filter_statuses": ["active"], "filter_op_statuses": ["published", "unpublished"], + "filter_validator_version_prefix": "8.", "validator_endpoint": "https://staging.example.com/api", "bypass_db_update": True, "force_update": True, @@ -67,8 +76,11 @@ def test_all_params(self): ( dry_run, filter_after_in_days, + filter_downloaded_after, + filter_downloaded_before, filter_statuses, filter_op_statuses, + filter_validator_version_prefix, prod_env, validator_endpoint, bypass_db_update, @@ -78,8 +90,11 @@ def test_all_params(self): ) = get_parameters(payload) self.assertFalse(dry_run) self.assertEqual(filter_after_in_days, 30) + self.assertEqual(filter_downloaded_after, "2026-03-01") + self.assertEqual(filter_downloaded_before, "2026-06-01") self.assertEqual(filter_statuses, ["active"]) self.assertEqual(filter_op_statuses, ["published", "unpublished"]) + self.assertEqual(filter_validator_version_prefix, "8.") self.assertEqual(validator_endpoint, "https://staging.example.com/api") self.assertTrue(bypass_db_update) self.assertTrue(force_update) @@ -100,6 +115,9 @@ def test_string_coercion(self): _, _, _, + _, + _, + _, bypass_db_update, force_update, limit, @@ -278,6 +296,9 @@ def test_handler_passes_all_params(self, rebuild_mock): payload = { "dry_run": False, "filter_after_in_days": 30, + "filter_downloaded_after": "2026-03-01", + "filter_downloaded_before": "2026-06-01", + "filter_validator_version_prefix": "8.", "validator_endpoint": "https://staging.example.com/api", "force_update": True, "limit": 10, @@ -290,8 +311,11 @@ def test_handler_passes_all_params(self, rebuild_mock): bypass_db_update=False, dry_run=False, filter_after_in_days=30, + filter_downloaded_after="2026-03-01", + filter_downloaded_before="2026-06-01", filter_statuses=None, filter_op_statuses=["published", "wip"], + filter_validator_version_prefix="8.", prod_env=False, force_update=True, limit=10, @@ -316,6 +340,67 @@ def test_default_op_status_filters_published( # Verify via the query mock that .filter was called (default published applied) self.assertTrue(session.query.called) + @patch(f"{_MODULE}._get_validator_version", return_value="8.0.1") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_version_prefix_filter_uses_exists_not_outerjoin( + self, tracker_cls, filter_blob_mock, version_mock + ): + """When filter_validator_version_prefix is set, the candidate query must use + the EXISTS branch (no outerjoin) and surface the prefix in the result params.""" + session = self._make_session_mock(datasets=[]) + query_mock = session.query.return_value + + result = rebuild_missing_validation_reports( + validator_endpoint="https://prod.example.com/api", + dry_run=True, + filter_validator_version_prefix="8.", + db_session=session, + ) + + query_mock.outerjoin.assert_not_called() + self.assertEqual(result["params"]["filter_validator_version_prefix"], "8.") + + @patch(f"{_MODULE}._get_validator_version", return_value="8.0.1") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_default_version_matching_uses_outerjoin( + self, tracker_cls, filter_blob_mock, version_mock + ): + """Without a version prefix, the query falls back to the exact-version + comparison via an outerjoin on validation reports.""" + session = self._make_session_mock(datasets=[]) + query_mock = session.query.return_value + + rebuild_missing_validation_reports( + validator_endpoint="https://prod.example.com/api", + dry_run=True, + db_session=session, + ) + + query_mock.outerjoin.assert_called_once() + + @patch(f"{_MODULE}._get_validator_version", return_value="8.0.1") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_downloaded_date_range_filters_surfaced_in_params( + self, tracker_cls, filter_blob_mock, version_mock + ): + """The downloaded_at date-window filters should be threaded through and + reflected in the returned params.""" + session = self._make_session_mock(datasets=[]) + + result = rebuild_missing_validation_reports( + validator_endpoint="https://prod.example.com/api", + dry_run=True, + filter_downloaded_after="2026-03-01", + filter_downloaded_before="2026-06-01", + db_session=session, + ) + + self.assertEqual(result["params"]["filter_downloaded_after"], "2026-03-01") + self.assertEqual(result["params"]["filter_downloaded_before"], "2026-06-01") + class TestFilterDatasetsWithExistingBlob(unittest.TestCase): @patch(f"{_MODULE}.storage") From d79a557b67ab51fb97fca9aaa2a7448a3db14896 Mon Sep 17 00:00:00 2001 From: cka-y Date: Thu, 23 Jul 2026 10:41:28 -0400 Subject: [PATCH 2/3] fix: readme --- .../src/tasks/validation_reports/README.md | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/README.md b/functions-python/tasks_executor/src/tasks/validation_reports/README.md index be53d51a1..577fef816 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/README.md +++ b/functions-python/tasks_executor/src/tasks/validation_reports/README.md @@ -285,14 +285,10 @@ curl -X POST "https://ingest-data-to-big-query-gtfs-563580583640.northamerica-no --- -## Re-validate a date window missing a major version (prod) +## Re-validate a date window missing a major version -Use this when a validator fix means a range of historical datasets needs a fresh report -from the **production** validator — e.g. re-running validator `8.x` on datasets published -between March and May 2026 that never got an `8.*` report. - -Run this from the **prod-deployed** function so `ENVIRONMENT=prod` makes the prod validator -URL and results bucket the defaults. +Use this when a validator fix means a range of historical datasets needs a fresh report (e.g. re-running validator `8.x` +on datasets published between March and May 2026 that never got an `8.*` report). ### Step 1 — Dry run (estimate scope) From b891929efbfe6e58ebc8fa436925a678dd7ed250 Mon Sep 17 00:00:00 2001 From: cka-y Date: Thu, 23 Jul 2026 11:01:32 -0400 Subject: [PATCH 3/3] fix: get all datasets --- .../src/tasks/validation_reports/README.md | 50 +++--------------- .../rebuild_missing_validation_reports.py | 38 +++++++++++--- ...test_rebuild_missing_validation_reports.py | 51 +++++++++++++++++++ 3 files changed, 89 insertions(+), 50 deletions(-) diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/README.md b/functions-python/tasks_executor/src/tasks/validation_reports/README.md index 577fef816..0616a1322 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/README.md +++ b/functions-python/tasks_executor/src/tasks/validation_reports/README.md @@ -69,6 +69,11 @@ datasets that lack **any** report matching that major-version prefix — useful want to re-validate everything missing an `8.*` report regardless of the exact patch version the endpoint reports. +By default the task only considers **each feed's latest dataset**. Set +`include_all_datasets: true` to consider **every** dataset of the feed — needed when +backfilling a historical window (e.g. so a feed's six-month reliability window is +complete), not just its most recent dataset. + The task is **resumable**: if it times out mid-loop, calling it again skips datasets that were already triggered (tracked in `task_execution_log`). @@ -102,6 +107,7 @@ that were already triggered (tracked in `task_execution_log`). | `filter_statuses` | list[str] | `null` | Filter feeds by status (e.g. `["active", "inactive"]`). Omit for all statuses | | `filter_op_statuses` | list[str] | `["published"]` | Filter feeds by operational status. Accepted values: `"published"`, `"unpublished"`, `"wip"` | | `filter_validator_version_prefix` | string | `null` | When set (e.g. `"8."`), only include datasets that do **not** already have a report whose `validator_version` starts with this prefix. Overrides the default exact-version comparison | +| `include_all_datasets` | bool | `false` | When `true`, consider **all** datasets of each feed, not just the latest one. Use for backfilling a historical window | | `force_update` | bool | `false` | Re-trigger even when a current report already exists | | `limit` | int | `null` | Cap the number of workflows triggered per call — useful for end-to-end testing | | `reports_bucket_name` | string | env-derived | Override the GCS bucket where validator results are stored. Use when running in prod but pointing to the staging validator (e.g. `"stg-gtfs-validator-results"`) | @@ -285,50 +291,6 @@ curl -X POST "https://ingest-data-to-big-query-gtfs-563580583640.northamerica-no --- -## Re-validate a date window missing a major version - -Use this when a validator fix means a range of historical datasets needs a fresh report (e.g. re-running validator `8.x` -on datasets published between March and May 2026 that never got an `8.*` report). - -### Step 1 — Dry run (estimate scope) - -```json -{ - "task": "rebuild_missing_validation_reports", - "payload": { - "dry_run": true, - "filter_downloaded_after": "2026-03-01", - "filter_downloaded_before": "2026-06-01", - "filter_validator_version_prefix": "8." - } -} -``` - -Check `total_candidates` in the response. `filter_validator_version_prefix: "8."` excludes -datasets that already have any `8.*` report, so re-running is idempotent. - -### Step 2 — Full run - -Flip `dry_run` to `false` (optionally add `"limit": 10` first for an end-to-end test): - -```json -{ - "task": "rebuild_missing_validation_reports", - "payload": { - "dry_run": false, - "filter_downloaded_after": "2026-03-01", - "filter_downloaded_before": "2026-06-01", - "filter_validator_version_prefix": "8." - } -} -``` - -`sync_task_run_status` then polls in the background until `ready_for_bigquery: true` (see -above). Because `bypass_db_update` is omitted (defaults to `false`), the new reports are -written to the DB/API. - ---- - ## GCP Environment Variables | Variable | Default | Description | diff --git a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py index 53e0e44dd..8f7203c9c 100644 --- a/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py @@ -71,6 +71,8 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: # have a validation report whose validator_version starts with this # prefix (e.g. "8." → datasets missing any 8.* report). When omitted, # datasets are selected by comparing against the exact target version. + "include_all_datasets": bool,# [optional] If True, consider ALL datasets of each feed, not just the + # latest one. Use for backfilling historical windows. Default: False "validator_endpoint": str, # [optional] Override validator URL (e.g. staging). Default: env-derived URL. "bypass_db_update": bool, # [optional] If True, results are NOT written to the DB/API (pre-release runs). Default: False @@ -89,6 +91,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: filter_statuses, filter_op_statuses, filter_validator_version_prefix, + include_all_datasets, prod_env, validator_endpoint, bypass_db_update, @@ -107,6 +110,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict: filter_statuses=filter_statuses, filter_op_statuses=filter_op_statuses, filter_validator_version_prefix=filter_validator_version_prefix, + include_all_datasets=include_all_datasets, prod_env=prod_env, force_update=force_update, limit=limit, @@ -125,6 +129,7 @@ def rebuild_missing_validation_reports( filter_statuses: List[str] | None = None, filter_op_statuses: List[str] | None = None, filter_validator_version_prefix: Optional[str] = None, + include_all_datasets: bool = False, prod_env: bool = False, force_update: bool = False, limit: Optional[int] = None, @@ -152,6 +157,8 @@ def rebuild_missing_validation_reports( NOT already have a validation report whose validator_version starts with this prefix. When None (default), datasets are selected by comparing against the exact target validator version. + include_all_datasets: If True, consider ALL datasets of each feed, not just the + latest one. Use for backfilling a historical window. Default: False. prod_env: True if targeting the production environment. Default: False force_update: Re-trigger even if a report already exists. Default: False limit: Max datasets to trigger per call (for end-to-end testing). Default: unlimited @@ -178,6 +185,7 @@ def rebuild_missing_validation_reports( filter_op_statuses if filter_op_statuses is not None else ["published"] ), filter_validator_version_prefix=filter_validator_version_prefix, + include_all_datasets=include_all_datasets, ) total_candidates = len(datasets) logging.info("Found %s candidate datasets", total_candidates) @@ -259,6 +267,7 @@ def rebuild_missing_validation_reports( "filter_downloaded_before": filter_downloaded_before, "filter_statuses": filter_statuses, "filter_validator_version_prefix": filter_validator_version_prefix, + "include_all_datasets": include_all_datasets, "prod_env": prod_env, "force_update": force_update, "limit": limit, @@ -289,6 +298,7 @@ def _get_datasets_for_validation( filter_statuses: Optional[List[str]], filter_op_statuses: Optional[List[str]], filter_validator_version_prefix: Optional[str], + include_all_datasets: bool = False, ) -> List[tuple]: """ Query datasets that need a (re)validation. @@ -298,6 +308,11 @@ def _get_datasets_for_validation( - Have a report from a different (older) validator version, OR - force_update is True + By default only each feed's latest dataset is considered. When + include_all_datasets is True, every dataset of the feed is considered — use + this for backfilling a historical window (e.g. re-validating all datasets + published in a date range so the six-month reliability window is complete). + When filter_validator_version_prefix is set (e.g. "8."), the exact-version comparison is replaced by "the dataset does not already have any validation report whose validator_version starts with the prefix". This is what lets us @@ -310,11 +325,13 @@ def _get_datasets_for_validation( When all age filters are None, all datasets are included regardless of age. filter_op_statuses filters by Feed.operational_status (e.g. ["published"]). """ - query = ( - db_session.query(Gtfsfeed.stable_id, Gtfsdataset.stable_id) - .select_from(Gtfsfeed) - .join(Gtfsdataset, Gtfsfeed.latest_dataset_id == Gtfsdataset.id) + query = db_session.query(Gtfsfeed.stable_id, Gtfsdataset.stable_id).select_from( + Gtfsfeed ) + if include_all_datasets: + query = query.join(Gtfsdataset, Gtfsdataset.feed_id == Gtfsfeed.id) + else: + query = query.join(Gtfsdataset, Gtfsfeed.latest_dataset_id == Gtfsdataset.id) if filter_validator_version_prefix: # Include datasets that do NOT already have a report matching the version @@ -400,8 +417,9 @@ def get_parameters(payload): Returns: Tuple of (dry_run, filter_after_in_days, filter_downloaded_after, filter_downloaded_before, filter_statuses, filter_op_statuses, - filter_validator_version_prefix, prod_env, validator_endpoint, - bypass_db_update, force_update, limit, reports_bucket_name) + filter_validator_version_prefix, include_all_datasets, prod_env, + validator_endpoint, bypass_db_update, force_update, limit, + reports_bucket_name) """ prod_env = os.getenv("ENVIRONMENT", "").lower() == "prod" default_endpoint = get_gtfs_validator_url(prod_env) @@ -428,6 +446,13 @@ def get_parameters(payload): "filter_validator_version_prefix", None ) + include_all_datasets = payload.get("include_all_datasets", False) + include_all_datasets = ( + include_all_datasets + if isinstance(include_all_datasets, bool) + else str(include_all_datasets).lower() == "true" + ) + validator_endpoint = payload.get("validator_endpoint", default_endpoint) bypass_db_update = payload.get("bypass_db_update", False) @@ -458,6 +483,7 @@ def get_parameters(payload): filter_statuses, filter_op_statuses, filter_validator_version_prefix, + include_all_datasets, prod_env, validator_endpoint, bypass_db_update, diff --git a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py index df83fda71..2bb8c8ca3 100644 --- a/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py +++ b/functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py @@ -37,6 +37,7 @@ def test_defaults(self): filter_statuses, filter_op_statuses, filter_validator_version_prefix, + include_all_datasets, prod_env, validator_endpoint, bypass_db_update, @@ -51,6 +52,7 @@ def test_defaults(self): self.assertIsNone(filter_statuses) self.assertIsNone(filter_op_statuses) self.assertIsNone(filter_validator_version_prefix) + self.assertFalse(include_all_datasets) self.assertFalse(prod_env) self.assertEqual(validator_endpoint, GTFS_VALIDATOR_URL_STAGING) self.assertFalse(bypass_db_update) @@ -67,6 +69,7 @@ def test_all_params(self): "filter_statuses": ["active"], "filter_op_statuses": ["published", "unpublished"], "filter_validator_version_prefix": "8.", + "include_all_datasets": True, "validator_endpoint": "https://staging.example.com/api", "bypass_db_update": True, "force_update": True, @@ -81,6 +84,7 @@ def test_all_params(self): filter_statuses, filter_op_statuses, filter_validator_version_prefix, + include_all_datasets, prod_env, validator_endpoint, bypass_db_update, @@ -95,6 +99,7 @@ def test_all_params(self): self.assertEqual(filter_statuses, ["active"]) self.assertEqual(filter_op_statuses, ["published", "unpublished"]) self.assertEqual(filter_validator_version_prefix, "8.") + self.assertTrue(include_all_datasets) self.assertEqual(validator_endpoint, "https://staging.example.com/api") self.assertTrue(bypass_db_update) self.assertTrue(force_update) @@ -118,6 +123,7 @@ def test_string_coercion(self): _, _, _, + _, bypass_db_update, force_update, limit, @@ -299,6 +305,7 @@ def test_handler_passes_all_params(self, rebuild_mock): "filter_downloaded_after": "2026-03-01", "filter_downloaded_before": "2026-06-01", "filter_validator_version_prefix": "8.", + "include_all_datasets": True, "validator_endpoint": "https://staging.example.com/api", "force_update": True, "limit": 10, @@ -316,6 +323,7 @@ def test_handler_passes_all_params(self, rebuild_mock): filter_statuses=None, filter_op_statuses=["published", "wip"], filter_validator_version_prefix="8.", + include_all_datasets=True, prod_env=False, force_update=True, limit=10, @@ -401,6 +409,49 @@ def test_downloaded_date_range_filters_surfaced_in_params( self.assertEqual(result["params"]["filter_downloaded_after"], "2026-03-01") self.assertEqual(result["params"]["filter_downloaded_before"], "2026-06-01") + @patch(f"{_MODULE}._get_validator_version", return_value="8.0.1") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_include_all_datasets_joins_on_feed_id( + self, tracker_cls, filter_blob_mock, version_mock + ): + """With include_all_datasets=True, the dataset join must be on feed_id + (all datasets of the feed), not latest_dataset_id.""" + session = self._make_session_mock(datasets=[]) + query_mock = session.query.return_value + + result = rebuild_missing_validation_reports( + validator_endpoint="https://prod.example.com/api", + dry_run=True, + include_all_datasets=True, + db_session=session, + ) + + join_onclause = str(query_mock.join.call_args[0][1]) + self.assertIn("feed_id", join_onclause) + self.assertNotIn("latest_dataset_id", join_onclause) + self.assertTrue(result["params"]["include_all_datasets"]) + + @patch(f"{_MODULE}._get_validator_version", return_value="8.0.1") + @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[]) + @patch(f"{_MODULE}.TaskExecutionTracker") + def test_latest_only_joins_on_latest_dataset_id( + self, tracker_cls, filter_blob_mock, version_mock + ): + """By default (include_all_datasets=False), the dataset join is on + latest_dataset_id — only each feed's latest dataset is considered.""" + session = self._make_session_mock(datasets=[]) + query_mock = session.query.return_value + + rebuild_missing_validation_reports( + validator_endpoint="https://prod.example.com/api", + dry_run=True, + db_session=session, + ) + + join_onclause = str(query_mock.join.call_args[0][1]) + self.assertIn("latest_dataset_id", join_onclause) + class TestFilterDatasetsWithExistingBlob(unittest.TestCase): @patch(f"{_MODULE}.storage")