diff --git a/pipeline/workflow/ingestion-helper/aggregation.yaml b/pipeline/workflow/ingestion-helper/aggregation.yaml new file mode 100644 index 000000000..b8786aa51 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation.yaml @@ -0,0 +1,17 @@ +# Data Commons Aggregation Configuration. See the README for details. + +aggregations: + # Generates linkedContainedInPlace, linkedMemberOf, etc. + - type: linked_edges + imports: ["*"] + stage: 1 + + # Generates summary statistics in the Cache table + - type: provenance_summary + imports: ["*"] + stage: 1 + + # Generates the Statistical Variable hierarchy/verticals + - type: stat_var_groups + imports: ["*"] + stage: 1 diff --git a/pipeline/workflow/ingestion-helper/aggregation/README.md b/pipeline/workflow/ingestion-helper/aggregation/README.md new file mode 100644 index 000000000..ec61e5a40 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/README.md @@ -0,0 +1,147 @@ +# Aggregations + +This module orchestrates the execution of Data Commons aggregations through BigQuery Federation. The aggregations include place rollups, statistical variable aggregations, linked edges, and metadata summaries. + +## Core Concepts + +* **Sequential Stages**: Aggregations are executed sequentially by their `stage` number (e.g., Stage 1 steps are guaranteed to complete before Stage 2 steps begin). This is useful when later steps depend on the output of earlier ones. +* **Parallel Execution**: All aggregation steps configured in the same stage are executed in parallel to maximize performance. + +--- + +## Configuration Guide (`aggregation.yaml`) + +The entire aggregation pipeline is configured via `aggregation.yaml`. This file defines which aggregations run, what their dependencies are, and in what order they execute. + +### Common Configuration Fields +Every step in the configuration supports these common fields: +* `type` (string, Required): The type of aggregation step to run. +* `stage` (integer, Optional, default: 1): The sequential stage number. Steps in lower stages are guaranteed to finish before higher stages start. +* `imports` (list of strings, Required): The list of import names this step applies to. Use `["*"]` (wildcard) to apply the step to **all** imports in the current run. +* `disabled` (boolean, Optional, default: false): Set to `true` to temporarily disable a step without deleting it. + +--- + +### Supported Aggregation Types + +#### 1. Place (`place`) +Aggregates and rolls up statistical data from a smaller place type (source) to a larger place type (destination). +* **Fields**: + * `source_type` (string, Required): The source place type (e.g., `County`). + * `destination_type` (string, Required): The destination place type (e.g., `State`). + * `allow_multiple_to_places` (boolean, Optional, default: false): Allows mapping to multiple parent places if true. +* **Example**: + ```yaml + - type: place + stage: 1 + imports: ["USFed_Census"] + source_type: County + destination_type: State + ``` + +#### 2. Statistical Variable Aggregation (`stat_var`) +Aggregates raw statistical variables into a summarized ancestor variable (e.g., summing up individual age group counts to get a total population count). +* **Fields**: + * `ancestor_sv_id` (string, Required): The ID of the parent/summary statistical variable (e.g., `Count_Person`). + * `source_sv_ids` (list of strings, Required): The list of individual statistical variables to sum up. + * `output_import_name` (string, Optional): Custom import name to write output under. + * `skip_all_sources_present_check` (boolean, Optional, default: false): If true, aggregates even if some source variables are missing. +* **Example**: + ```yaml + - type: stat_var + stage: 2 + imports: ["USFed_Census"] + ancestor_sv_id: Count_Person + source_sv_ids: + - Count_Person_Male + - Count_Person_Female + ``` + +#### 3. Linked Edges (`linked_edges`) +Constructs and aggregates structural graph links (edges) between nodes in the Data Commons graph. +* **Example**: + ```yaml + - type: linked_edges + stage: 1 + imports: ["*"] # Runs for all imports + ``` + +#### 4. Provenance Summary (`provenance_summary`) +Generates metadata and provenance summaries for all aggregated statistical observations, establishing data lineage. +* **Example**: + ```yaml + - type: provenance_summary + stage: 3 + imports: ["USFed_Census"] + ``` + +#### 5. Statistical Variable Groups (`stat_var_groups`) +Aggregates and structures statistical variables into hierarchical groups for display in the Data Commons UI. +* **Example**: + ```yaml + - type: stat_var_groups + stage: 3 + imports: ["*"] + ``` + +--- + +### Example `aggregation.yaml` + +This example demonstrates a typical multi-stage aggregation workflow. + +```yaml +# aggregation.yaml +aggregations: + # Stage 1: Parallel Place Rollups and Linked Edges + - type: linked_edges + stage: 1 + imports: ["*"] + + - type: place + stage: 1 + imports: ["USFed_Census"] + source_type: County + destination_type: State + + # Stage 2: Parallel Stat Var Aggregations (Depends on Stage 1 completing) + - type: stat_var + stage: 2 + imports: ["USFed_Census"] + ancestor_sv_id: Count_Person + source_sv_ids: + - Count_Person_Male + - Count_Person_Female + + # Stage 3: Metadata and UI Summaries (Depends on Stage 2 completing) + - type: provenance_summary + stage: 3 + imports: ["USFed_Census"] + + - type: stat_var_groups + stage: 3 + imports: ["*"] +``` + +--- + +## Local Configuration Validation + +The orchestrator strictly validates the `aggregation.yaml` file on startup against a strict JSON Schema (`schema.json`). If there is any syntax error, type mismatch, or missing required field, the service will fail to start. + +### Running the Validator Locally +You can validate your `aggregation.yaml` file locally using the built-in CLI tool before committing or deploying changes. + +1. **Navigate to the ingestion-helper root**: + ```bash + cd pipeline/workflow/ingestion-helper + ``` +2. **Run the validator**: + ```bash + python3 -m aggregation.validator --config ../aggregation.yaml + + # sample output... + # Validating 'aggregation.yaml' against 'schema.json'... + # [SUCCESS] Configuration is valid! + # Parsed 5 aggregation steps successfully. + ``` diff --git a/pipeline/workflow/ingestion-helper/aggregation/__init__.py b/pipeline/workflow/ingestion-helper/aggregation/__init__.py index 0b76748ab..5fedbd914 100644 --- a/pipeline/workflow/ingestion-helper/aggregation/__init__.py +++ b/pipeline/workflow/ingestion-helper/aggregation/__init__.py @@ -23,6 +23,8 @@ from .stat_var_aggregator import StatVarAggregator from .place_aggregation_generator import PlaceAggregationGenerator from .stat_var_group_generator import StatVarGroupGenerator +from .orchestrator import AggregationOrchestrator +from .validator import validate_config __all__ = [ 'BigQueryExecutor', @@ -30,5 +32,7 @@ 'ProvenanceSummaryGenerator', 'StatVarAggregator', 'PlaceAggregationGenerator', - 'StatVarGroupGenerator' + 'StatVarGroupGenerator', + 'AggregationOrchestrator', + 'validate_config' ] diff --git a/pipeline/workflow/ingestion-helper/aggregation/orchestrator.py b/pipeline/workflow/ingestion-helper/aggregation/orchestrator.py new file mode 100644 index 000000000..4ef337aa8 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/orchestrator.py @@ -0,0 +1,245 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import os +from typing import Any, Dict, List, Optional + +from .bq_executor import BigQueryExecutor +from .linked_edge_generator import LinkedEdgeGenerator +from .provenance_summary_generator import ProvenanceSummaryGenerator +from .stat_var_aggregator import StatVarAggregator +from .place_aggregation_generator import PlaceAggregationGenerator +from .stat_var_group_generator import StatVarGroupGenerator +from .validator import validate_config + + +class AggregationOrchestrator: + """Orchestrates the overall aggregation workflow.""" + + def __init__(self, + connection_id: str, + project_id: str, + instance_id: str, + database_id: str, + location: Optional[str] = None, + is_base_dc: bool = True, + config_file_path: Optional[str] = None) -> None: + """Initializes the orchestrator and loads/validates the configuration. + + Args: + connection_id: BigQuery connection ID to Spanner. + project_id: GCP Project ID. + instance_id: Spanner Instance ID. + database_id: Spanner Database ID. + location: BigQuery location. + is_base_dc: Whether this is running in the base Data Commons environment. + config_file_path: Optional custom path to the aggregation.yaml file. + If not specified, defaults to the aggregation.yaml in the parent directory. + """ + self.executor = BigQueryExecutor(connection_id=connection_id, + project_id=project_id, + instance_id=instance_id, + database_id=database_id, + location=location, + run_sequential=False) + + self.is_base_dc = is_base_dc + + # Resolve paths for default config and schema + curr_dir = os.path.dirname(os.path.abspath(__file__)) + if not config_file_path: + config_file_path = os.path.join(curr_dir, "..", "aggregation.yaml") + schema_file_path = os.path.join(curr_dir, "schema.json") + + # Load and validate configuration + self.aggregations = validate_config(config_file_path, schema_file_path) + + def execute_stage(self, stage_num: int, active_imports: List[str]) -> List[str]: + """Executes all enabled aggregations in the specified stage in parallel. + + Args: + stage_num: The stage number to execute. + active_imports: The list of active import names in this run. + + Returns: + A list of BigQuery job IDs submitted for this stage. + """ + logging.info(f"Starting Aggregation Orchestration for Stage {stage_num}") + logging.info(f"Active imports in this run: {active_imports}") + jobs = [] + + for config in self.aggregations: + # 1. Skip if disabled + if config.get("disabled", False): + continue + + # 2. Filter by stage + if config.get("stage", 1) != stage_num: + continue + + # 3. Filter by active imports + applicable_imports = self._get_applicable_imports(config, active_imports) + if not applicable_imports: + continue + + # 4. Route to correct generator helper + step_type = config["type"] + logging.info(f"Triggering step '{step_type}' in Stage {stage_num}...") + + step_jobs = [] + if step_type == "place": + step_jobs = self._trigger_place(config, applicable_imports) + elif step_type == "stat_var": + step_jobs = self._trigger_stat_var(config, applicable_imports) + elif step_type == "linked_edges": + step_jobs = self._trigger_linked_edges(config, applicable_imports) + elif step_type == "provenance_summary": + step_jobs = self._trigger_provenance_summary(config, applicable_imports) + elif step_type == "stat_var_groups": + step_jobs = self._trigger_stat_var_groups(config, applicable_imports) + else: + raise ValueError(f"Unsupported or unimplemented aggregation step type: {step_type}") + + # Collect BQ jobs + for job in step_jobs: + if job and job.job_id: + jobs.append(job.job_id) + + logging.info(f"=== Stage {stage_num} initiated successfully. Submitted {len(jobs)} BigQuery jobs: {jobs} ===") + return jobs + + def has_stage(self, stage_num: int, active_imports: List[str]) -> bool: + """Checks if there are any active, enabled aggregations configured for the stage. + + Args: + stage_num: The stage number to check. + active_imports: The list of active import names. + + Returns: + True if the stage has at least one aggregation that will run, False otherwise. + """ + for config in self.aggregations: + if config.get("disabled", False): + continue + if config.get("stage", 1) != stage_num: + continue + + # Check if it applies to any active imports + if self._get_applicable_imports(config, active_imports): + return True + + return False + + def get_active_stages(self, active_imports: List[str]) -> List[int]: + """Returns a sorted list of unique, active, and enabled stage numbers. + + Args: + active_imports: The list of active import names. + + Returns: + A sorted list of unique active stage numbers. + """ + stages = set() + for config in self.aggregations: + step_type = config.get("type") + stage_num = config.get("stage", 1) + + if config.get("disabled", False): + logging.info(f"[Config Scan] Skipping step '{step_type}' in Stage {stage_num} because it is disabled.") + continue + + applicable_imports = self._get_applicable_imports(config, active_imports) + if not applicable_imports: + logging.info(f"[Config Scan] Skipping step '{step_type}' in Stage {stage_num} because it does not apply to active imports: {active_imports}.") + continue + + logging.info(f"[Config Scan] Step '{step_type}' in Stage {stage_num} is ACTIVE for imports: {applicable_imports}.") + stages.add(stage_num) + + sorted_stages = sorted(list(stages)) + logging.info(f"[Config Scan] Active stages resolved: {sorted_stages}") + return sorted_stages + + def check_jobs_status(self, job_ids: List[str]) -> Dict[str, Any]: + """Checks the status of the specified BigQuery job IDs. + + Delegates to the BigQueryExecutor's get_jobs_status. + """ + try: + return self.executor.get_jobs_status(job_ids) + except Exception as e: + logging.error(f"Failed to check jobs status: {e}") + raise e + + def _trigger_place(self, config: Dict[str, Any], applicable_imports: List[str]) -> List[Any]: + """Triggers place-level rollup aggregations.""" + source_type = config["source_type"] + destination_type = config["destination_type"] + logging.info( + f" -> Place Rollup: {source_type} -> {destination_type} for imports {applicable_imports}" + ) + generator = PlaceAggregationGenerator(self.executor, self.is_base_dc) + job = generator.aggregate_places( + import_names=applicable_imports, + source_type=source_type, + destination_type=destination_type, + allow_multiple_to_places=config.get("allow_multiple_to_places", False) + ) + return [job] if job else [] + + def _trigger_stat_var(self, config: Dict[str, Any], applicable_imports: List[str]) -> List[Any]: + """Triggers statistical variable aggregations.""" + ancestor_sv = config["ancestor_sv_id"] + source_svs = config["source_sv_ids"] + logging.info( + f" -> Stat Var Aggregation: ancestor '{ancestor_sv}' (sources: {source_svs}) for imports {applicable_imports}" + ) + generator = StatVarAggregator(self.executor, self.is_base_dc) + return generator.aggregate_stat_vars( + ancestor_sv=ancestor_sv, + source_svs=source_svs, + import_names=applicable_imports, + output_import_name=config.get("output_import_name"), + skip_all_sources_present_check=config.get("skip_all_sources_present_check", False) + ) + + def _trigger_linked_edges(self, config: Dict[str, Any], applicable_imports: List[str]) -> List[Any]: + """Triggers linked edge aggregations.""" + logging.info(f" -> Linked Edges Aggregation for imports {applicable_imports}") + generator = LinkedEdgeGenerator(self.executor, self.is_base_dc) + return generator.run_all(applicable_imports) + + def _trigger_provenance_summary(self, config: Dict[str, Any], applicable_imports: List[str]) -> List[Any]: + """Triggers provenance summary aggregations.""" + logging.info(f" -> Provenance Summary Aggregation for imports {applicable_imports}") + generator = ProvenanceSummaryGenerator(self.executor, self.is_base_dc) + return generator.run_all(applicable_imports) + + def _trigger_stat_var_groups(self, config: Dict[str, Any], applicable_imports: List[str]) -> List[Any]: + """Triggers statistical variable group aggregations.""" + logging.info(f" -> Stat Var Groups Aggregation for imports {applicable_imports}") + generator = StatVarGroupGenerator(self.executor, self.is_base_dc) + return generator.run_all(applicable_imports) + + def _get_applicable_imports(self, config: Dict[str, Any], active_imports: List[str]) -> List[str]: + """Determines which active imports apply to this aggregation config.""" + configured_imports = config["imports"] + + # Explicit wildcard check + if "*" in configured_imports: + return active_imports + + # Intersection of configured and active imports + return list(set(configured_imports).intersection(active_imports)) diff --git a/pipeline/workflow/ingestion-helper/aggregation/orchestrator_test.py b/pipeline/workflow/ingestion-helper/aggregation/orchestrator_test.py new file mode 100644 index 000000000..a0b2627f4 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/orchestrator_test.py @@ -0,0 +1,208 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the AggregationOrchestrator class.""" + +import json +import os +import sys +import tempfile +import textwrap +import unittest +from unittest.mock import MagicMock, patch + +sys.path.append(os.path.dirname(os.path.dirname(__file__))) + +from aggregation import AggregationOrchestrator + +VALID_CONFIG_YAML = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: ["*"] + stage: 1 + + - type: place + source_type: County + destination_type: State + allow_multiple_to_places: false + imports: ["USFed_Census"] + stage: 1 + + - type: place + source_type: State + destination_type: Country + imports: ["*"] + stage: 2 + disabled: true + + - type: stat_var + ancestor_sv_id: Count_Person + source_sv_ids: ["Count_Person_Male", "Count_Person_Female"] + skip_all_sources_present_check: true + imports: ["USFed_Census"] + stage: 2 +""") + + +@patch('aggregation.orchestrator.BigQueryExecutor') +class TestOrchestratorScanning(unittest.TestCase): + """Tests the stage scanning and active stage resolution methods.""" + + def setUp(self): + self.tmpdir = tempfile.TemporaryDirectory() + config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + with open(config_path, "w") as f: + f.write(VALID_CONFIG_YAML) + + self.orchestrator = AggregationOrchestrator( + connection_id="conn", + project_id="proj", + instance_id="inst", + database_id="db", + config_file_path=config_path + ) + + def tearDown(self): + self.tmpdir.cleanup() + + def test_has_stage(self, mock_executor): + """Tests the has_stage method for active, disabled, and non-matching stages.""" + self.assertTrue(self.orchestrator.has_stage(1, ["AnyImport"])) + self.assertTrue(self.orchestrator.has_stage(1, ["USFed_Census"])) + + self.assertFalse(self.orchestrator.has_stage(2, ["OtherImport"])) + self.assertTrue(self.orchestrator.has_stage(2, ["USFed_Census"])) + + self.assertFalse(self.orchestrator.has_stage(3, ["USFed_Census"])) + + def test_get_active_stages(self, mock_executor): + """Tests that get_active_stages correctly extracts, filters, and sorts active stages.""" + stages = self.orchestrator.get_active_stages(active_imports=["USFed_Census"]) + self.assertEqual(stages, [1, 2]) + + stages = self.orchestrator.get_active_stages(active_imports=["OtherImport"]) + self.assertEqual(stages, [1]) + + +@patch('aggregation.orchestrator.BigQueryExecutor') +@patch('aggregation.orchestrator.PlaceAggregationGenerator') +@patch('aggregation.orchestrator.StatVarAggregator') +@patch('aggregation.orchestrator.LinkedEdgeGenerator') +@patch('aggregation.orchestrator.ProvenanceSummaryGenerator') +@patch('aggregation.orchestrator.StatVarGroupGenerator') +class TestOrchestratorExecution(unittest.TestCase): + """Tests stage execution, verifying parallel job submission and routing. + + These tests execute stages, so they mock the executor and all five generators + to verify correct parameters are passed and jobs are collected. + """ + + def setUp(self): + self.tmpdir = tempfile.TemporaryDirectory() + config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + with open(config_path, "w") as f: + f.write(VALID_CONFIG_YAML) + + self.orchestrator = AggregationOrchestrator( + connection_id="conn", + project_id="proj", + instance_id="inst", + database_id="db", + config_file_path=config_path + ) + + def tearDown(self): + self.tmpdir.cleanup() + + def test_execute_stage_1(self, mock_svg_gen, mock_prov_gen, mock_edge_gen, + mock_sv_agg, mock_place_gen, mock_executor): + """Tests executing Stage 1, verifying parallel job submission and wildcard resolution.""" + mock_job1 = MagicMock() + mock_job1.job_id = "job-edge-1" + mock_edge_gen.return_value.run_all.return_value = [mock_job1] + + mock_job2 = MagicMock() + mock_job2.job_id = "job-place-1" + mock_place_gen.return_value.aggregate_places.return_value = mock_job2 + + job_ids = self.orchestrator.execute_stage(stage_num=1, active_imports=["USFed_Census"]) + + self.assertEqual(len(job_ids), 2) + self.assertIn("job-edge-1", job_ids) + self.assertIn("job-place-1", job_ids) + + mock_edge_gen.return_value.run_all.assert_called_once_with(["USFed_Census"]) + + mock_place_gen.return_value.aggregate_places.assert_called_once_with( + import_names=["USFed_Census"], + source_type="County", + destination_type="State", + allow_multiple_to_places=False + ) + + def test_execute_stage_2_with_disabled_and_filtering(self, mock_svg_gen, mock_prov_gen, mock_edge_gen, + mock_sv_agg, mock_place_gen, mock_executor): + """Tests Stage 2, verifying that disabled steps are skipped and non-matching imports are filtered.""" + mock_job_sv = MagicMock() + mock_job_sv.job_id = "job-sv-1" + mock_sv_agg.return_value.aggregate_stat_vars.return_value = [mock_job_sv] + + job_ids = self.orchestrator.execute_stage(stage_num=2, active_imports=["OtherImport"]) + self.assertEqual(len(job_ids), 0) + mock_place_gen.return_value.aggregate_places.assert_not_called() + mock_sv_agg.return_value.aggregate_stat_vars.assert_not_called() + + job_ids = self.orchestrator.execute_stage(stage_num=2, active_imports=["USFed_Census"]) + + self.assertEqual(job_ids, ["job-sv-1"]) + mock_place_gen.return_value.aggregate_places.assert_not_called() + mock_sv_agg.return_value.aggregate_stat_vars.assert_called_once_with( + ancestor_sv="Count_Person", + source_svs=["Count_Person_Male", "Count_Person_Female"], + import_names=["USFed_Census"], + output_import_name=None, + skip_all_sources_present_check=True + ) + + def test_execute_stage_unsupported_type(self, *mocks): + """Tests that an unsupported aggregation step type raises ValueError.""" + unimplemented_config = textwrap.dedent("""\ + aggregations: + - type: entity + entity_types: ["MortalityEvent"] + location_props: ["location"] + imports: ["*"] + stage: 1 + """) + + with tempfile.TemporaryDirectory() as local_tmpdir: + local_config_path = os.path.join(local_tmpdir, "aggregation.yaml") + with open(local_config_path, "w") as f: + f.write(unimplemented_config) + + local_orchestrator = AggregationOrchestrator( + connection_id="conn", + project_id="proj", + instance_id="inst", + database_id="db", + config_file_path=local_config_path + ) + + with self.assertRaises(ValueError) as ctx: + local_orchestrator.execute_stage(stage_num=1, active_imports=["USFed_Census"]) + self.assertIn("Unsupported or unimplemented aggregation step type: entity", str(ctx.exception)) + + +if __name__ == '__main__': + unittest.main() diff --git a/pipeline/workflow/ingestion-helper/aggregation/schema.json b/pipeline/workflow/ingestion-helper/aggregation/schema.json new file mode 100644 index 000000000..b0deba8f6 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/schema.json @@ -0,0 +1,93 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "AggregationConfig", + "type": "object", + "properties": { + "aggregations": { + "type": "array", + "items": { + "type": "object", + "required": ["type", "imports"], + "properties": { + "type": { + "type": "string", + "enum": ["place", "stat_var", "entity", "linked_edges", "provenance_summary", "stat_var_groups"] + }, + "disabled": { + "type": "boolean", + "default": false + }, + "stage": { + "type": "integer", + "minimum": 1, + "default": 1 + }, + "imports": { + "type": "array", + "items": { "type": "string" }, + "minItems": 1 + } + }, + "dependencies": { + "type": { + "oneOf": [ + { + "properties": { + "type": { "const": "place" }, + "source_type": { "type": "string" }, + "destination_type": { "type": "string" }, + "allow_multiple_to_places": { "type": "boolean" } + }, + "required": ["source_type", "destination_type"] + }, + { + "properties": { + "type": { "const": "stat_var" }, + "ancestor_sv_id": { "type": "string" }, + "source_sv_ids": { + "type": "array", + "items": { "type": "string" }, + "minItems": 1 + }, + "skip_all_sources_present_check": { "type": "boolean" }, + "output_import_name": { "type": "string" } + }, + "required": ["ancestor_sv_id", "source_sv_ids"] + }, + { + "properties": { + "type": { "const": "entity" }, + "entity_types": { + "type": "array", + "items": { "type": "string" } + }, + "location_props": { + "type": "array", + "items": { "type": "string" } + }, + "date_prop": { "type": "string" }, + "agg_date_formats": { + "type": "array", + "items": { "type": "string" } + }, + "constraints": { + "type": "array", + "items": { "type": "string" } + } + }, + "required": ["entity_types", "location_props"] + }, + { + "properties": { + "type": { "enum": ["linked_edges", "provenance_summary", "stat_var_groups"] } + } + } + ] + } + } + } + } + }, + "required": ["aggregations"], + "additionalProperties": false +} diff --git a/pipeline/workflow/ingestion-helper/aggregation/validator.py b/pipeline/workflow/ingestion-helper/aggregation/validator.py new file mode 100644 index 000000000..c26427e8f --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/validator.py @@ -0,0 +1,130 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Configuration validator and CLI tool for Data Commons aggregations.""" + +import argparse +import json +import logging +import os +import sys +from typing import Any, Dict, List +import yaml +import jsonschema + +# ANSI escape codes for colored terminal output +GREEN = "\033[92m" +RED = "\033[91m" +RESET = "\033[0m" + + +def validate_config(config_file_path: str, schema_file_path: str) -> List[Dict[str, Any]]: + """Loads and validates the aggregation YAML configuration against the JSON Schema. + + Args: + config_file_path: Path to the aggregation.yaml configuration file. + schema_file_path: Path to the aggregation_schema.json validation file. + + Returns: + A list of validated aggregation dictionaries. + + Raises: + FileNotFoundError: If either the config or schema file is missing. + jsonschema.exceptions.ValidationError: If schema validation fails. + yaml.YAMLError: If the YAML file is malformed. + """ + if not os.path.exists(config_file_path): + raise FileNotFoundError(f"Aggregation config file not found: {config_file_path}") + if not os.path.exists(schema_file_path): + raise FileNotFoundError(f"JSON Schema file not found: {schema_file_path}") + + # 1. Load and parse YAML + try: + with open(config_file_path, "r") as f: + config = yaml.safe_load(f) + except yaml.YAMLError as e: + logging.error(f"Failed to parse YAML file {config_file_path}: {e}") + raise e + + if config is None: + config = {} + + # 2. Load JSON Schema + try: + with open(schema_file_path, "r") as f: + schema = json.load(f) + except Exception as e: + logging.error(f"Failed to load JSON Schema file {schema_file_path}: {e}") + raise e + + # 3. Validate against Schema + try: + jsonschema.validate(instance=config, schema=schema) + except jsonschema.exceptions.ValidationError as e: + logging.error(f"Schema validation failed for config {config_file_path}: {e.message}") + raise e + + return config["aggregations"] + + +def main(): + """CLI entry point for standalone configuration validation.""" + logging.basicConfig(level=logging.INFO) + parser = argparse.ArgumentParser(description="Validate Data Commons aggregation configuration files against the JSON Schema.") + + # Resolve default paths relative to this script's directory (aggregation/) + curr_dir = os.path.dirname(os.path.abspath(__file__)) + default_config = os.path.join(curr_dir, "..", "aggregation.yaml") + default_schema = os.path.join(curr_dir, "schema.json") + + parser.add_argument( + "--config", + type=str, + default=default_config, + help=f"Path to the aggregation YAML config file (default: {default_config})" + ) + parser.add_argument( + "--schema", + type=str, + default=default_schema, + help=f"Path to the JSON Schema validation file (default: {default_schema})" + ) + + args = parser.parse_args() + + print(f"Validating '{os.path.basename(args.config)}' against '{os.path.basename(args.schema)}'...") + + try: + aggregations = validate_config(args.config, args.schema) + print(f"{GREEN}[SUCCESS] Configuration is valid!{RESET}") + print(f"Parsed {len(aggregations)} aggregation steps successfully.") + sys.exit(0) + except FileNotFoundError as e: + print(f"{RED}[ERROR] File not found: {e}{RESET}", file=sys.stderr) + sys.exit(1) + except jsonschema.exceptions.ValidationError as e: + print(f"{RED}[ERROR] Schema Validation Failed:{RESET}", file=sys.stderr) + print(f"{RED} - Path: {'.'.join(str(p) for p in e.path)}{RESET}", file=sys.stderr) + print(f"{RED} - Message: {e.message}{RESET}", file=sys.stderr) + sys.exit(1) + except yaml.YAMLError as e: + print(f"{RED}[ERROR] YAML Syntax Error: {e}{RESET}", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"{RED}[ERROR] Unexpected validation failure: {e}{RESET}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/pipeline/workflow/ingestion-helper/aggregation/validator_test.py b/pipeline/workflow/ingestion-helper/aggregation/validator_test.py new file mode 100644 index 000000000..617e43a97 --- /dev/null +++ b/pipeline/workflow/ingestion-helper/aggregation/validator_test.py @@ -0,0 +1,324 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the aggregation configuration validator using real temporary files.""" + +import os +import sys +import tempfile +import textwrap +import unittest +import jsonschema +import yaml + +sys.path.append(os.path.dirname(os.path.dirname(__file__))) + +from aggregation import validate_config + + +class TestValidatorSuccess(unittest.TestCase): + """Verifies successful validation paths for valid configurations.""" + + def setUp(self): + self.schema_path = os.path.join(os.path.dirname(__file__), "schema.json") + self.tmpdir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + + def tearDown(self): + self.tmpdir.cleanup() + + def test_validate_config_success_all_types(self): + """Verifies that a comprehensive, valid config with all types passes validation.""" + valid_all_types_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: ["*"] + stage: 1 + disabled: false + + - type: place + source_type: County + destination_type: State + allow_multiple_to_places: true + imports: ["ImportA", "ImportB"] + stage: 2 + + - type: stat_var + ancestor_sv_id: Count_Person + source_sv_ids: ["Count_Person_Male", "Count_Person_Female"] + skip_all_sources_present_check: true + output_import_name: "Aggregated_Pop" + imports: ["ImportC"] + stage: 3 + + - type: entity + entity_types: ["MortalityEvent"] + location_props: ["location"] + date_prop: "date" + agg_date_formats: ["%Y"] + imports: ["ImportD"] + + - type: provenance_summary + imports: ["*"] + + - type: stat_var_groups + imports: ["*"] + """) + + with open(self.config_path, "w") as f: + f.write(valid_all_types_yaml) + + aggregations = validate_config(self.config_path, self.schema_path) + + self.assertEqual(len(aggregations), 6) + self.assertEqual(aggregations[0]["type"], "linked_edges") + self.assertEqual(aggregations[1]["source_type"], "County") + self.assertEqual(aggregations[2]["ancestor_sv_id"], "Count_Person") + self.assertEqual(aggregations[3]["entity_types"], ["MortalityEvent"]) + + +class TestValidatorSchemaConstraints(unittest.TestCase): + """Verifies core schema constraint failures (types, required fields, values).""" + + def setUp(self): + self.schema_path = os.path.join(os.path.dirname(__file__), "schema.json") + self.tmpdir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + + def tearDown(self): + self.tmpdir.cleanup() + + def test_validate_config_missing_type(self): + """Verifies that missing the required 'type' field raises ValidationError.""" + invalid_missing_type_yaml = textwrap.dedent("""\ + aggregations: + - imports: ["*"] + """) + with open(self.config_path, "w") as f: + f.write(invalid_missing_type_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("'type' is a required property", ctx.exception.message) + + def test_validate_config_missing_imports(self): + """Verifies that missing the required 'imports' field raises ValidationError.""" + invalid_missing_imports_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + """) + with open(self.config_path, "w") as f: + f.write(invalid_missing_imports_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("'imports' is a required property", ctx.exception.message) + + def test_validate_config_invalid_imports_type(self): + """Verifies that imports field being a string instead of an array raises ValidationError.""" + invalid_imports_type_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: "*" + """) + with open(self.config_path, "w") as f: + f.write(invalid_imports_type_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is not of type 'array'", ctx.exception.message) + + def test_validate_config_invalid_stage_type(self): + """Verifies that stage field being a string instead of an integer raises ValidationError.""" + invalid_stage_type_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: ["*"] + stage: "first" + """) + with open(self.config_path, "w") as f: + f.write(invalid_stage_type_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is not of type 'integer'", ctx.exception.message) + + def test_validate_config_invalid_stage_value(self): + """Verifies that a stage value of 0 (minimum is 1) raises ValidationError.""" + invalid_stage_value_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: ["*"] + stage: 0 + """) + with open(self.config_path, "w") as f: + f.write(invalid_stage_value_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is less than the minimum of 1", ctx.exception.message) + + def test_validate_config_empty_imports_list(self): + """Verifies that an empty imports list raises ValidationError.""" + invalid_empty_imports_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: [] + """) + with open(self.config_path, "w") as f: + f.write(invalid_empty_imports_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("should be non-empty", ctx.exception.message) + + def test_validate_config_missing_aggregations_key(self): + """Verifies that missing the required 'aggregations' root key raises ValidationError.""" + missing_aggregations_yaml = textwrap.dedent("""\ + some_other_key: [] + """) + with open(self.config_path, "w") as f: + f.write(missing_aggregations_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("'aggregations' is a required property", ctx.exception.message) + + def test_validate_config_empty_file(self): + """Verifies that a completely empty configuration file raises ValidationError.""" + empty_yaml = "" + with open(self.config_path, "w") as f: + f.write(empty_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("'aggregations' is a required property", ctx.exception.message) + + +class TestValidatorConditionalDependencies(unittest.TestCase): + """Verifies type-specific conditional dependencies (OneOf / dependencies).""" + + def setUp(self): + self.schema_path = os.path.join(os.path.dirname(__file__), "schema.json") + self.tmpdir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + + def tearDown(self): + self.tmpdir.cleanup() + + def test_validate_config_place_missing_field(self): + """Verifies that a place step missing the required 'source_type' raises ValidationError.""" + invalid_place_missing_field_yaml = textwrap.dedent("""\ + aggregations: + - type: place + destination_type: State + imports: ["*"] + """) + with open(self.config_path, "w") as f: + f.write(invalid_place_missing_field_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is not valid under any of the given schemas", ctx.exception.message) + + def test_validate_config_stat_var_missing_field(self): + """Verifies that a stat_var step missing the required 'source_sv_ids' raises ValidationError.""" + invalid_stat_var_missing_field_yaml = textwrap.dedent("""\ + aggregations: + - type: stat_var + ancestor_sv_id: Count_Person + imports: ["*"] + """) + with open(self.config_path, "w") as f: + f.write(invalid_stat_var_missing_field_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is not valid under any of the given schemas", ctx.exception.message) + + def test_validate_config_stat_var_empty_source_svs(self): + """Verifies that a stat_var step with an empty source_sv_ids array raises ValidationError.""" + invalid_stat_var_empty_svs_yaml = textwrap.dedent("""\ + aggregations: + - type: stat_var + ancestor_sv_id: Count_Person + source_sv_ids: [] + imports: ["*"] + """) + with open(self.config_path, "w") as f: + f.write(invalid_stat_var_empty_svs_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("should be non-empty", ctx.exception.message) + + def test_validate_config_entity_missing_field(self): + """Verifies that an entity step missing the required 'location_props' raises ValidationError.""" + invalid_entity_missing_field_yaml = textwrap.dedent("""\ + aggregations: + - type: entity + entity_types: ["Event"] + imports: ["*"] + """) + with open(self.config_path, "w") as f: + f.write(invalid_entity_missing_field_yaml) + + with self.assertRaises(jsonschema.exceptions.ValidationError) as ctx: + validate_config(self.config_path, self.schema_path) + self.assertIn("is not valid under any of the given schemas", ctx.exception.message) + + +class TestValidatorErrorsAndFileSystem(unittest.TestCase): + """Verifies file-system issues and non-schema parsing errors (YAML syntax).""" + + def setUp(self): + self.schema_path = os.path.join(os.path.dirname(__file__), "schema.json") + self.tmpdir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.tmpdir.name, "aggregation.yaml") + + def tearDown(self): + self.tmpdir.cleanup() + + def test_validate_config_yaml_syntax_error(self): + """Verifies that malformed YAML syntax raises YAMLError.""" + malformed_yaml = textwrap.dedent("""\ + aggregations: + - type: linked_edges + imports: + - "*" + """) + with open(self.config_path, "w") as f: + f.write(malformed_yaml) + + with self.assertRaises(yaml.YAMLError): + validate_config(self.config_path, self.schema_path) + + def test_validate_config_missing_config_file(self): + """Verifies that a missing config file path raises FileNotFoundError.""" + with self.assertRaises(FileNotFoundError) as ctx: + validate_config("non_existent_config.yaml", self.schema_path) + self.assertIn("Aggregation config file not found", str(ctx.exception)) + + def test_validate_config_missing_schema_file(self): + """Verifies that a missing schema file path raises FileNotFoundError.""" + with open(self.config_path, "w") as f: + f.write("aggregations: []") + + with self.assertRaises(FileNotFoundError) as ctx: + validate_config(self.config_path, "non_existent_schema.json") + self.assertIn("JSON Schema file not found", str(ctx.exception)) + + +if __name__ == '__main__': + unittest.main() diff --git a/pipeline/workflow/ingestion-helper/app_test.py b/pipeline/workflow/ingestion-helper/app_test.py index 8150e5a6b..acf0b7796 100644 --- a/pipeline/workflow/ingestion-helper/app_test.py +++ b/pipeline/workflow/ingestion-helper/app_test.py @@ -15,7 +15,6 @@ import unittest from unittest.mock import MagicMock, patch from datetime import datetime -import os from fastapi.testclient import TestClient from app import app @@ -263,6 +262,106 @@ def test_update_import_version_override_success(self, mock_get_caller_identity): # Verify get_caller_identity was called exactly once outside of the loop mock_get_caller_identity.assert_called_once() + @patch('routes.aggregation._get_orchestrator') + def test_aggregation_initiate_success(self, mock_aggregation_utils): + # Setup mock orchestrator + mock_instance = MagicMock() + mock_aggregation_utils.return_value = mock_instance + mock_instance.get_active_stages.return_value = [1] + mock_instance.execute_stage.return_value = ["job-1", "job-2"] + + # Call endpoint + payload = { + "importList": [{"importName": "USFed_Census"}] + } + response = client.post("/aggregation/initiate", json=payload) + + # Assertions + self.assertEqual(response.status_code, 200) + state = response.json() + self.assertEqual(state["status"], "RUNNING") + self.assertEqual(state["current_stage"], 1) + self.assertEqual(state["active_job_ids"], ["job-1", "job-2"]) + self.assertEqual(state["import_list"], [{"importName": "USFed_Census"}]) + + @patch('routes.aggregation._get_orchestrator') + def test_aggregation_poll_transition(self, mock_aggregation_utils): + # Setup mock orchestrator to simulate Stage 1 completion and Stage 2 execution + mock_instance = MagicMock() + mock_aggregation_utils.return_value = mock_instance + mock_instance.get_active_stages.return_value = [1, 2] + + # Mock BQ reporting Stage 1 jobs are DONE + mock_instance.check_jobs_status.return_value = {"status": "DONE"} + mock_instance.execute_stage.return_value = ["job-stage2-1"] + + # Input state (Stage 1 completed) + payload = { + "status": "RUNNING", + "current_stage": 1, + "active_job_ids": ["job-1", "job-2"], + "import_list": [{"importName": "USFed_Census"}] + } + + # Call endpoint + response = client.post("/aggregation/poll", json=payload) + + # Assertions + self.assertEqual(response.status_code, 200) + state = response.json() + self.assertEqual(state["status"], "RUNNING") + self.assertEqual(state["current_stage"], 2) # Transitioned to 2! + self.assertEqual(state["active_job_ids"], ["job-stage2-1"]) + + @patch('routes.aggregation._get_orchestrator') + def test_aggregation_poll_still_running(self, mock_aggregation_utils): + # Setup mock orchestrator to simulate jobs still in PENDING state + mock_instance = MagicMock() + mock_aggregation_utils.return_value = mock_instance + mock_instance.get_active_stages.return_value = [1] + + # Mock BQ reporting Stage 1 jobs are PENDING (still executing) + mock_instance.check_jobs_status.return_value = {"status": "PENDING"} + + # Input state + payload = { + "status": "RUNNING", + "current_stage": 1, + "active_job_ids": ["job-1", "job-2"], + "import_list": [{"importName": "USFed_Census"}] + } + + # Call endpoint + response = client.post("/aggregation/poll", json=payload) + + # Assertions + self.assertEqual(response.status_code, 200) + state = response.json() + # Verify state is returned unchanged + self.assertEqual(state["status"], "RUNNING") + self.assertEqual(state["current_stage"], 1) + self.assertEqual(state["active_job_ids"], ["job-1", "job-2"]) + + @patch('routes.aggregation._get_orchestrator') + def test_aggregation_run(self, mock_aggregation_utils): + # Setup mock orchestrator + mock_instance = MagicMock() + mock_aggregation_utils.return_value = mock_instance + mock_instance.get_active_stages.return_value = [1, 2] + mock_instance.execute_stage.side_effect = lambda stage, imports: [f"job-stage{stage}-1"] + + # Call legacy endpoint + payload = { + "importList": [{"importName": "USFed_Census"}] + } + response = client.post("/aggregation/run", json=payload) + + # Assertions (should return all jobs from all stages in parallel) + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data["status"], "SUBMITTED") + self.assertEqual(data["jobIds"], ["job-stage1-1", "job-stage2-1"]) + if __name__ == '__main__': unittest.main() diff --git a/pipeline/workflow/ingestion-helper/pyproject.toml b/pipeline/workflow/ingestion-helper/pyproject.toml index 6301d7b2e..4ce4a251a 100644 --- a/pipeline/workflow/ingestion-helper/pyproject.toml +++ b/pipeline/workflow/ingestion-helper/pyproject.toml @@ -33,6 +33,8 @@ dependencies = [ "google-cloud-bigquery", "redis", "jinja2", + "pyyaml>=6.0.3", + "jsonschema>=4.26.0", ] [tool.hatch.version] diff --git a/pipeline/workflow/ingestion-helper/routes/aggregation.py b/pipeline/workflow/ingestion-helper/routes/aggregation.py index 7433f906d..f15e8a4e1 100644 --- a/pipeline/workflow/ingestion-helper/routes/aggregation.py +++ b/pipeline/workflow/ingestion-helper/routes/aggregation.py @@ -13,44 +13,70 @@ # limitations under the License. import logging -from fastapi import APIRouter, HTTPException -from utils.aggregation import AggregationUtils -import config from typing import Any, Dict, List, Optional +from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field + +import config from routes.models import BaseResponse, ResponseStatus +from aggregation import AggregationOrchestrator from utils.logging import log_start + + +class AggregationWorkflowState(BaseModel): + """Represents the execution state of a multi-stage aggregation pipeline run. + + This state object is passed back and forth between the client (Google Cloud + Workflows) and the helper service endpoints to durably maintain the progress + of a stateless, sequential aggregation run across multiple stages. + """ + status: str = Field(..., description="Overall status of the run: RUNNING, SUCCEEDED, FAILED") + current_stage: int = Field(..., description="The stage currently executing") + active_job_ids: List[str] = Field(default_factory=list, description="BQ job IDs running in the current stage") + import_list: List[Dict[str, Any]] = Field(default_factory=list, description="Original list of imports") + error: Optional[str] = Field(default=None, description="Detailed error message if failed") + +class InitiateRequest(BaseModel): + importList: List[Dict[str, Any]] = Field(default_factory=list) + +# TODO: Remove AggregationRequest once all consumers migrate to /initiate and /poll class AggregationRequest(BaseModel): + """Temporary request model for compatibility run endpoint.""" importList: List[Dict[str, Any]] = Field(default_factory=list) + +# TODO: Remove AggregationStatusRequest once all consumers migrate to /initiate and /poll class AggregationStatusRequest(BaseModel): + """Temporary request model for compatibility status endpoint.""" jobIds: List[str] = Field(default_factory=list) + +# TODO: Remove AggregationResponse once all consumers migrate to /initiate and /poll class AggregationResponse(BaseResponse): + """Temporary response model for compatibility run endpoint.""" jobIds: List[str] = Field(default_factory=list, description="BigQuery job IDs submitted for async aggregation") + +# TODO: Remove AggregationStatusResponse once all consumers migrate to /initiate and /poll class AggregationStatusResponse(BaseResponse): + """Temporary response model for compatibility status endpoint.""" error: Optional[str] = Field(default=None, description="Detailed error message if failed") failedJobs: Optional[List[str]] = Field(default_factory=list, description="List of failed BigQuery job IDs") + + router = APIRouter(prefix="/aggregation", tags=["aggregation"]) -@router.post("/run", response_model=AggregationResponse) -@log_start -def run_aggregation(req: AggregationRequest): - """Runs aggregation logic asynchronously for the specified imports, returning BigQuery job IDs.""" - if not req.importList: - logging.info("Empty import list. Skipping aggregation.") - return AggregationResponse(status=ResponseStatus.SUBMITTED, jobIds=[]) - + +def _get_orchestrator() -> AggregationOrchestrator: + """Helper to initialize the orchestrator using global config.""" if not all([config.SPANNER_CONNECTION_ID, config.SPANNER_PROJECT_ID, config.SPANNER_INSTANCE_ID, config.SPANNER_GRAPH_DATABASE_ID]): raise HTTPException( status_code=400, detail="Missing required configuration environment variables: SPANNER_CONNECTION_ID, SPANNER_PROJECT_ID, SPANNER_INSTANCE_ID, or SPANNER_GRAPH_DATABASE_ID" ) - - aggregation = AggregationUtils( + return AggregationOrchestrator( connection_id=config.SPANNER_CONNECTION_ID, project_id=config.SPANNER_PROJECT_ID, instance_id=config.SPANNER_INSTANCE_ID, @@ -58,39 +84,158 @@ def run_aggregation(req: AggregationRequest): location=config.LOCATION, is_base_dc=config.IS_BASE_DC, ) + + + +@router.post("/initiate", response_model=AggregationWorkflowState) +@log_start +def initiate_aggregation(req: InitiateRequest): + """Initiates the aggregation run by executing Stage 1 and returning the initial state.""" + if not req.importList: + logging.info("Empty import list. Skipping aggregation.") + return AggregationWorkflowState(status="SUCCEEDED", current_stage=0, active_job_ids=[], import_list=[]) + try: - job_ids = aggregation.run_aggregation(req.importList) + orchestrator = _get_orchestrator() + import_names = [item.get('importName') for item in req.importList if item.get('importName')] + + active_stages = orchestrator.get_active_stages(import_names) + if not active_stages: + logging.info("No stages have active aggregations for the current imports. Completing immediately.") + return AggregationWorkflowState(status="SUCCEEDED", current_stage=0, active_job_ids=[], import_list=req.importList) + + first_stage = active_stages[0] + + logging.info(f"Initiating aggregation at Stage {first_stage}") + job_ids = orchestrator.execute_stage(first_stage, import_names) + + return AggregationWorkflowState( + status="RUNNING", + current_stage=first_stage, + active_job_ids=job_ids, + import_list=req.importList + ) + except Exception as e: + logging.error(f"Failed to initiate aggregation: {e}") + raise HTTPException(status_code=500, detail=f"Failed to initiate aggregation: {str(e)}") + + +@router.post("/poll", response_model=AggregationWorkflowState) +@log_start +def poll_aggregation(state: AggregationWorkflowState): + """Checks progress of active jobs and transitions to the next stage if complete.""" + if state.status != "RUNNING": + return state # Already in a terminal state + + try: + orchestrator = _get_orchestrator() + import_names = [item.get('importName') for item in state.import_list if item.get('importName')] + + # 1. Check status of active jobs in BigQuery + if not state.active_job_ids: + bq_status = {"status": "DONE"} + else: + logging.info(f"Polling status for jobs in Stage {state.current_stage}: {state.active_job_ids}") + bq_status = orchestrator.check_jobs_status(state.active_job_ids) + + # Case A: Any job failed + if bq_status["status"] == "FAILED": + logging.error(f"Stage {state.current_stage} failed with error: {bq_status.get('error')}") + return AggregationWorkflowState( + status="FAILED", + current_stage=state.current_stage, + active_job_ids=[], + import_list=state.import_list, + error=bq_status.get("error") + ) + + # Case B: Jobs are still executing (explicitly check for DONE to transition) + if bq_status["status"] != "DONE": + logging.info(f"Stage {state.current_stage} is still executing (status: {bq_status['status']}).") + return state # Return unchanged + + # Case C: All jobs succeeded -> Find and execute the next active stage + active_stages = orchestrator.get_active_stages(import_names) + next_stages = [s for s in active_stages if s > state.current_stage] + + if next_stages: + next_stage = next_stages[0] + logging.info(f"Stage {state.current_stage} completed. Transitioning to Stage {next_stage}...") + new_job_ids = orchestrator.execute_stage(next_stage, import_names) + return AggregationWorkflowState( + status="RUNNING", + current_stage=next_stage, + active_job_ids=new_job_ids, + import_list=state.import_list + ) + + # If we exit the loop, there are no more active stages left + logging.info("All aggregation stages completed successfully!") + return AggregationWorkflowState( + status="SUCCEEDED", + current_stage=state.current_stage, + active_job_ids=[], + import_list=state.import_list + ) + + except Exception as e: + logging.error(f"Error during polling: {e}") + return AggregationWorkflowState( + status="FAILED", + current_stage=state.current_stage, + active_job_ids=[], + import_list=state.import_list, + error=f"Orchestrator error: {str(e)}" + ) + +# TODO: Remove the /run endpoint once all consumers migrate to /initiate and /poll +@router.post("/run", response_model=AggregationResponse, deprecated=True) +@log_start +def run_aggregation(req: AggregationRequest): + """Temporary endpoint. Runs ALL enabled aggregations in parallel (ignores stages). + + Please migrate to /initiate and /poll endpoints. + """ + if not req.importList: + logging.info("Empty import list. Skipping temporary aggregation.") + return AggregationResponse(status=ResponseStatus.SUBMITTED, jobIds=[]) + + try: + orchestrator = _get_orchestrator() + import_names = [item.get('importName') for item in req.importList if item.get('importName')] + + # Compatibility Mode: Submit ALL enabled stages in parallel + job_ids = [] + active_stages = orchestrator.get_active_stages(import_names) + for stage_num in active_stages: + job_ids.extend(orchestrator.execute_stage(stage_num, import_names)) + return AggregationResponse(status=ResponseStatus.SUBMITTED, jobIds=job_ids) except Exception as e: - raise HTTPException(status_code=500, detail=f"Aggregation failed: {str(e)}") + logging.error(f"Temporary aggregation failed: {e}") + raise HTTPException(status_code=500, detail=f"Temporary aggregation failed: {str(e)}") -@router.post("/status", response_model=AggregationStatusResponse) -def check_aggregation_status(req: AggregationStatusRequest): - """Checks the status of the submitted aggregation BigQuery jobs.""" + +# TODO: Remove the /status endpoint once all consumers migrate to /initiate and /poll +@router.post("/status", response_model=AggregationStatusResponse, deprecated=True) +@log_start +def get_aggregation_status(req: AggregationStatusRequest): + """Temporary endpoint. Checks the status of the submitted BigQuery jobs. + + Please migrate to /initiate and /poll endpoints. + """ if not req.jobIds: logging.info("Empty jobIds. Returning status DONE.") return AggregationStatusResponse(status=ResponseStatus.DONE) - if not all([config.SPANNER_CONNECTION_ID, config.SPANNER_PROJECT_ID, config.SPANNER_INSTANCE_ID, config.SPANNER_GRAPH_DATABASE_ID]): - raise HTTPException( - status_code=400, - detail="Missing required configuration environment variables." - ) - - aggregation = AggregationUtils( - connection_id=config.SPANNER_CONNECTION_ID, - project_id=config.SPANNER_PROJECT_ID, - instance_id=config.SPANNER_INSTANCE_ID, - database_id=config.SPANNER_GRAPH_DATABASE_ID, - location=config.LOCATION, - is_base_dc=config.IS_BASE_DC, - ) try: - status_info = aggregation.check_aggregation_status(req.jobIds) + orchestrator = _get_orchestrator() + status_info = orchestrator.check_jobs_status(req.jobIds) return AggregationStatusResponse( status=ResponseStatus.from_str(status_info.get("status", "ERROR")), error=status_info.get("error"), - failedJobs=status_info.get("failedJobs", []) + failedJobs=status_info.get("failed_jobs", []) ) except Exception as e: - raise HTTPException(status_code=500, detail=f"Aggregation status check failed: {str(e)}") + logging.error(f"Temporary check status failed: {e}") + raise HTTPException(status_code=500, detail=f"Temporary check status failed: {str(e)}") diff --git a/pipeline/workflow/ingestion-helper/utils/aggregation.py b/pipeline/workflow/ingestion-helper/utils/aggregation.py deleted file mode 100644 index 432a4000f..000000000 --- a/pipeline/workflow/ingestion-helper/utils/aggregation.py +++ /dev/null @@ -1,98 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import logging -from typing import Any, Dict, List, Optional - -from aggregation import BigQueryExecutor -from aggregation import LinkedEdgeGenerator -from aggregation import ProvenanceSummaryGenerator -from aggregation import StatVarAggregator -from aggregation import StatVarGroupGenerator -from google.cloud import bigquery - -logging.getLogger().setLevel(logging.INFO) - - -class AggregationUtils: - """Orchestrates the overall aggregation workflow.""" - - def __init__(self, - connection_id: str, - project_id: str, - instance_id: str, - database_id: str, - location: Optional[str] = None, - is_base_dc: bool = True) -> None: - # TODO: remove sequential execution once DCP changes are made - # Use sequential execution for DCP (backward compatibility) - run_sequential = not is_base_dc - self.executor = BigQueryExecutor(connection_id=connection_id, - project_id=project_id, - instance_id=instance_id, - database_id=database_id, - location=location, - run_sequential=run_sequential) - self.linked_edge_generator = LinkedEdgeGenerator( - self.executor, is_base_dc) - self.provenance_summary_generator = ProvenanceSummaryGenerator( - self.executor, is_base_dc) - - def run_aggregation(self, import_list: List[Dict[str, Any]]) -> List[str]: - """ - Orchestrates standard per-import aggregations and global aggregations. - Returns a list of BigQuery job IDs for async polling. - """ - logging.info(f"Received request for importList: {import_list}") - - try: - import_names = [] - # 1. Run standard per-import aggregations - for import_item in import_list: - import_name = import_item.get('importName') - if import_name: - import_names.append(import_name) - query = "SELECT @import_name as import_name, CURRENT_TIMESTAMP() as execution_time" - job_config = bigquery.QueryJobConfig(query_parameters=[ - bigquery.ScalarQueryParameter("import_name", "STRING", - import_name), - ]) - self.executor.execute(query, job_config=job_config) - else: - logging.info( - 'Skipping aggregation logic for empty importName') - - # 2. Run global aggregations asynchronously - jobs = [] - jobs.extend(self.linked_edge_generator.run_all(import_names)) - jobs.extend(self.provenance_summary_generator.run_all(import_names)) - - job_ids = [job.job_id for job in jobs if job] - logging.info(f"Submitted async aggregation jobs: {job_ids}") - - return job_ids - except Exception as e: - logging.error(f"Aggregation failed: {e}") - raise e - - def check_aggregation_status(self, job_ids: List[str]) -> Dict[str, Any]: - """ - Checks the status of the provided BigQuery job IDs. - """ - logging.info(f"Checking status for jobs: {job_ids}") - try: - return self.executor.get_jobs_status(job_ids) - except Exception as e: - logging.error(f"Failed to check aggregation status: {e}") - raise e diff --git a/pipeline/workflow/ingestion-helper/utils/aggregation_test.py b/pipeline/workflow/ingestion-helper/utils/aggregation_test.py deleted file mode 100644 index f4afb6e3f..000000000 --- a/pipeline/workflow/ingestion-helper/utils/aggregation_test.py +++ /dev/null @@ -1,87 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import os -import sys -import unittest -from unittest.mock import MagicMock -from unittest.mock import patch - -sys.path.append(os.path.dirname(os.path.dirname(__file__))) - -from utils.aggregation import AggregationUtils - - -@patch('utils.aggregation.BigQueryExecutor') -@patch('utils.aggregation.LinkedEdgeGenerator') -@patch('utils.aggregation.ProvenanceSummaryGenerator') -@patch('utils.aggregation.StatVarGroupGenerator') -class TestAggregationUtils(unittest.TestCase): - - def test_run_aggregation(self, mock_prov_gen, mock_edge_gen, mock_executor): - # Setup mocks - mock_executor_instance = MagicMock() - mock_executor.return_value = mock_executor_instance - - mock_edge_gen_instance = MagicMock() - mock_edge_gen.return_value = mock_edge_gen_instance - mock_job1 = MagicMock() - mock_job1.job_id = "job1" - mock_edge_gen_instance.run_all.return_value = [mock_job1] - - mock_prov_gen_instance = MagicMock() - mock_prov_gen.return_value = mock_prov_gen_instance - mock_job2 = MagicMock() - mock_job2.job_id = "job2" - mock_prov_gen_instance.run_all.return_value = [mock_job2] - - utils = AggregationUtils(connection_id="conn", - project_id="proj", - instance_id="inst", - database_id="db", - is_base_dc=True) - - import_list = [{'importName': 'import1'}, {'importName': 'import2'}] - job_ids = utils.run_aggregation(import_list) - - # Verify standard import queries were executed - self.assertEqual(mock_executor_instance.execute.call_count, 2) - - # Verify generators were called - mock_edge_gen_instance.run_all.assert_called_once_with( - ["import1", "import2"]) - mock_prov_gen_instance.run_all.assert_called_once_with( - ["import1", "import2"]) - - self.assertEqual(job_ids, ["job1", "job2"]) - - def test_check_aggregation_status(self, mock_prov_gen, mock_edge_gen, - mock_executor): - mock_executor_instance = MagicMock() - mock_executor.return_value = mock_executor_instance - mock_executor_instance.get_jobs_status.return_value = {"status": "DONE"} - - utils = AggregationUtils(connection_id="conn", - project_id="proj", - instance_id="inst", - database_id="db") - - status = utils.check_aggregation_status(["job1", "job2"]) - mock_executor_instance.get_jobs_status.assert_called_once_with( - ["job1", "job2"]) - self.assertEqual(status, {"status": "DONE"}) - - -if __name__ == '__main__': - unittest.main() diff --git a/pipeline/workflow/ingestion-helper/uv.lock b/pipeline/workflow/ingestion-helper/uv.lock index 6fa14bcb8..79e17f455 100644 --- a/pipeline/workflow/ingestion-helper/uv.lock +++ b/pipeline/workflow/ingestion-helper/uv.lock @@ -47,6 +47,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, ] +[[package]] +name = "attrs" +version = "26.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9a/8e/82a0fe20a541c03148528be8cac2408564a6c9a0cc7e9171802bc1d26985/attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32", size = 952055, upload-time = "2026-03-19T14:22:25.026Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" }, +] + [[package]] name = "certifi" version = "2026.5.20" @@ -272,7 +281,9 @@ dependencies = [ { name = "google-cloud-spanner" }, { name = "google-cloud-storage" }, { name = "jinja2" }, + { name = "jsonschema" }, { name = "pydantic" }, + { name = "pyyaml" }, { name = "redis" }, { name = "uvicorn", extra = ["standard"] }, ] @@ -293,7 +304,9 @@ requires-dist = [ { name = "google-cloud-spanner" }, { name = "google-cloud-storage" }, { name = "jinja2" }, + { name = "jsonschema", specifier = ">=4.26.0" }, { name = "pydantic", specifier = ">=2.0.0" }, + { name = "pyyaml", specifier = ">=6.0.3" }, { name = "redis" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.28.0" }, ] @@ -722,6 +735,33 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" }, ] +[[package]] +name = "jsonschema" +version = "4.26.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "attrs" }, + { name = "jsonschema-specifications" }, + { name = "referencing" }, + { name = "rpds-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b3/fc/e067678238fa451312d4c62bf6e6cf5ec56375422aee02f9cb5f909b3047/jsonschema-4.26.0.tar.gz", hash = "sha256:0c26707e2efad8aa1bfc5b7ce170f3fccc2e4918ff85989ba9ffa9facb2be326", size = 366583, upload-time = "2026-01-07T13:41:07.246Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/69/90/f63fb5873511e014207a475e2bb4e8b2e570d655b00ac19a9a0ca0a385ee/jsonschema-4.26.0-py3-none-any.whl", hash = "sha256:d489f15263b8d200f8387e64b4c3a75f06629559fb73deb8fdfb525f2dab50ce", size = 90630, upload-time = "2026-01-07T13:41:05.306Z" }, +] + +[[package]] +name = "jsonschema-specifications" +version = "2025.9.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "referencing" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/19/74/a633ee74eb36c44aa6d1095e7cc5569bebf04342ee146178e2d36600708b/jsonschema_specifications-2025.9.1.tar.gz", hash = "sha256:b540987f239e745613c7a9176f3edb72b832a4ac465cf02712288397832b5e8d", size = 32855, upload-time = "2025-09-08T01:34:59.186Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" }, +] + [[package]] name = "markupsafe" version = "3.0.3" @@ -1196,6 +1236,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/27/e3/b519734372d305bd547534a9f32e4ce9f98552af753dce72cf3483a0ff0b/redis-8.0.0-py3-none-any.whl", hash = "sha256:c938c18338585009f0bc310f4c7e4e4b4d37639356c4ac072cedf3af570c8dc7", size = 499870, upload-time = "2026-05-28T12:45:11.697Z" }, ] +[[package]] +name = "referencing" +version = "0.37.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "attrs" }, + { name = "rpds-py" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/22/f5/df4e9027acead3ecc63e50fe1e36aca1523e1719559c499951bb4b53188f/referencing-0.37.0.tar.gz", hash = "sha256:44aefc3142c5b842538163acb373e24cce6632bd54bdb01b21ad5863489f50d8", size = 78036, upload-time = "2025-10-13T15:30:48.871Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2c/58/ca301544e1fa93ed4f80d724bf5b194f6e4b945841c5bfd555878eea9fcb/referencing-0.37.0-py3-none-any.whl", hash = "sha256:381329a9f99628c9069361716891d34ad94af76e461dcb0335825aecc7692231", size = 26766, upload-time = "2025-10-13T15:30:47.625Z" }, +] + [[package]] name = "requests" version = "2.34.2" @@ -1211,6 +1265,116 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a0/f4/c67b0b3f1b9245e8d266f0f112c500d50e5b4e83cb6f3b71b6528104182a/requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0", size = 73075, upload-time = "2026-05-14T19:25:26.443Z" }, ] +[[package]] +name = "rpds-py" +version = "2026.5.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/2e/43/25a8dcd3feedd735039a8f0b5b7e3b118232b5eae288c4fd9ab200d41094/rpds_py-2026.5.1.tar.gz", hash = "sha256:07b24fea40541e28570e5b795a4a38fbdcd12550c06bd0748005ecc8116ca256", size = 64459, upload-time = "2026-05-28T12:02:13.232Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d4/e7/a78582dc57caa592dcc7d4fb69b61390561e908eb3d2f5df5928a8e354c0/rpds_py-2026.5.1-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:3abe24a66e57adcfa645d718063a5fa5103ecc71ddbf26d78af8f9368018ff1d", size = 353040, upload-time = "2026-05-28T11:59:12.531Z" }, + { url = "https://files.pythonhosted.org/packages/a3/43/35e3f136343aef451e545ce8c38d36c2f93c0ed88703db8b64ba2b205c68/rpds_py-2026.5.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:58b1d94308ddf0b1982f61f2eb54bf92997c9ece8a8093ef014250f4a517906c", size = 345775, upload-time = "2026-05-28T11:59:13.827Z" }, + { url = "https://files.pythonhosted.org/packages/20/e1/0f2160c5982d3157734d5cb3ed63d8b2d583a73c9864f77b666449f32cf8/rpds_py-2026.5.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0fa92420128dadce7f54bd73ba1825a273e9268fe9e35dbf7e6362890efa4e08", size = 376329, upload-time = "2026-05-28T11:59:15.271Z" }, + { url = "https://files.pythonhosted.org/packages/d0/11/ee0ba42aff83bf4effdbc576673c6be64c5e173978c3f6d537e94482f77d/rpds_py-2026.5.1-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:ca653c6546386227cd9800d1bef6a348099acf8db4250341da6d90f663d6dfcb", size = 383539, upload-time = "2026-05-28T11:59:16.665Z" }, + { url = "https://files.pythonhosted.org/packages/11/df/d94aa6a499d4ac40afe2d7620f2c597fd3c0f182e854ad7cf3f596a81cb6/rpds_py-2026.5.1-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:66c93681c4729e4e3ecba31b8179fae083ff3118841672835140338b4b9867c1", size = 494674, upload-time = "2026-05-28T11:59:17.991Z" }, + { url = "https://files.pythonhosted.org/packages/1f/75/33d30f43bb2f458de11979486a591b1bf6e5651765ed1704c6197c2dc773/rpds_py-2026.5.1-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:40ff257542e04796880e011e15cd4dc21c2599975df2aaa8f2c8495ca574e1a5", size = 389268, upload-time = "2026-05-28T11:59:19.434Z" }, + { url = "https://files.pythonhosted.org/packages/f4/1e/2c9096fc19d5fd084b0184ca2b651e659aa0a37e6fdbecf6ece47f147fe1/rpds_py-2026.5.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b6825cc329b290e93c5f6a9be2393118a763f6ccf6abd83704e0c102ca583644", size = 376280, upload-time = "2026-05-28T11:59:21Z" }, + { url = "https://files.pythonhosted.org/packages/b9/e5/61ec9f8be8211ea7f48448195549e4aaf02004083475493b0e137702ecb2/rpds_py-2026.5.1-cp312-cp312-manylinux_2_31_riscv64.whl", hash = "sha256:de42116e69cb53b911cc34aee5ab98f36c597b822545045d49e938818b99e5e4", size = 387233, upload-time = "2026-05-28T11:59:22.454Z" }, + { url = "https://files.pythonhosted.org/packages/0d/ca/bcec1005c4f4a234f92a29078631fee49206c7265ccae966f18fd332e80e/rpds_py-2026.5.1-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:c0f920015df2a504bebaba6d4c31ccf3fcf942f92655c086da30b671aad19aa6", size = 405009, upload-time = "2026-05-28T11:59:23.845Z" }, + { url = "https://files.pythonhosted.org/packages/72/e6/4d5718c5cf26c522dc7c9999e238da1e77380b81d0c5d1df11e271ddfeb1/rpds_py-2026.5.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:0408a24e44feb919423dc6d9da677cb5cddb894d2ca9e763967d156d9c60fab4", size = 553113, upload-time = "2026-05-28T11:59:25.184Z" }, + { url = "https://files.pythonhosted.org/packages/d4/25/2ee807bdb3e1f0b7eddf7782acd5665a8b5205a331a7d7244a52c4812fd9/rpds_py-2026.5.1-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:cea68bcd53467561ae2f96a6bdad1544299ba97b5b0ddcd5ac3d376e5c781c24", size = 618838, upload-time = "2026-05-28T11:59:26.749Z" }, + { url = "https://files.pythonhosted.org/packages/6a/c1/7d4c26f167f8c41501cc073d30ee22082b16ce358cf5b00ec97cbc7804ea/rpds_py-2026.5.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:4be8b1d2a705cc37d08256004e1d07de143fa0075c8e85a3df020b776f62b732", size = 582436, upload-time = "2026-05-28T11:59:28.11Z" }, + { url = "https://files.pythonhosted.org/packages/04/1d/9d12b0a337bab46f4769f8857f4007e3b2d639e14f9a44a0efe157696e64/rpds_py-2026.5.1-cp312-cp312-win32.whl", hash = "sha256:6736718bd4fc49cbcb538ba30516fdbef161522acefb739657d48b97bd864fed", size = 212734, upload-time = "2026-05-28T11:59:29.689Z" }, + { url = "https://files.pythonhosted.org/packages/c5/93/e4116f2de7f56bc7406a76033dc501811ddeb22b7f056b92d632871ebb0c/rpds_py-2026.5.1-cp312-cp312-win_amd64.whl", hash = "sha256:0a7d1eec967df0e9b22614a5e177622e0c89611d03727fa0cb48e45028907870", size = 229045, upload-time = "2026-05-28T11:59:31.033Z" }, + { url = "https://files.pythonhosted.org/packages/cb/53/6c3419d85eb2ec5938a37627c585b42d76a63bb731d6e42ed4b079ebf486/rpds_py-2026.5.1-cp312-cp312-win_arm64.whl", hash = "sha256:1841d067089e117142d79b98aa0df2f08b52f2ecc1819dd2700636c0db74a473", size = 223967, upload-time = "2026-05-28T11:59:32.318Z" }, + { url = "https://files.pythonhosted.org/packages/6c/32/14c961ad295f490eb0849ada8b79683e93a59b9de3afdd983eaf55fa6867/rpds_py-2026.5.1-cp313-cp313-macosx_10_12_x86_64.whl", hash = "sha256:efef4ac29c6ff495531eb17ee705b62841ecaa291b7c7077e848ea03e237164d", size = 352787, upload-time = "2026-05-28T11:59:33.655Z" }, + { url = "https://files.pythonhosted.org/packages/ca/bb/d1b85117967c11191441a7274ae616c65d93901d082c588f89a50a8da5ae/rpds_py-2026.5.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:c39f5b67a8a2e67179ada2a954227d670fe65fa9098457f698f56ddf248709b3", size = 345179, upload-time = "2026-05-28T11:59:35Z" }, + { url = "https://files.pythonhosted.org/packages/7c/46/d84105f062e626a1b233f863907288a4708c2d833b8b4c6fb2764bc080c0/rpds_py-2026.5.1-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b5c30f3f04eef4fbd362226a6f31d7c8895ca4fbb6e0b790f6890a98d8da8559", size = 376173, upload-time = "2026-05-28T11:59:36.43Z" }, + { url = "https://files.pythonhosted.org/packages/e2/ae/469d7959ce5b1201e1de135dc735b86db3b35dd0d1734f6a44246d5f061c/rpds_py-2026.5.1-cp313-cp313-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:277f6c82f0580848796c7ecc8a7173aa3bfb928e4ff831261c2f60a81dc270db", size = 383162, upload-time = "2026-05-28T11:59:37.995Z" }, + { url = "https://files.pythonhosted.org/packages/dc/a2/57853d31a1116a561aa072794602ad3f6341e18d70a8523f1bd5b9fc1e5a/rpds_py-2026.5.1-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:63c2c4c213f1a4e3f3de28ecab029dbdee976324e729c0d7a55211be72576b02", size = 495093, upload-time = "2026-05-28T11:59:39.453Z" }, + { url = "https://files.pythonhosted.org/packages/99/63/3a8eabcad9314b7daf5c65f451d2c33d989235cd8a5762186cf2c3f5a4f8/rpds_py-2026.5.1-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3350ec808fb538fe71a1f94dfaa0e29c598dfad805ce49f0caec5ae3183c652b", size = 389829, upload-time = "2026-05-28T11:59:40.896Z" }, + { url = "https://files.pythonhosted.org/packages/4b/25/05678d97fc25e2622df14dc530fb82023174ecfff6733991ed0d78f167bd/rpds_py-2026.5.1-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b1b964e3ab599e718dc46c018d104b1ebc007cbc6567d827c94a687fca56d77e", size = 374786, upload-time = "2026-05-28T11:59:42.626Z" }, + { url = "https://files.pythonhosted.org/packages/88/d1/8c90b6431e80a3b91b284a5c7c8c0c4f9c006444d90477a740d6e0f9c694/rpds_py-2026.5.1-cp313-cp313-manylinux_2_31_riscv64.whl", hash = "sha256:19cb09fab7b7fc96b2a6e28f2e34b72a3705ff27b37edb77455316e5d3f3dc9b", size = 386920, upload-time = "2026-05-28T11:59:44.124Z" }, + { url = "https://files.pythonhosted.org/packages/ff/99/4638f672ab356682d633ee0da9255f5b67ce6efd0b85eb94ad3e255e65a5/rpds_py-2026.5.1-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:abe76bcdba31e576cb83eeb8797aa0d882b738fef6dc65d0601fc753806a5b46", size = 405059, upload-time = "2026-05-28T11:59:47.177Z" }, + { url = "https://files.pythonhosted.org/packages/66/3f/3546524b6eb4cc2e1f363a3d638fa52f6c24faae3500c25fb488b02f1740/rpds_py-2026.5.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:8bff7073db3899158fff55ebf57b113a67030af26f80a18978f9f0aa60250ddf", size = 553030, upload-time = "2026-05-28T11:59:48.603Z" }, + { url = "https://files.pythonhosted.org/packages/c6/c3/7b3388c796fcf471bd17194242d4dc1a7608567c0fa422bcc1c5e79f9c1e/rpds_py-2026.5.1-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:8ba264fa49be666cd9cc56bf34ec7002fb3d27a4aee5bcb4d43d0d18feb1bb6f", size = 618975, upload-time = "2026-05-28T11:59:50.314Z" }, + { url = "https://files.pythonhosted.org/packages/61/1e/a3cb07f2795075d1d88efddae2f541359fde5f08c81ee114c29c2949c90a/rpds_py-2026.5.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:4860b603ddda0475a8885499b3729e90229d480105b42651962a5397d995fa89", size = 581178, upload-time = "2026-05-28T11:59:51.673Z" }, + { url = "https://files.pythonhosted.org/packages/a1/74/e758c03a5ef46f04c37f2651a2893db846d569ba8a7bca469d4b58939bcd/rpds_py-2026.5.1-cp313-cp313-win32.whl", hash = "sha256:7944270ae71383f6e2657dd7d5ce4eeb4ac2d0059a6738f0510583d462ab4842", size = 212481, upload-time = "2026-05-28T11:59:53.148Z" }, + { url = "https://files.pythonhosted.org/packages/70/ec/a2aca432db9c7359b40fa393eeeaa0d166c2f70175be956e75fa24197c44/rpds_py-2026.5.1-cp313-cp313-win_amd64.whl", hash = "sha256:88647f43a73c4e01be19b04ceef0c8d3a1958153604d13c773becd8016f2a0cf", size = 228519, upload-time = "2026-05-28T11:59:54.505Z" }, + { url = "https://files.pythonhosted.org/packages/29/60/a73bfdd45b096574556acf303bbd9fa9eed36ca8a818b514e2a5d5fe2b9d/rpds_py-2026.5.1-cp313-cp313-win_arm64.whl", hash = "sha256:453895624ecf7db7063b1004e44037522bbaef9ff6a945e59bc71662d7a03abd", size = 223446, upload-time = "2026-05-28T11:59:56.081Z" }, + { url = "https://files.pythonhosted.org/packages/18/e2/408105fd611823f00882aea810f3989a30d26b1bab8b6beb20f98c724e0e/rpds_py-2026.5.1-cp313-cp313t-macosx_10_12_x86_64.whl", hash = "sha256:b4e4bc98639ec915f512fde3aa7a95e0041d95d9c3cc86eea841fa63cb1e8600", size = 355287, upload-time = "2026-05-28T11:59:57.448Z" }, + { url = "https://files.pythonhosted.org/packages/8d/58/5c4a43436843c90d0f6d19f82c200c80e3843ca9fa07b237623327f6d384/rpds_py-2026.5.1-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:cacedb7a6e167680acba45ad5716e89067d225dc80da0d7040cae8c81d4572fa", size = 347033, upload-time = "2026-05-28T11:59:58.881Z" }, + { url = "https://files.pythonhosted.org/packages/fb/c2/1a71acdacaf4e259b10278fb87b039ded3cf80041bcd89dd8a3ea702ded6/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:68700371c5d7ae1412862ddfa719090925c93ecf351c566d66f09d04b136ea00", size = 376891, upload-time = "2026-05-28T12:00:00.516Z" }, + { url = "https://files.pythonhosted.org/packages/c2/c8/535f3d9b65addd8e28aa87b83c6e526799c3717a88273db8ea795beeef7a/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:296c799becfa849c779c8725494fe9ed94959ed886787df4364b058465bad7f0", size = 385646, upload-time = "2026-05-28T12:00:02.394Z" }, + { url = "https://files.pythonhosted.org/packages/1c/91/dc033f313345c354ade914dbe73cdb90b615a4409ea02430d5356794f3d8/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d3858b908218ee108d0bbfb2095ccc237648053c9bf98affad7cb079acaf1d97", size = 498830, upload-time = "2026-05-28T12:00:04.189Z" }, + { url = "https://files.pythonhosted.org/packages/27/fc/90fcbea459dbb8ddc18a2e0fd1de9412b48bc84ffff2db771cf714bacfd6/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:4fb8d2e7cb2f850b169806d61d1b991738acec96500a75c30f49caf064ce7cef", size = 392830, upload-time = "2026-05-28T12:00:05.797Z" }, + { url = "https://files.pythonhosted.org/packages/b2/1d/46cd11a228c9750684a798d98f878be6f614aa762438da7378f035e79e35/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:27b74c10ed6a8f190f4287f53bcfea348b92a84a9c9f70d30183d1e6172d580d", size = 379613, upload-time = "2026-05-28T12:00:07.433Z" }, + { url = "https://files.pythonhosted.org/packages/24/4a/d9b0c6af3a1de03eb93741bbe8be2bdce84d8fda8224f3005451d86df389/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_31_riscv64.whl", hash = "sha256:b9a6528956191c48c52294a592dbd4a8386d7048bdb25c0efcb6b966466c6d83", size = 388183, upload-time = "2026-05-28T12:00:09.227Z" }, + { url = "https://files.pythonhosted.org/packages/c5/b4/db7aaabdda6d020afc87d981bcc2f57a434c7dec60ecfc2ab3dd50b20351/rpds_py-2026.5.1-cp313-cp313t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:af03e34e860047bc7a352b842856fcf78798fbb81132cc98bd2f907ab4eb9cd2", size = 408578, upload-time = "2026-05-28T12:00:10.779Z" }, + { url = "https://files.pythonhosted.org/packages/08/d6/070f6a41cbb343e2ac4171859bf3f3623e0ab002f72619d6d505313ec2de/rpds_py-2026.5.1-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:fea6e836d10abbe191d557d33bd58bd5987725fe63aa1eefe557d230209855bd", size = 553573, upload-time = "2026-05-28T12:00:12.443Z" }, + { url = "https://files.pythonhosted.org/packages/75/ab/1a71ea3589c4345dac0a0518f0e6a031cb42689277851b683c46d27463a5/rpds_py-2026.5.1-cp313-cp313t-musllinux_1_2_i686.whl", hash = "sha256:fc0c0f878ea770a0a8a462456c5ad36fc9fe6358e6b76fdadc7f17575e0b8bf1", size = 620861, upload-time = "2026-05-28T12:00:14.09Z" }, + { url = "https://files.pythonhosted.org/packages/8a/22/9bf80a56069c0c443fcfefac639a86a744550a2898817a6dfd3e26654924/rpds_py-2026.5.1-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:e0b360f316d966b048b085857630b3cc51f3db2f07b06f440eac8f695374d1e3", size = 585633, upload-time = "2026-05-28T12:00:15.66Z" }, + { url = "https://files.pythonhosted.org/packages/da/68/3b2c0a75c9e04125696f84ebdbbf304acf5a40b58ba4481cdb98a922c3ba/rpds_py-2026.5.1-cp313-cp313t-win32.whl", hash = "sha256:a2999883eedf72fdfb7520b92c7d4ec2572a71ff40239377aa604cc529eecafc", size = 210074, upload-time = "2026-05-28T12:00:17.291Z" }, + { url = "https://files.pythonhosted.org/packages/e7/8b/609157d5a25d37d4f29f92840ba531f416907c34ae5c5739dd21fc2bef98/rpds_py-2026.5.1-cp313-cp313t-win_amd64.whl", hash = "sha256:e07be2a9d7122bd6e82dea89814ef8dc893feb1aae97fec1630f3263bbb30e55", size = 228635, upload-time = "2026-05-28T12:00:18.73Z" }, + { url = "https://files.pythonhosted.org/packages/d4/6f/19c1918a4b590d8de87e712e4abe4b3875771eff60216fb6153cf6665c68/rpds_py-2026.5.1-cp314-cp314-macosx_10_12_x86_64.whl", hash = "sha256:1f2c391c3059798093b65df23aca2cac150460ae9c630d99dec83d703d9485b9", size = 349756, upload-time = "2026-05-28T12:00:20.217Z" }, + { url = "https://files.pythonhosted.org/packages/e5/60/a06fe7da34eca79dacbf958a2ba0c6eea85bc2b29de20080bf40f72f66fa/rpds_py-2026.5.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:413b424f7c4ee65ab5e5be91f5731be0f8b41a1ee2b12dfe810d716312e95a78", size = 343831, upload-time = "2026-05-28T12:00:21.711Z" }, + { url = "https://files.pythonhosted.org/packages/bf/ec/b2333b97b90e2a6ef6ca8ad386ee284968e74bcfe113b3f1a8d9036429a9/rpds_py-2026.5.1-cp314-cp314-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:2c595a1d9255dce0599e13130d1440ab2506654f2b50294226ee06402f8fef63", size = 375127, upload-time = "2026-05-28T12:00:23.326Z" }, + { url = "https://files.pythonhosted.org/packages/14/7f/e00aae54067f2b488c4637961d5f58204d470795fc791085fa3f15060d2e/rpds_py-2026.5.1-cp314-cp314-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:1c27c5f6102eac8c03e7595a00827a53b271ba40a53b59ff8709170e0855ea4a", size = 379034, upload-time = "2026-05-28T12:00:24.89Z" }, + { url = "https://files.pythonhosted.org/packages/be/cc/423999bbb8ae8dc93c77fc1d5e984ade5eb89d237d3bb884ccfa72ae2890/rpds_py-2026.5.1-cp314-cp314-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:6c7fcf61d44cacecaf3aea542b0e053db77972a4573e7ceda16fb2b399161195", size = 490823, upload-time = "2026-05-28T12:00:26.676Z" }, + { url = "https://files.pythonhosted.org/packages/0f/aa/c671bf660f12e68d3c52ff86c7066ed1372df5a0f4f2ff584e419b8207e7/rpds_py-2026.5.1-cp314-cp314-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:2c817a189d4ee14290420e5ff051e4dd6baa13f3edf84685071dee07a6d538ee", size = 388144, upload-time = "2026-05-28T12:00:28.577Z" }, + { url = "https://files.pythonhosted.org/packages/19/c8/d63bb75b68afe77b229e3021c6031bcaf01da5db5b0e69d0d10f9ba679a7/rpds_py-2026.5.1-cp314-cp314-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:21846aac0ed2e0589f38c12dc44e77bb64e494b771eadbcf169cba00566ba7ba", size = 371959, upload-time = "2026-05-28T12:00:30.304Z" }, + { url = "https://files.pythonhosted.org/packages/82/35/c51122014d8274ff37dc606d60049c3db7d83da02b5b282511e5a906a9a6/rpds_py-2026.5.1-cp314-cp314-manylinux_2_31_riscv64.whl", hash = "sha256:b317c87a13f769a4e787819bd508aaa5d69aa09b0880de9af6d3a8a54571cdec", size = 383558, upload-time = "2026-05-28T12:00:31.764Z" }, + { url = "https://files.pythonhosted.org/packages/e3/f9/2790cb99c136a5363acdeacf5c27c56f3de0d4118a1f48fca83404c99c89/rpds_py-2026.5.1-cp314-cp314-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:ce87129d9f2c14fa6c4a8601fb80eb4488c80d38a20cd13758ef11123e14995d", size = 402789, upload-time = "2026-05-28T12:00:33.247Z" }, + { url = "https://files.pythonhosted.org/packages/e5/1b/e4fb584f8c75d35c38150ff6a332cda949e6f97acba1f4fd123b14ab56fe/rpds_py-2026.5.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:9cdddb6c1207d284d94fd1530adf57fbd797fe7c4b8704ba85f49414f2557e7d", size = 551405, upload-time = "2026-05-28T12:00:34.819Z" }, + { url = "https://files.pythonhosted.org/packages/d8/f7/a6731b4216cb3793ea1af5391da240f5683dacc0d13e034fe5fc3503f240/rpds_py-2026.5.1-cp314-cp314-musllinux_1_2_i686.whl", hash = "sha256:4e237e139f94d3c036fd28eb9f564c99055476ff4ff05cd42be55ce349b5aa02", size = 616975, upload-time = "2026-05-28T12:00:36.268Z" }, + { url = "https://files.pythonhosted.org/packages/2c/ea/2e051a81d95d8e63f4b35a1c463a87e8766bc3d083c067c5dfb6bf220747/rpds_py-2026.5.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ed0954b524873214369184a9c82b0eaa45a3fbb9a798cd95b17e0d98499e7ea0", size = 578701, upload-time = "2026-05-28T12:00:37.82Z" }, + { url = "https://files.pythonhosted.org/packages/65/56/b5f6fdb2083e32bca8a8993d89e70db114b4756c9e2c38421328126689d2/rpds_py-2026.5.1-cp314-cp314-win32.whl", hash = "sha256:2d88621d6a7d4dfa633d21abe90f280bb205274e16b1d1e61c6ad4640b2453b7", size = 209806, upload-time = "2026-05-28T12:00:39.492Z" }, + { url = "https://files.pythonhosted.org/packages/fb/80/65a5aa96c155e611d1ed844e4e1f57f3e36b021f396d9f8585d756e6b90d/rpds_py-2026.5.1-cp314-cp314-win_amd64.whl", hash = "sha256:cef8ac28d26f4dda3533060c20fbf80a325458fa9fd23ea72a73cdfa8e978838", size = 225985, upload-time = "2026-05-28T12:00:40.94Z" }, + { url = "https://files.pythonhosted.org/packages/27/7c/ad185212e87b05f196daef92bc5f3caf07298eb47c295b5585c3dd3093ac/rpds_py-2026.5.1-cp314-cp314-win_arm64.whl", hash = "sha256:eaaea962c68cdc68d4a533ba985ab8e9484277910bbfaa2ab3ef7732667bfed8", size = 221219, upload-time = "2026-05-28T12:00:43.15Z" }, + { url = "https://files.pythonhosted.org/packages/23/58/e14ae18759020334646b031e708ab4158d653a938822bfb7b95ef2e93aa3/rpds_py-2026.5.1-cp314-cp314t-macosx_10_12_x86_64.whl", hash = "sha256:21942f52dbbd5f8758bf021213d28bd45c39e873e65e2407faf5f1846f5761ad", size = 352148, upload-time = "2026-05-28T12:00:44.638Z" }, + { url = "https://files.pythonhosted.org/packages/31/9b/5f4a1e2f960bca3ac5d052b139dd31eed97b259f9d909173821760d542e8/rpds_py-2026.5.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:f414556f6e3958300ff941e40c9f97e3dc9774ddd1b3434c475d73dd354bbed3", size = 345196, upload-time = "2026-05-28T12:00:46.14Z" }, + { url = "https://files.pythonhosted.org/packages/1a/71/1d9574d6a2fa20ab60eaa55c7467f5aa20cbc770f341a05f09c0876f59e2/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ef1013a8625c74043210190b246f5b1551e09757c1f356c6e4160ef96c5bc081", size = 374981, upload-time = "2026-05-28T12:00:47.531Z" }, + { url = "https://files.pythonhosted.org/packages/0c/9a/37e99f4915a80aa71670263c1267f7ae0af95f53a3f61e6c3bdc016d4515/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:cc68e231a77a5f0d774ae278a1f8e55c0456501820847c1e4efb3829f3441df6", size = 379961, upload-time = "2026-05-28T12:00:49.216Z" }, + { url = "https://files.pythonhosted.org/packages/a8/ff/6e73f74b89d2e0715e0fc86b7dde893f9a61ae2f9b256ff3bdfe41ac4e94/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:9baffb505aff33acc69b422a19f77806680f3c8632227d79f48de8a810d1c2c5", size = 495965, upload-time = "2026-05-28T12:00:51.111Z" }, + { url = "https://files.pythonhosted.org/packages/ea/e0/425faba25f59d74d4638b267f7c7a80e8649d2ef4db10a19b0c4a71e6e6f/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b8d2f912928d426e8cfa396f7f3f8d29a59e6689c86dcca3c420730c1096322b", size = 389526, upload-time = "2026-05-28T12:00:52.77Z" }, + { url = "https://files.pythonhosted.org/packages/c6/76/7a41960e3fddae47fab43a28684d5da981401dffd88253de0944148654cb/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:90f628283be835db980c941767d41c9a27b5239e54ba0a9c1335247e82406964", size = 376190, upload-time = "2026-05-28T12:00:54.215Z" }, + { url = "https://files.pythonhosted.org/packages/27/60/5f38dc70824fc6951b51d35377e577a3a3a4c81a6769cc5a2de25ebe0ad1/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_31_riscv64.whl", hash = "sha256:1ebb2f0ab7e16132995a72de805170e0203df0c3dd22e1ef1cd1fdd90bd7a131", size = 383921, upload-time = "2026-05-28T12:00:55.673Z" }, + { url = "https://files.pythonhosted.org/packages/60/1a/d60a38caa1505f4b9483c3fbbde12c94e1079154f4f401a6da96f7e77621/rpds_py-2026.5.1-cp314-cp314t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:f3df3d16ded76f1f8c9cdebd0e1ea55fdf4c23b812de189814da7cf229c22a81", size = 404766, upload-time = "2026-05-28T12:00:57.518Z" }, + { url = "https://files.pythonhosted.org/packages/87/ff/602fd3f174d6425f0bce05ad0dfbec0e96b38d0f7d08a79af5aa20083885/rpds_py-2026.5.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:9af8905b8f854990e40d5206aa5ac58d9b0fe0b7f351ff2bb086c20f6c8c6a47", size = 551343, upload-time = "2026-05-28T12:00:58.978Z" }, + { url = "https://files.pythonhosted.org/packages/b8/c1/1be13327acdbead3eca1fde03b6a34dbb011f1e864e217f0d32cc1779a7f/rpds_py-2026.5.1-cp314-cp314t-musllinux_1_2_i686.whl", hash = "sha256:036a36a87fb1cd3b214d11c4b3c4f7d2ddad933625dca1c900b56a057c07740a", size = 618502, upload-time = "2026-05-28T12:01:00.656Z" }, + { url = "https://files.pythonhosted.org/packages/f3/d7/afb49b49d7f2be8b7ba1a9f0977fa5168003437b93086726f066544e8351/rpds_py-2026.5.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:62ae3853454fe9ef283a03c96c2d835d39e84b14643a9d62c82ef0fb87d702ca", size = 581916, upload-time = "2026-05-28T12:01:02.22Z" }, + { url = "https://files.pythonhosted.org/packages/25/d1/dbef8c1f8a10f07beb62b5f054e20099fd9924b3ec001b8f0b6ac7813a85/rpds_py-2026.5.1-cp314-cp314t-win32.whl", hash = "sha256:6c3d771a46ec18b12af06ce36243a9a80b07a5d0515236332d90863ca8bb326a", size = 207855, upload-time = "2026-05-28T12:01:03.821Z" }, + { url = "https://files.pythonhosted.org/packages/2a/72/bfa4e61ab8e7dc1c8adf397e05e6cbdd4239357bd72b248d3de662f23915/rpds_py-2026.5.1-cp314-cp314t-win_amd64.whl", hash = "sha256:c93c629be4636cf54337bd5f06c104d55e42ced54d681f6fe21ae510a65116f6", size = 225422, upload-time = "2026-05-28T12:01:05.194Z" }, + { url = "https://files.pythonhosted.org/packages/27/3a/7b5da92b640f67b6717ccafc83cdd06bfa7ff2395c3685c68922bb54d703/rpds_py-2026.5.1-cp315-cp315-macosx_10_12_x86_64.whl", hash = "sha256:3574b55c604b8f75dacb007136508bbc0db406e626301778096a133327e7f2fb", size = 349576, upload-time = "2026-05-28T12:01:06.722Z" }, + { url = "https://files.pythonhosted.org/packages/d7/8a/2aafd7ad355a1bd48ca76e2262b74b15e6432b5a1efe150efd4d779cd55d/rpds_py-2026.5.1-cp315-cp315-macosx_11_0_arm64.whl", hash = "sha256:94068eb3ae6d43f5a786b7db96a406a34e6d5c24489feef32fd6e8946ea7b291", size = 343640, upload-time = "2026-05-28T12:01:08.441Z" }, + { url = "https://files.pythonhosted.org/packages/f7/7d/6c9523c1abbe840a1b7fba3c516d48e1d3487cc80fea4366c4071cf56784/rpds_py-2026.5.1-cp315-cp315-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f3a5b10e8ce894825f380a8f1b6444cf73c294dfea62afbb2d13e3a9e630cec1", size = 375322, upload-time = "2026-05-28T12:01:09.934Z" }, + { url = "https://files.pythonhosted.org/packages/5a/5d/0b7b03fb1dc509321f01de3149784ab773e34c8573022029af8076afcb9c/rpds_py-2026.5.1-cp315-cp315-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:fc09f82e63d4bcd58149572f857a431bae851dc747e313c3b5bdf7abb907fda8", size = 379066, upload-time = "2026-05-28T12:01:11.48Z" }, + { url = "https://files.pythonhosted.org/packages/d7/e2/8ef6012999ebf1cb1c22f876d9ce5e63d960fd4631d2af3202d3f480aa25/rpds_py-2026.5.1-cp315-cp315-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:e10464d17df3b582745c25cec695cb9558bca2cb6ddb631aee1787fc72c767b2", size = 494586, upload-time = "2026-05-28T12:01:13.051Z" }, + { url = "https://files.pythonhosted.org/packages/80/af/1eeb029bec67582c226b7809172207cd005073af4ebd906e65ff494f4983/rpds_py-2026.5.1-cp315-cp315-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:ba05adbf15d994c38ec0b7ab32e858e5110c21e9009a00a86545fd220f84e038", size = 388415, upload-time = "2026-05-28T12:01:14.631Z" }, + { url = "https://files.pythonhosted.org/packages/18/23/ffbe10711c4d766c1cab0557d6906c074f795814863c67b351355d29354a/rpds_py-2026.5.1-cp315-cp315-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:77c004fdc7b891967106f78ddfd7b076bfe6813c6139c6fff6aed3bcaa960b26", size = 372427, upload-time = "2026-05-28T12:01:16.153Z" }, + { url = "https://files.pythonhosted.org/packages/bd/3a/30ba4a6ad457e5b070c18d742a33fb77d8d922b565cc881f8a5313d63bfe/rpds_py-2026.5.1-cp315-cp315-manylinux_2_31_riscv64.whl", hash = "sha256:83bcf894486c9d78dd290d3c0124ff6dd8875d3025e2090a8ec49fcc37c55fdd", size = 383615, upload-time = "2026-05-28T12:01:17.809Z" }, + { url = "https://files.pythonhosted.org/packages/d3/69/62e242b53ce39c0814bd24e1a6e6eba6c92be716277745f317f9540a2e7b/rpds_py-2026.5.1-cp315-cp315-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:c3df104083952a0e0c6f10de33e440eabe98fb6317d23e1a58c68f6df08d01b9", size = 402786, upload-time = "2026-05-28T12:01:19.419Z" }, + { url = "https://files.pythonhosted.org/packages/38/c1/a770b9c186928a1ed0f7e6d7ae50e7f3950ed23e3f9e366dbc8e38cb55de/rpds_py-2026.5.1-cp315-cp315-musllinux_1_2_aarch64.whl", hash = "sha256:980450826cf22e133c57e0835070bdd0dd3f73b9b708c3ce223def2cb9469e14", size = 551583, upload-time = "2026-05-28T12:01:21.013Z" }, + { url = "https://files.pythonhosted.org/packages/21/7c/68e8579b95375b70d2a963103c42e705856cdb98569258bd807f4423891c/rpds_py-2026.5.1-cp315-cp315-musllinux_1_2_i686.whl", hash = "sha256:205dde846f24332ab0c1188699a043b8d165b79bb84529ce272c45048ff6be01", size = 616941, upload-time = "2026-05-28T12:01:22.548Z" }, + { url = "https://files.pythonhosted.org/packages/70/a1/a6135aed5730ff03ab957182259987ac11e55fb392a28dc6f0592048a280/rpds_py-2026.5.1-cp315-cp315-musllinux_1_2_x86_64.whl", hash = "sha256:3966b82dd563176396df030f3dd52a6e54cb69b718e95e78bd555ed3d1e0185d", size = 578349, upload-time = "2026-05-28T12:01:24.118Z" }, + { url = "https://files.pythonhosted.org/packages/09/6e/f24201a76a84e6c49d0bdfdfcb735210e21701e9b21c5bfc0ba497dd62f6/rpds_py-2026.5.1-cp315-cp315-win32.whl", hash = "sha256:7818f8d0a415be74d2be3590b0a1c1f463a642f4d0217e7d10602dceef5b79aa", size = 209922, upload-time = "2026-05-28T12:01:25.522Z" }, + { url = "https://files.pythonhosted.org/packages/9e/e4/966bc240bb0485fc265278f6de44d05834bf0b3618886e0b22e33d54c49a/rpds_py-2026.5.1-cp315-cp315-win_amd64.whl", hash = "sha256:b3cc20c0d800af78fd0fac68086e28c1856cec51ea528bb81ea851aa40d39325", size = 226003, upload-time = "2026-05-28T12:01:27.062Z" }, + { url = "https://files.pythonhosted.org/packages/5c/5c/a15a59269cd5e74472734516c73795c15eccfc841b3d4b0228c3f53f19d0/rpds_py-2026.5.1-cp315-cp315-win_arm64.whl", hash = "sha256:3609e9939a8a76cd904cf98a3f1f13b5dc7e150adeaee89e0ea09652ea213e16", size = 221245, upload-time = "2026-05-28T12:01:28.51Z" }, + { url = "https://files.pythonhosted.org/packages/e0/22/135ce03804e179a71ceb13be095deda4a279bc88f7a6b8fa161c5ad44e12/rpds_py-2026.5.1-cp315-cp315t-macosx_10_12_x86_64.whl", hash = "sha256:5d333a7127d4b307601ac37792bee01bb95c867cbfacf21b6375b804d6bbd723", size = 352015, upload-time = "2026-05-28T12:01:30.214Z" }, + { url = "https://files.pythonhosted.org/packages/3b/5f/f1f6d2652eb9d848f6eb369d8db83a2da6249bb49ad2c2a48f45d54538d3/rpds_py-2026.5.1-cp315-cp315t-macosx_11_0_arm64.whl", hash = "sha256:b5f077b44a4f7808520f66dae234988d867deb9aed9be5da057ce9ba831b2a41", size = 345016, upload-time = "2026-05-28T12:01:31.656Z" }, + { url = "https://files.pythonhosted.org/packages/88/66/b74182775691ea2290c99e52ac8d5db844e56fbec90ce421f107658c8314/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:55d8f9b7b78c9538fc9e04e82ec0e888ff0c3cffcfad152c77e57cd09351a98a", size = 374775, upload-time = "2026-05-28T12:01:33.136Z" }, + { url = "https://files.pythonhosted.org/packages/ff/8f/15e5a61d9f0a43902d36561d4f07cae6ae9f4716be825159fd72717f33af/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:e3a8ae58895ac107ed934a6bf51e5846f95c53b9b940c2c6d310838fd5846358", size = 380270, upload-time = "2026-05-28T12:01:34.574Z" }, + { url = "https://files.pythonhosted.org/packages/02/c3/f859b12763a80540cdf2af0f15b19904cf756a71d7bdd3f82ff3e5b1bbf9/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:0957cf3c2b8632ec7aaebffebea8005b353cc2a237b6e2ae3c2cac0820704cfb", size = 495285, upload-time = "2026-05-28T12:01:36.127Z" }, + { url = "https://files.pythonhosted.org/packages/1c/c7/ff27c2ac8411d30b03b1829fd88cae8dad1a4d0da48dd25e57c4038042e6/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:c396c1304de421050b3681ea70f371874b54d41b0151e96109758144c231e30b", size = 389581, upload-time = "2026-05-28T12:01:37.635Z" }, + { url = "https://files.pythonhosted.org/packages/6e/67/fe92ee32a6cc05c77228a2f8b1762e7124f386ec20ff83d0757b762d58d0/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:aad1bff7f666b9598e573815affd666aac6a13a585dde336f843e33350c7fadc", size = 376041, upload-time = "2026-05-28T12:01:39.307Z" }, + { url = "https://files.pythonhosted.org/packages/f8/91/b4d6685c27aba55bd82f25b278be8237038117d05f9659a6213ad3408130/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_31_riscv64.whl", hash = "sha256:656a042550878f12d45752452d47094b7cfe5ad1e9d7b87b5a22ad3ae5ff8015", size = 383946, upload-time = "2026-05-28T12:01:41.043Z" }, + { url = "https://files.pythonhosted.org/packages/bd/79/2c1d832a53c8e0f8e98fc970ec257b950fecd4f62be2ab7182b500a0cbc8/rpds_py-2026.5.1-cp315-cp315t-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:73c4bd4f70294737b5206a3e8e30ccadbf8a60301831c8ea23eec5dbeea1ecfa", size = 405526, upload-time = "2026-05-28T12:01:43.032Z" }, + { url = "https://files.pythonhosted.org/packages/78/c4/c98117b03c6a8581ab2c2dfccfe9a5ad82bd8128a3c28b46a6ad2d97c393/rpds_py-2026.5.1-cp315-cp315t-musllinux_1_2_aarch64.whl", hash = "sha256:43bca78665423cabae77146f2fe7ce55272b6c8d55d82cca83effd42c7e13972", size = 551165, upload-time = "2026-05-28T12:01:44.648Z" }, + { url = "https://files.pythonhosted.org/packages/3b/c1/bc479ca069200af730881b1bd525e3114b2b391a351509fcb1b772f28086/rpds_py-2026.5.1-cp315-cp315t-musllinux_1_2_i686.whl", hash = "sha256:42d0f20e85e549c870749d0e247f0c10d318a45b7e9676d575d2dcb04a1b2e66", size = 618778, upload-time = "2026-05-28T12:01:46.337Z" }, + { url = "https://files.pythonhosted.org/packages/77/65/38ab2f90df44c2febfb63cc10ced40763d9b4bc94d173e734528663fe7f5/rpds_py-2026.5.1-cp315-cp315t-musllinux_1_2_x86_64.whl", hash = "sha256:b1be5c35683684d5331b93600c210e8367c254683d8a6df6bd21bd2da3a334fb", size = 581839, upload-time = "2026-05-28T12:01:48.109Z" }, + { url = "https://files.pythonhosted.org/packages/15/2d/ce1f605fe036aadd460e5822e578c6c7ec3a860936cca37d6e0f299daa77/rpds_py-2026.5.1-cp315-cp315t-win32.whl", hash = "sha256:75808f6c38ce7749bb68cc2770161aae5045e6c6f6781a9782e74b93304399df", size = 207866, upload-time = "2026-05-28T12:01:49.648Z" }, + { url = "https://files.pythonhosted.org/packages/79/cb/966040123eb102371559746908ef2c9471f4d43e17ec9a645a2258dab64b/rpds_py-2026.5.1-cp315-cp315t-win_amd64.whl", hash = "sha256:90bd6630002a1c7f09e7843dd79f0d24f3d2897cc25a753480917865d14f15b3", size = 225441, upload-time = "2026-05-28T12:01:51.408Z" }, +] + [[package]] name = "six" version = "1.17.0" diff --git a/pipeline/workflow/spanner-ingestion-workflow.yaml b/pipeline/workflow/spanner-ingestion-workflow.yaml index d0a878067..4bc2d5c6e 100644 --- a/pipeline/workflow/spanner-ingestion-workflow.yaml +++ b/pipeline/workflow/spanner-ingestion-workflow.yaml @@ -121,38 +121,52 @@ main: run_aggregation_job: params: [import_list, helper_url] steps: - - run_aggregation: + # 1. Start the aggregation and get the initial state + - initiate_aggregation: call: http.post args: - url: ${helper_url + "/aggregation/run"} + url: ${helper_url + "/aggregation/initiate"} timeout: 300 auth: type: OIDC body: importList: ${import_list} - result: aggregation_response - - check_aggregation_status_loop: + result: initiate_response + # 2. Store the state in a workflow variable + - assign_state: + assign: + - state: ${initiate_response.body} + # 3. State Check loop + - check_status_loop: + switch: + # Exit successfully if done + - condition: ${state.status == "SUCCEEDED"} + return: "OK" + # Raise error if failed + - condition: ${state.status == "FAILED"} + raise: ${state.error} + next: poll_and_wait + # 4. Sleep and Poll + - poll_and_wait: steps: - - wait_for_aggregation: + - wait_step: call: sys.sleep args: seconds: 300 - - check_aggregation_status: + # Pass the state back to the server, get the new state + - poll_server: call: http.post args: - url: ${helper_url + "/aggregation/status"} + url: ${helper_url + "/aggregation/poll"} auth: type: OIDC - body: - jobIds: ${aggregation_response.body.jobIds} - result: aggregation_status_response - - evaluate_aggregation_status: - switch: - - condition: ${aggregation_status_response.body.status == "DONE"} - return: 'OK' - - condition: ${aggregation_status_response.body.status == "FAILED"} - raise: ${aggregation_status_response.body.error} - next: check_aggregation_status_loop + body: ${state} + result: poll_response + - update_state: + assign: + - state: ${poll_response.body} + next: check_status_loop + # This sub-workflow launches a Dataflow job and waits for it to complete. run_dataflow_job: