From 8e8262fdcab6fc4e1839bd349ee9d2dfa36f98a8 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Fri, 23 Oct 2026 17:35:13 +0200 Subject: [PATCH 1/3] chore: drop the DCP preprocessing job name from ingestion Data Commons Platform 1.1.6 deletes the Cloud Run preprocessing job, and dcp-tools 1.1.0 starts the ingestion workflow without its name. KnowledgeGraphResource loses prep_job_name, definitions.py stops reading DCP_PREP_JOB_NAME, and .env.example and the test fixtures drop it. The provenance expectation in test_export.py follows dcp-tools 1.1.0, which writes importedSourceReleaseDate as a declared Provenance field. --- .env.example | 1 - src/dfo_pipelines/definitions.py | 1 - src/dfo_pipelines/defs/resources.py | 2 -- tests/data_commons/test_export.py | 2 +- tests/data_commons/test_ingestion.py | 1 - tests/dfo/ontology/test_assets.py | 1 - tests/dfo/shared/test_assets.py | 1 - tests/dfo/skills/test_assets.py | 1 - tests/imf/weo/test_assets.py | 1 - tests/oecd/dac1/test_assets.py | 1 - tests/oecd/dac2a/test_assets.py | 1 - tests/oecd/dac2b/test_assets.py | 1 - tests/oecd/dac_crs/test_assets.py | 1 - tests/one/conversion_tables/test_assets.py | 1 - tests/one/imputed_multilateral_sectors/test_assets.py | 1 - tests/world_bank/ids/test_assets.py | 1 - 16 files changed, 1 insertion(+), 17 deletions(-) diff --git a/.env.example b/.env.example index 308235b..c932b38 100644 --- a/.env.example +++ b/.env.example @@ -15,7 +15,6 @@ DUCKDB_PATH=path/to/repo/data/dfo.duckdb # DCP_PROJECT_ID= # DCP_REGION= # DCP_BUCKET_NAME= -# DCP_PREP_JOB_NAME= # DCP_WORKFLOW_NAME= # The GitHub repository the DFO agent skills are fetched from. Only the skills assets read it. diff --git a/src/dfo_pipelines/definitions.py b/src/dfo_pipelines/definitions.py index c41ea3f..9eb8b87 100644 --- a/src/dfo_pipelines/definitions.py +++ b/src/dfo_pipelines/definitions.py @@ -26,7 +26,6 @@ def defs(): project_id=dg.EnvVar("DCP_PROJECT_ID"), region=dg.EnvVar("DCP_REGION"), bucket_name=dg.EnvVar("DCP_BUCKET_NAME"), - prep_job_name=dg.EnvVar("DCP_PREP_JOB_NAME"), workflow_name=dg.EnvVar("DCP_WORKFLOW_NAME"), ), "skills_repository": SkillsRepositoryResource( diff --git a/src/dfo_pipelines/defs/resources.py b/src/dfo_pipelines/defs/resources.py index 5ebcebc..67d5900 100644 --- a/src/dfo_pipelines/defs/resources.py +++ b/src/dfo_pipelines/defs/resources.py @@ -35,7 +35,6 @@ class KnowledgeGraphResource(dg.ConfigurableResource): project_id: str region: str bucket_name: str - prep_job_name: str workflow_name: str service_account: str | None = None @@ -46,7 +45,6 @@ def settings(self) -> KGSettings: gcp_project_id=self.project_id, gcp_region=self.region, gcs_bucket_name=self.bucket_name, - ingestion_prep_job_name=self.prep_job_name, ingestion_workflow_name=self.workflow_name, ingestion_service_account=self.service_account, # Required by KGSettings but read by nothing on the ingestion path, diff --git a/tests/data_commons/test_export.py b/tests/data_commons/test_export.py index 30cae49..fe6ad82 100644 --- a/tests/data_commons/test_export.py +++ b/tests/data_commons/test_export.py @@ -531,11 +531,11 @@ def test_export_bundle_writes_the_provenance_from_the_spec_and_model( 'licenseAttribution: "Source One (2026), Provenance One ' '(accessed on 2026-03-04)"\n' 'lastDataRefreshDate: "2026-10-14"\n' + 'importedSourceReleaseDate: "2026-02-15"\n' 'nextSourceReleaseDate: "2026-06"\n' 'sourceReleaseFrequency: "P1Y"\n' 'earliestObservationDate: "2000"\n' 'latestObservationDate: "2001"\n' - 'importedSourceReleaseDate: "2026-02-15"\n' 'dataTransformationDescription: "## Step one\\n\\nWhat the step does.\\n"\n' 'dataTransformationChangelog: "## 2026-01-01\\n\\n### Added\\n\\n' '- Dataset added to the graph.\\n"' diff --git a/tests/data_commons/test_ingestion.py b/tests/data_commons/test_ingestion.py index 99ec774..9b26b73 100644 --- a/tests/data_commons/test_ingestion.py +++ b/tests/data_commons/test_ingestion.py @@ -18,7 +18,6 @@ project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) diff --git a/tests/dfo/ontology/test_assets.py b/tests/dfo/ontology/test_assets.py index 4dde07a..bc2bf29 100644 --- a/tests/dfo/ontology/test_assets.py +++ b/tests/dfo/ontology/test_assets.py @@ -336,7 +336,6 @@ def test_ingestion_ingests_the_bundle_the_export_wrote( project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" diff --git a/tests/dfo/shared/test_assets.py b/tests/dfo/shared/test_assets.py index c684023..1741e17 100644 --- a/tests/dfo/shared/test_assets.py +++ b/tests/dfo/shared/test_assets.py @@ -65,7 +65,6 @@ def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" diff --git a/tests/dfo/skills/test_assets.py b/tests/dfo/skills/test_assets.py index 9add563..e42bf53 100644 --- a/tests/dfo/skills/test_assets.py +++ b/tests/dfo/skills/test_assets.py @@ -358,7 +358,6 @@ def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" diff --git a/tests/imf/weo/test_assets.py b/tests/imf/weo/test_assets.py index 01a8e12..bb116a1 100644 --- a/tests/imf/weo/test_assets.py +++ b/tests/imf/weo/test_assets.py @@ -644,7 +644,6 @@ def test_imf_weo_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) - project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = IMF_WEO_EXPORT_SPEC.import_name diff --git a/tests/oecd/dac1/test_assets.py b/tests/oecd/dac1/test_assets.py index 15d91b0..c03f0ae 100644 --- a/tests/oecd/dac1/test_assets.py +++ b/tests/oecd/dac1/test_assets.py @@ -693,7 +693,6 @@ def test_oecd_dac1_ingestion_ingests_the_bundle_the_export_wrote( project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = OECD_DAC1_EXPORT_SPEC.import_name diff --git a/tests/oecd/dac2a/test_assets.py b/tests/oecd/dac2a/test_assets.py index f934839..84440a0 100644 --- a/tests/oecd/dac2a/test_assets.py +++ b/tests/oecd/dac2a/test_assets.py @@ -674,7 +674,6 @@ def test_oecd_dac2a_ingestion_ingests_the_bundle_the_export_wrote( project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = OECD_DAC2A_EXPORT_SPEC.import_name diff --git a/tests/oecd/dac2b/test_assets.py b/tests/oecd/dac2b/test_assets.py index 07873f6..0c07e30 100644 --- a/tests/oecd/dac2b/test_assets.py +++ b/tests/oecd/dac2b/test_assets.py @@ -666,7 +666,6 @@ def test_oecd_dac2b_ingestion_ingests_the_bundle_the_export_wrote( project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = OECD_DAC2B_EXPORT_SPEC.import_name diff --git a/tests/oecd/dac_crs/test_assets.py b/tests/oecd/dac_crs/test_assets.py index 5867568..9224de3 100644 --- a/tests/oecd/dac_crs/test_assets.py +++ b/tests/oecd/dac_crs/test_assets.py @@ -695,7 +695,6 @@ def test_dac_crs_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) - project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = CRS_EXPORT_SPEC.import_name diff --git a/tests/one/conversion_tables/test_assets.py b/tests/one/conversion_tables/test_assets.py index b55a22e..0d4203d 100644 --- a/tests/one/conversion_tables/test_assets.py +++ b/tests/one/conversion_tables/test_assets.py @@ -295,7 +295,6 @@ def test_dc_ingestion_ingests_the_bundle_the_export_wrote( project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) destination = f"gs://test-bucket/ingestion/input/{spec.import_name}" diff --git a/tests/one/imputed_multilateral_sectors/test_assets.py b/tests/one/imputed_multilateral_sectors/test_assets.py index ddd3e12..384dc0b 100644 --- a/tests/one/imputed_multilateral_sectors/test_assets.py +++ b/tests/one/imputed_multilateral_sectors/test_assets.py @@ -1345,7 +1345,6 @@ def test_dc_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> Non project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = IMPUTED_MULTILATERAL_SECTORS_EXPORT_SPEC.import_name diff --git a/tests/world_bank/ids/test_assets.py b/tests/world_bank/ids/test_assets.py index 948ada3..8f22b3b 100644 --- a/tests/world_bank/ids/test_assets.py +++ b/tests/world_bank/ids/test_assets.py @@ -1137,7 +1137,6 @@ def test_ids_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> No project_id="test-project", region="test-region", bucket_name="test-bucket", - prep_job_name="test-prep-job", workflow_name="test-workflow", ) import_name = IDS_EXPORT_SPEC.import_name From e2d79640b15279eb87c6bca2b237d0af65dc8467 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Fri, 23 Oct 2026 18:21:08 +0200 Subject: [PATCH 2/3] feat(ingestion): refuse the wrong stack and watch the execution start ingest_bundle checks the DCP workflow contract with dcp-tools before it uploads anything, so a stack older than 1.1.6 gets no bundle and the run fails with WorkflowContractError. After starting the workflow it watches the execution for up to two minutes and fails the run on a FAILED or CANCELLED state. Each ingestion asset records the execution name as metadata. The GCS client gets the project explicitly. KnowledgeGraphResource builds dcp-tools settings from its own values only, so environment variables such as GCS_INPUT_FOLDER_PATH no longer override it. The export passes the source release date through add_provenance's imported_source_release_date keyword. Tests share one knowledge_graph fixture and compare provenance as parsed properties. --- README.md | 6 + src/dfo_pipelines/data_commons/export.py | 17 +- src/dfo_pipelines/data_commons/ingestion.py | 121 ++++++++- src/dfo_pipelines/defs/dfo/ontology/assets.py | 5 +- src/dfo_pipelines/defs/dfo/shared/assets.py | 5 +- src/dfo_pipelines/defs/dfo/skills/assets.py | 5 +- src/dfo_pipelines/defs/imf/weo/assets.py | 5 +- src/dfo_pipelines/defs/oecd/dac1/assets.py | 5 +- src/dfo_pipelines/defs/oecd/dac2a/assets.py | 5 +- src/dfo_pipelines/defs/oecd/dac2b/assets.py | 5 +- src/dfo_pipelines/defs/oecd/dac_crs/assets.py | 5 +- .../defs/one/conversion_tables/assets.py | 5 +- .../imputed_multilateral_sectors/assets.py | 5 +- src/dfo_pipelines/defs/resources.py | 28 +- .../defs/world_bank/ids/assets.py | 5 +- tests/conftest.py | 14 + tests/data_commons/test_export.py | 50 ++-- tests/data_commons/test_ingestion.py | 240 +++++++++++++++--- tests/dfo/ontology/test_assets.py | 14 +- tests/dfo/shared/test_assets.py | 16 +- tests/dfo/skills/test_assets.py | 16 +- tests/helpers.py | 6 + tests/imf/weo/test_assets.py | 15 +- tests/oecd/dac1/test_assets.py | 12 +- tests/oecd/dac2a/test_assets.py | 12 +- tests/oecd/dac2b/test_assets.py | 12 +- tests/oecd/dac_crs/test_assets.py | 16 +- tests/one/conversion_tables/test_assets.py | 13 +- .../test_assets.py | 19 +- tests/world_bank/ids/test_assets.py | 15 +- 30 files changed, 506 insertions(+), 191 deletions(-) create mode 100644 tests/conftest.py diff --git a/README.md b/README.md index 365af87..3d81e36 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,12 @@ Automation runs through one sensor, `automation_condition_sensor`. On a new loca On a fresh environment, materialize the dbt seeds, and any other asset with no parents, once by hand. `eager()` never builds an asset with no parents, so nothing downstream of them automates until they exist. +### Ingestion + +Each `*_dc_ingestion` asset syncs its export's bundle to the ingestion bucket, starts the Data Commons Platform ingestion workflow, and watches the execution for two minutes. An execution that fails or is cancelled in that time fails the asset. One still running is left to finish, and the asset records its resource name as `execution_name` metadata, the name to look it up by in Cloud Workflows. + +The identity that starts the workflow also needs `roles/workflows.viewer`. The asset reads the workflow's definition before it uploads the bundle and stops if the workflow does not come from a Data Commons Platform 1.1.6 stack, and the asset reads the execution's state while it watches. + ## Running dbt The dbt models can also be built without Dagster. Run this from the repository root: diff --git a/src/dfo_pipelines/data_commons/export.py b/src/dfo_pipelines/data_commons/export.py index fd9d2a7..d82cee5 100644 --- a/src/dfo_pipelines/data_commons/export.py +++ b/src/dfo_pipelines/data_commons/export.py @@ -53,10 +53,9 @@ # The provenance property `ExportSpec.conversion_recipe` is written onto. CONVERSION_RECIPE_PROPERTY = "conversionRecipe" -# The provenance model column every pipeline emits, and the property it is written -# onto. +# The provenance model column every pipeline emits, written onto the provenance as +# its `importedSourceReleaseDate`. SOURCE_RELEASE_DATE_COLUMN = "source_release_date" -SOURCE_RELEASE_DATE_PROPERTY = "importedSourceReleaseDate" class ConversionRecipe(StrEnum): @@ -157,7 +156,7 @@ class ExportSpec: def __post_init__(self) -> None: reserved = DOCUMENTATION_FILES.keys() | { CONVERSION_RECIPE_PROPERTY, - SOURCE_RELEASE_DATE_PROPERTY, + "importedSourceReleaseDate", "lastDataRefreshDate", } clashes = sorted(set(self.provenance_column_to_property.values()) & reserved) @@ -280,17 +279,16 @@ def export_bundle( log.info("Registering Provenance nodes") (provenance,) = get_model_rows(connection, spec.provenance_model) - column_to_property = spec.provenance_column_to_property | { - SOURCE_RELEASE_DATE_COLUMN: SOURCE_RELEASE_DATE_PROPERTY - } - if missing := column_to_property.keys() - provenance.keys(): + required = spec.provenance_column_to_property.keys() | {SOURCE_RELEASE_DATE_COLUMN} + if missing := required - provenance.keys(): raise ValueError( f"{spec.provenance_model} lacks the mapped columns {sorted(missing)}" ) mapped_properties = { property_name: _provenance_property_value(column, provenance.pop(column)) - for column, property_name in column_to_property.items() + for column, property_name in spec.provenance_column_to_property.items() } + source_release_date = provenance.pop(SOURCE_RELEASE_DATE_COLUMN) recipe = ( {} if spec.conversion_recipe is None @@ -306,6 +304,7 @@ def export_bundle( manager.add_provenance( **provenance, dcid=spec.provenance_dcid, + imported_source_release_date=source_release_date, additional_properties=recipe | mapped_properties | documentation, ) diff --git a/src/dfo_pipelines/data_commons/ingestion.py b/src/dfo_pipelines/data_commons/ingestion.py index 709cbd0..3dcb14a 100644 --- a/src/dfo_pipelines/data_commons/ingestion.py +++ b/src/dfo_pipelines/data_commons/ingestion.py @@ -5,39 +5,129 @@ workflow that reads it. An ingestion rebuilds every node and edge carrying the import's provenance and leaves the rest of the graph alone, which is why the folder holds that import's current bundle and nothing else. + +The workflow runs for as long as the ingestion takes, which is longer than an +asset should wait. Its execution is watched for its first minutes only, which is +when a bad request or a failed launch shows. """ import logging +import time +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path from dcp_tools.gcp_utilities.clients import get_gcs_client -from dcp_tools.gcp_utilities.ingestion import run_ingestion_workflow +from dcp_tools.gcp_utilities.ingestion import ( + IngestionExecutionState, + check_ingestion_workflow_contract, + get_ingestion_execution_state, + run_ingestion_workflow, +) from dcp_tools.gcp_utilities.settings import KGSettings from dcp_tools.gcp_utilities.storage import sync_directory_to_gcs from upath import UPath +# Workflow execution states that end an execution without ingesting it. +FAILED_STATES = frozenset({"FAILED", "CANCELLED"}) + + +class IngestionExecutionError(RuntimeError): + """The ingestion workflow's execution failed or was cancelled.""" + + +@dataclass(frozen=True, kw_only=True) +class Ingestion: + """An ingestion that has started. + + Attributes: + destination: URI of the folder the bundle was synced to. + execution_name: Resource name of the workflow execution ingesting it. + """ + + destination: str + execution_name: str + + +def watch_execution_start( + settings: KGSettings, + execution_name: str, + *, + log: logging.Logger, + watch_seconds: float, + poll_seconds: float, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, +) -> IngestionExecutionState: + """Poll an ingestion's execution until it succeeds or the watch ends. + + Args: + settings: Knowledge graph the execution ingests into. + execution_name: Resource name of the execution. + log: Logger to use, e.g. an asset's `context.log`. + watch_seconds: How long to poll for before leaving the execution running. + poll_seconds: How long to wait between polls. + sleep: Waits the given number of seconds. + clock: Monotonic time in seconds. + + Returns: + IngestionExecutionState: The last state read, `SUCCEEDED` or still running. + + Raises: + IngestionExecutionError: If the execution fails or is cancelled while + watched. + """ + deadline = clock() + watch_seconds + while True: + execution = get_ingestion_execution_state(settings, execution_name) + if execution.state in FAILED_STATES: + raise IngestionExecutionError( + f"Ingestion execution {execution_name} ended {execution.state}: " + f"{execution.error}" + ) + if execution.state == "SUCCEEDED" or clock() >= deadline: + log.info( + f"Ingestion execution {execution_name} is {execution.state}" + f" at steps {list(execution.current_steps)}" + ) + return execution + sleep(poll_seconds) + def ingest_bundle( - bundle: UPath, *, import_name: str, settings: KGSettings, log: logging.Logger -) -> str: + bundle: UPath, + *, + import_name: str, + settings: KGSettings, + log: logging.Logger, + watch_seconds: float = 120, + poll_seconds: float = 5, +) -> Ingestion: """Sync a bundle into the import's folder in the ingestion bucket and ingest it. The sync removes anything in the folder that the bundle does not contain, so the graph is built from what the export wrote rather than from that plus - whatever an earlier bundle left behind. The workflow is only started: it runs - on for as long as the ingestion takes, after this returns. + whatever an earlier bundle left behind. The workflow is checked before the + sync, so a workflow dcp-tools cannot start leaves the folder untouched. Its + execution is watched for `watch_seconds` and runs on after this returns. Args: bundle: Local directory the export wrote the bundle to. import_name: Name of the import, which names its folder in the bucket. settings: Knowledge graph the bundle is ingested into. log: Logger to use, e.g. an asset's `context.log`. + watch_seconds: How long to watch the execution for after starting it. + poll_seconds: How long to wait between reads of its state. Returns: - str: URI of the folder the bundle was synced to. + Ingestion: Where the bundle was synced to and the execution ingesting it. Raises: ValueError: If the bundle is not on local disk, or holds no files to upload. + WorkflowContractError: If the workflow is not one dcp-tools can start, + raised before anything is synced. + IngestionExecutionError: If the execution fails or is cancelled while + watched. """ # dcp-tools uploads from local disk. if bundle.protocol not in ("", "file"): @@ -52,12 +142,25 @@ def ingest_bundle( ): raise ValueError(f"Found no bundle to ingest at {bundle}") + check_ingestion_workflow_contract(settings) + folder = f"{settings.gcs_input_folder_path}/{import_name}" - bucket = get_gcs_client(settings.gcp_credentials).bucket(settings.gcs_bucket_name) + client = get_gcs_client(settings.gcp_credentials, project=settings.gcp_project_id) + bucket = client.bucket(settings.gcs_bucket_name) sync_directory_to_gcs(bucket, directory, gcs_folder_name=folder) destination = f"gs://{settings.gcs_bucket_name}/{folder}" log.info(f"Synced {bundle} to {destination}") log.info(f"Starting the ingestion workflow for {import_name}") - run_ingestion_workflow(settings, imports=import_name) - return destination + execution_name = run_ingestion_workflow( + settings, imports=import_name, skip_workflow_contract_check=True + ) + log.info(f"Started ingestion execution {execution_name}") + watch_execution_start( + settings, + execution_name, + log=log, + watch_seconds=watch_seconds, + poll_seconds=poll_seconds, + ) + return Ingestion(destination=destination, execution_name=execution_name) diff --git a/src/dfo_pipelines/defs/dfo/ontology/assets.py b/src/dfo_pipelines/defs/dfo/ontology/assets.py index 0659305..c374b76 100644 --- a/src/dfo_pipelines/defs/dfo/ontology/assets.py +++ b/src/dfo_pipelines/defs/dfo/ontology/assets.py @@ -205,7 +205,7 @@ def dfo_ontology_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(dfo_ontology_dc_export), import_name=IMPORT_NAME, settings=knowledge_graph.settings, @@ -213,7 +213,8 @@ def dfo_ontology_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(IMPORT_NAME), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/dfo/shared/assets.py b/src/dfo_pipelines/defs/dfo/shared/assets.py index c2d3594..1d8e930 100644 --- a/src/dfo_pipelines/defs/dfo/shared/assets.py +++ b/src/dfo_pipelines/defs/dfo/shared/assets.py @@ -87,7 +87,7 @@ def dfo_shared_dc_ingestion( for as long as it takes after the workflow is started. It follows the ontology's ingestion, which defines the source the provenance names. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(dfo_shared_dc_export), import_name=IMPORT_NAME, settings=knowledge_graph.settings, @@ -95,7 +95,8 @@ def dfo_shared_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(IMPORT_NAME), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/dfo/skills/assets.py b/src/dfo_pipelines/defs/dfo/skills/assets.py index 3225c5d..f13b76b 100644 --- a/src/dfo_pipelines/defs/dfo/skills/assets.py +++ b/src/dfo_pipelines/defs/dfo/skills/assets.py @@ -693,7 +693,7 @@ def dfo_skills_dc_ingestion( ontology's ingestion, which defines the source, the skill classes and properties, and the nodes skills link to. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(dfo_skills_dc_export), import_name=IMPORT_NAME, settings=knowledge_graph.settings, @@ -701,7 +701,8 @@ def dfo_skills_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(IMPORT_NAME), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/imf/weo/assets.py b/src/dfo_pipelines/defs/imf/weo/assets.py index 5659031..7ef38a4 100644 --- a/src/dfo_pipelines/defs/imf/weo/assets.py +++ b/src/dfo_pipelines/defs/imf/weo/assets.py @@ -273,7 +273,7 @@ def imf_weo_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(imf_weo_dc_export), import_name=IMF_WEO_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -281,7 +281,8 @@ def imf_weo_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(IMF_WEO_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/oecd/dac1/assets.py b/src/dfo_pipelines/defs/oecd/dac1/assets.py index b64e6eb..69b97a8 100644 --- a/src/dfo_pipelines/defs/oecd/dac1/assets.py +++ b/src/dfo_pipelines/defs/oecd/dac1/assets.py @@ -298,7 +298,7 @@ def oecd_dac1_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(oecd_dac1_dc_export), import_name=OECD_DAC1_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -306,7 +306,8 @@ def oecd_dac1_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(OECD_DAC1_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/oecd/dac2a/assets.py b/src/dfo_pipelines/defs/oecd/dac2a/assets.py index abe35f8..2f8571c 100644 --- a/src/dfo_pipelines/defs/oecd/dac2a/assets.py +++ b/src/dfo_pipelines/defs/oecd/dac2a/assets.py @@ -287,7 +287,7 @@ def oecd_dac2a_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(oecd_dac2a_dc_export), import_name=OECD_DAC2A_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -295,7 +295,8 @@ def oecd_dac2a_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(OECD_DAC2A_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/oecd/dac2b/assets.py b/src/dfo_pipelines/defs/oecd/dac2b/assets.py index fe02c80..37315c2 100644 --- a/src/dfo_pipelines/defs/oecd/dac2b/assets.py +++ b/src/dfo_pipelines/defs/oecd/dac2b/assets.py @@ -281,7 +281,7 @@ def oecd_dac2b_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(oecd_dac2b_dc_export), import_name=OECD_DAC2B_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -289,7 +289,8 @@ def oecd_dac2b_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(OECD_DAC2B_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/oecd/dac_crs/assets.py b/src/dfo_pipelines/defs/oecd/dac_crs/assets.py index 39e27bb..a7ace4a 100644 --- a/src/dfo_pipelines/defs/oecd/dac_crs/assets.py +++ b/src/dfo_pipelines/defs/oecd/dac_crs/assets.py @@ -288,7 +288,7 @@ def oecd_dac_crs_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(oecd_dac_crs_dc_export), import_name=CRS_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -296,7 +296,8 @@ def oecd_dac_crs_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(CRS_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/one/conversion_tables/assets.py b/src/dfo_pipelines/defs/one/conversion_tables/assets.py index 0bd48cc..d82a635 100644 --- a/src/dfo_pipelines/defs/one/conversion_tables/assets.py +++ b/src/dfo_pipelines/defs/one/conversion_tables/assets.py @@ -159,7 +159,7 @@ def _ingest( bundle: str, ) -> dg.MaterializeResult: """Ingest the bundle `spec`'s export wrote and report where it was staged.""" - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(bundle), import_name=spec.import_name, settings=knowledge_graph.settings, @@ -167,8 +167,9 @@ def _ingest( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(spec.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/one/imputed_multilateral_sectors/assets.py b/src/dfo_pipelines/defs/one/imputed_multilateral_sectors/assets.py index 85862de..bd0e66f 100644 --- a/src/dfo_pipelines/defs/one/imputed_multilateral_sectors/assets.py +++ b/src/dfo_pipelines/defs/one/imputed_multilateral_sectors/assets.py @@ -595,7 +595,7 @@ def one_imputed_multilateral_sectors_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(one_imputed_multilateral_sectors_dc_export), import_name=IMPUTED_MULTILATERAL_SECTORS_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -603,9 +603,10 @@ def one_imputed_multilateral_sectors_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text( IMPUTED_MULTILATERAL_SECTORS_EXPORT_SPEC.import_name ), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/src/dfo_pipelines/defs/resources.py b/src/dfo_pipelines/defs/resources.py index 67d5900..a79fa6a 100644 --- a/src/dfo_pipelines/defs/resources.py +++ b/src/dfo_pipelines/defs/resources.py @@ -1,9 +1,11 @@ import re from datetime import date from pathlib import Path +from typing import override import dagster as dg from dcp_tools.gcp_utilities.settings import KGSettings +from pydantic_settings import BaseSettings, PydanticBaseSettingsSource from upath import UPath @@ -23,6 +25,27 @@ def build_date_partitioned_path( return UPath(self.base_path) / path / partition +class _ResourceKGSettings(KGSettings): + """KGSettings built from the values passed in and nothing else + + KGSettings otherwise lets the environment and a .env file override those + values, and forbids any variable it does not declare, which is every other + variable in this project. + """ + + @classmethod + @override + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + return (init_settings,) + + class KnowledgeGraphResource(dg.ConfigurableResource): """Data Commons knowledge graph a bundle is ingested into @@ -41,7 +64,7 @@ class KnowledgeGraphResource(dg.ConfigurableResource): @property def settings(self) -> KGSettings: """Settings as dcp-tools expects them""" - return KGSettings( + return _ResourceKGSettings( gcp_project_id=self.project_id, gcp_region=self.region, gcs_bucket_name=self.bucket_name, @@ -51,9 +74,6 @@ def settings(self) -> KGSettings: # which is handed the bundle's directory and never writes an output. local_path=Path("."), gcs_output_folder_path="", - # KGSettings otherwise reads .env itself and forbids any variable it - # does not declare, which is every other variable in this project. - _env_file=None, ) diff --git a/src/dfo_pipelines/defs/world_bank/ids/assets.py b/src/dfo_pipelines/defs/world_bank/ids/assets.py index 0a4883a..44a852a 100644 --- a/src/dfo_pipelines/defs/world_bank/ids/assets.py +++ b/src/dfo_pipelines/defs/world_bank/ids/assets.py @@ -454,7 +454,7 @@ def world_bank_ids_dc_ingestion( Materialized by hand only: an ingestion rewrites the shared graph and runs for as long as it takes after the workflow is started. """ - destination = ingest_bundle( + ingestion = ingest_bundle( UPath(world_bank_ids_dc_export), import_name=IDS_EXPORT_SPEC.import_name, settings=knowledge_graph.settings, @@ -462,7 +462,8 @@ def world_bank_ids_dc_ingestion( ) return dg.MaterializeResult( metadata={ - "dagster/uri": dg.MetadataValue.text(destination), + "dagster/uri": dg.MetadataValue.text(ingestion.destination), "import_name": dg.MetadataValue.text(IDS_EXPORT_SPEC.import_name), + "execution_name": dg.MetadataValue.text(ingestion.execution_name), }, ) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..55df36d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,14 @@ +import pytest + +from dfo_pipelines.defs.resources import KnowledgeGraphResource + + +@pytest.fixture +def knowledge_graph() -> KnowledgeGraphResource: + """Knowledge graph a test hands an ingestion, never reached""" + return KnowledgeGraphResource( + project_id="test-project", + region="test-region", + bucket_name="test-bucket", + workflow_name="test-workflow", + ) diff --git a/tests/data_commons/test_export.py b/tests/data_commons/test_export.py index fe6ad82..f778b4d 100644 --- a/tests/data_commons/test_export.py +++ b/tests/data_commons/test_export.py @@ -182,6 +182,16 @@ def statvar_nodes(out_dir: Path) -> dict[str, str]: } +def mcf_properties(mcf: str) -> dict[str, str]: + """One MCF node's values keyed on property, which must each appear once""" + properties: dict[str, str] = {} + for line in mcf.strip().splitlines(): + name, _, value = line.partition(": ") + assert name not in properties, f"{name} repeats" + properties[name] = value + return properties + + def test_export_bundle_writes_a_bundle(tmp_path: Path) -> None: """Test the export writes a bundle of MCF, observations and a config""" metadata, out_dir = run_export(tmp_path) @@ -519,27 +529,25 @@ def test_export_bundle_writes_the_provenance_from_the_spec_and_model( ) _, out_dir = run_export(tmp_path) - assert (out_dir / "provenance.mcf").read_text().strip() == ( - "Node: dcid:provenance/Provenance1\n" - 'name: "Provenance One"\n' - "typeOf: dcid:Provenance\n" - 'url: "https://example.org/source"\n' - "source: dcid:dc/s/Source1\n" - 'sourceDataUrl: "https://example.org/source/data.parquet"\n' - 'license: "https://example.org/terms"\n' - "licenseType: dcid:LicenseType1\n" - 'licenseAttribution: "Source One (2026), Provenance One ' - '(accessed on 2026-03-04)"\n' - 'lastDataRefreshDate: "2026-10-14"\n' - 'importedSourceReleaseDate: "2026-02-15"\n' - 'nextSourceReleaseDate: "2026-06"\n' - 'sourceReleaseFrequency: "P1Y"\n' - 'earliestObservationDate: "2000"\n' - 'latestObservationDate: "2001"\n' - 'dataTransformationDescription: "## Step one\\n\\nWhat the step does.\\n"\n' - 'dataTransformationChangelog: "## 2026-01-01\\n\\n### Added\\n\\n' - '- Dataset added to the graph.\\n"' - ) + assert mcf_properties((out_dir / "provenance.mcf").read_text()) == { + "Node": "dcid:provenance/Provenance1", + "name": '"Provenance One"', + "typeOf": "dcid:Provenance", + "url": '"https://example.org/source"', + "source": "dcid:dc/s/Source1", + "sourceDataUrl": '"https://example.org/source/data.parquet"', + "license": '"https://example.org/terms"', + "licenseType": "dcid:LicenseType1", + "licenseAttribution": '"Source One (2026), Provenance One (accessed on 2026-03-04)"', + "lastDataRefreshDate": '"2026-10-14"', + "importedSourceReleaseDate": '"2026-02-15"', + "nextSourceReleaseDate": '"2026-06"', + "sourceReleaseFrequency": '"P1Y"', + "earliestObservationDate": '"2000"', + "latestObservationDate": '"2001"', + "dataTransformationDescription": '"## Step one\\n\\nWhat the step does.\\n"', + "dataTransformationChangelog": '"## 2026-01-01\\n\\n### Added\\n\\n- Dataset added to the graph.\\n"', + } @pytest.mark.parametrize("missing", DOCUMENTATION_FILES.values()) diff --git a/tests/data_commons/test_ingestion.py b/tests/data_commons/test_ingestion.py index 9b26b73..73cbcd1 100644 --- a/tests/data_commons/test_ingestion.py +++ b/tests/data_commons/test_ingestion.py @@ -1,32 +1,59 @@ import logging +import warnings from collections.abc import Iterator from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, call, patch import pytest -from dcp_tools.gcp_utilities.settings import KGSettings +from dcp_tools.gcp_utilities.ingestion import ( + IngestionExecutionState, + WorkflowContractError, +) from upath import UPath -from dfo_pipelines.data_commons.ingestion import ingest_bundle +from dfo_pipelines.data_commons.ingestion import ( + Ingestion, + IngestionExecutionError, + ingest_bundle, + watch_execution_start, +) from dfo_pipelines.defs.resources import KnowledgeGraphResource +from tests.helpers import EXECUTION_NAME LOG = logging.getLogger(__name__) IMPORT_NAME = "test_import" -KNOWLEDGE_GRAPH = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", +SUCCEEDED = IngestionExecutionState(state="SUCCEEDED", current_steps=(), error=None) +ACTIVE = IngestionExecutionState( + state="ACTIVE", current_steps=("main.run_preprocessing",), error=None ) +class FakeTime: + """A clock that moves only when slept on""" + + def __init__(self) -> None: + self.now = 0.0 + + def clock(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.now += seconds + + @pytest.fixture def gcp() -> Iterator[MagicMock]: """dcp-tools' GCP calls, recorded in the order they are made""" calls = MagicMock() + calls.run_ingestion_workflow.return_value = EXECUTION_NAME + calls.get_ingestion_execution_state.return_value = SUCCEEDED with ( + patch( + "dfo_pipelines.data_commons.ingestion.check_ingestion_workflow_contract", + calls.check_ingestion_workflow_contract, + ), patch( "dfo_pipelines.data_commons.ingestion.get_gcs_client", calls.get_gcs_client, @@ -39,6 +66,10 @@ def gcp() -> Iterator[MagicMock]: "dfo_pipelines.data_commons.ingestion.run_ingestion_workflow", calls.run_ingestion_workflow, ), + patch( + "dfo_pipelines.data_commons.ingestion.get_ingestion_execution_state", + calls.get_ingestion_execution_state, + ), ): yield calls @@ -51,62 +82,179 @@ def write_bundle(directory: Path) -> Path: return directory -def ingest(bundle: Path, settings: KGSettings | None = None) -> str: +def ingest(bundle: Path, knowledge_graph: KnowledgeGraphResource) -> Ingestion: """Ingest a bundle into the test knowledge graph""" return ingest_bundle( UPath(bundle), import_name=IMPORT_NAME, - settings=settings or KNOWLEDGE_GRAPH.settings, + settings=knowledge_graph.settings, log=LOG, ) def test_ingest_syncs_the_bundle_into_the_import_folder_then_ingests_it( - tmp_path: Path, gcp: MagicMock + tmp_path: Path, gcp: MagicMock, knowledge_graph: KnowledgeGraphResource ) -> None: """Test the sync lands in the import's own folder before the workflow starts""" bundle = write_bundle(tmp_path / "bundle") - settings = KNOWLEDGE_GRAPH.settings + settings = knowledge_graph.settings folder = f"{settings.gcs_input_folder_path}/{IMPORT_NAME}" - destination = ingest(bundle, settings) + ingestion = ingest(bundle, knowledge_graph) + gcp.get_gcs_client.assert_called_once_with( + settings.gcp_credentials, project=settings.gcp_project_id + ) bucket = gcp.get_gcs_client.return_value.bucket bucket.assert_called_once_with(settings.gcs_bucket_name) - sync, trigger = [ - call - for call in gcp.mock_calls - if call[0] in ("sync_directory_to_gcs", "run_ingestion_workflow") + steps = [ + step + for step in gcp.mock_calls + if step[0] + in ( + "check_ingestion_workflow_contract", + "sync_directory_to_gcs", + "run_ingestion_workflow", + "get_ingestion_execution_state", + ) ] - assert sync == ( - "sync_directory_to_gcs", - (bucket.return_value, bundle), - {"gcs_folder_name": folder}, + assert steps == [ + call.check_ingestion_workflow_contract(settings), + call.sync_directory_to_gcs(bucket.return_value, bundle, gcs_folder_name=folder), + call.run_ingestion_workflow( + settings, imports=IMPORT_NAME, skip_workflow_contract_check=True + ), + call.get_ingestion_execution_state(settings, EXECUTION_NAME), + ] + assert ingestion == Ingestion( + destination=f"gs://{settings.gcs_bucket_name}/{folder}", + execution_name=EXECUTION_NAME, ) - assert trigger == ("run_ingestion_workflow", (settings,), {"imports": IMPORT_NAME}) - assert destination == f"gs://{settings.gcs_bucket_name}/{folder}" def test_ingest_starts_no_workflow_when_the_sync_fails( - tmp_path: Path, gcp: MagicMock + tmp_path: Path, gcp: MagicMock, knowledge_graph: KnowledgeGraphResource ) -> None: """Test a failed sync is not followed by an ingestion of the old bundle""" bundle = write_bundle(tmp_path / "bundle") gcp.sync_directory_to_gcs.side_effect = RuntimeError("upload failed") with pytest.raises(RuntimeError, match="upload failed"): - ingest(bundle) + ingest(bundle, knowledge_graph) gcp.run_ingestion_workflow.assert_not_called() +def test_ingest_uploads_nothing_to_a_workflow_dcp_tools_refuses( + tmp_path: Path, gcp: MagicMock, knowledge_graph: KnowledgeGraphResource +) -> None: + """Test a workflow from the wrong stack fails the ingestion before the sync""" + bundle = write_bundle(tmp_path / "bundle") + gcp.check_ingestion_workflow_contract.side_effect = WorkflowContractError( + "not DCP 1.1.6" + ) + + with pytest.raises(WorkflowContractError): + ingest(bundle, knowledge_graph) + + gcp.sync_directory_to_gcs.assert_not_called() + gcp.run_ingestion_workflow.assert_not_called() + + +@pytest.mark.parametrize("state", ["FAILED", "CANCELLED"]) +def test_ingest_fails_when_the_execution_ends_while_watched( + tmp_path: Path, + gcp: MagicMock, + knowledge_graph: KnowledgeGraphResource, + state: str, +) -> None: + """Test an execution that ends early fails the ingestion with its error""" + bundle = write_bundle(tmp_path / "bundle") + gcp.get_ingestion_execution_state.return_value = IngestionExecutionState( + state=state, current_steps=(), error="preprocessing job failed" + ) + + with pytest.raises(IngestionExecutionError, match="preprocessing job failed"): + ingest(bundle, knowledge_graph) + + +def test_watch_returns_a_succeeded_execution_without_waiting( + gcp: MagicMock, knowledge_graph: KnowledgeGraphResource +) -> None: + """Test a finished execution ends the watch at the first read""" + time = FakeTime() + + execution = watch_execution_start( + knowledge_graph.settings, + EXECUTION_NAME, + log=LOG, + watch_seconds=120, + poll_seconds=5, + sleep=time.sleep, + clock=time.clock, + ) + + assert execution == SUCCEEDED + assert time.now == 0 + + +def test_watch_leaves_an_active_execution_running_when_the_watch_ends( + gcp: MagicMock, knowledge_graph: KnowledgeGraphResource +) -> None: + """Test an execution still running at the deadline is returned, not failed""" + gcp.get_ingestion_execution_state.return_value = ACTIVE + time = FakeTime() + + execution = watch_execution_start( + knowledge_graph.settings, + EXECUTION_NAME, + log=LOG, + watch_seconds=12, + poll_seconds=5, + sleep=time.sleep, + clock=time.clock, + ) + + assert execution == ACTIVE + # Read at 0, 5, 10 and 15 seconds, the last past the 12-second deadline. + assert gcp.get_ingestion_execution_state.call_count == 4 + assert time.now == 15 + + +def test_watch_fails_an_execution_that_fails_after_starting( + gcp: MagicMock, knowledge_graph: KnowledgeGraphResource +) -> None: + """Test a failure inside the watch is caught after the execution was running""" + gcp.get_ingestion_execution_state.side_effect = [ + ACTIVE, + IngestionExecutionState(state="FAILED", current_steps=(), error="lock busy"), + ] + time = FakeTime() + + with pytest.raises(IngestionExecutionError, match="lock busy"): + watch_execution_start( + knowledge_graph.settings, + EXECUTION_NAME, + log=LOG, + watch_seconds=120, + poll_seconds=5, + sleep=time.sleep, + clock=time.clock, + ) + + assert time.now == 5 + + @pytest.mark.parametrize( "contents", [None, [], ["subdirectory/"], ["report.txt"]], ids=["missing", "empty", "only-a-subdirectory", "only-files-not-uploaded"], ) def test_ingest_refuses_a_bundle_with_no_files( - tmp_path: Path, gcp: MagicMock, contents: list[str] | None + tmp_path: Path, + gcp: MagicMock, + knowledge_graph: KnowledgeGraphResource, + contents: list[str] | None, ) -> None: """Test nothing reaches GCP when there is no bundle to ingest""" bundle = tmp_path / "bundle" @@ -119,19 +267,21 @@ def test_ingest_refuses_a_bundle_with_no_files( (bundle / name).write_text("") with pytest.raises(ValueError, match="Found no bundle"): - ingest(bundle) + ingest(bundle, knowledge_graph) gcp.sync_directory_to_gcs.assert_not_called() gcp.run_ingestion_workflow.assert_not_called() -def test_ingest_refuses_a_bundle_not_on_local_disk(gcp: MagicMock) -> None: +def test_ingest_refuses_a_bundle_not_on_local_disk( + gcp: MagicMock, knowledge_graph: KnowledgeGraphResource +) -> None: """Test a remote bundle fails rather than reaching dcp-tools' local upload""" with pytest.raises(ValueError, match="local disk"): ingest_bundle( UPath("memory://bundle"), import_name=IMPORT_NAME, - settings=KNOWLEDGE_GRAPH.settings, + settings=knowledge_graph.settings, log=LOG, ) @@ -139,17 +289,35 @@ def test_ingest_refuses_a_bundle_not_on_local_disk(gcp: MagicMock) -> None: def test_knowledge_graph_settings_ignore_the_environment( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + knowledge_graph: KnowledgeGraphResource, ) -> None: - """Test the settings build from the resource alone, with a .env file present + """Test the settings build from the resource alone - KGSettings would otherwise read .env and reject every variable it does not - declare. + KGSettings would otherwise let variables it declares override the resource, + warn about a deprecated one, and reject every variable in a .env file it does + not declare. """ (tmp_path / ".env").write_text("DUCKDB_PATH=elsewhere.duckdb\n") monkeypatch.chdir(tmp_path) + monkeypatch.setenv("INGESTION_PREP_JOB_NAME", "prep-job") + monkeypatch.setenv("GCS_INPUT_FOLDER_PATH", "elsewhere/input") + monkeypatch.setenv("GCP_CREDENTIALS", '{"type": "service_account"}') - settings = KNOWLEDGE_GRAPH.settings + with warnings.catch_warnings(): + warnings.simplefilter("error", FutureWarning) + settings = knowledge_graph.settings - assert settings.gcs_bucket_name == KNOWLEDGE_GRAPH.bucket_name - assert settings.ingestion_workflow_name == KNOWLEDGE_GRAPH.workflow_name + assert settings.model_dump() == { + "local_path": Path("."), + "gcp_project_id": knowledge_graph.project_id, + "gcp_region": knowledge_graph.region, + "gcp_credentials": None, + "gcs_bucket_name": knowledge_graph.bucket_name, + "gcs_input_folder_path": "ingestion/input", + "gcs_output_folder_path": "", + "ingestion_workflow_name": knowledge_graph.workflow_name, + "ingestion_service_account": None, + "ingestion_prep_job_name": None, + } diff --git a/tests/dfo/ontology/test_assets.py b/tests/dfo/ontology/test_assets.py index bc2bf29..24d57c7 100644 --- a/tests/dfo/ontology/test_assets.py +++ b/tests/dfo/ontology/test_assets.py @@ -11,6 +11,7 @@ from dagster_duckdb import DuckDBResource from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.dfo.ontology.assets import ( CODE_MAPPING_CSV, COMPILE_REPORT, @@ -26,7 +27,7 @@ KnowledgeGraphResource, UPathFileStoreResource, ) -from tests.helpers import column_list, meta, row_count +from tests.helpers import EXECUTION_NAME, column_list, meta, row_count FIXTURES = Path(__file__).parent / "fixtures" @@ -329,20 +330,14 @@ def test_the_compiled_ontology_reaches_the_table_and_the_bundle( def test_ingestion_ingests_the_bundle_the_export_wrote( - tmp_path: Path, compiler: Any + tmp_path: Path, compiler: Any, knowledge_graph: KnowledgeGraphResource ) -> None: """Test the ingestion is handed the export's directory under its import name""" - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" with patch( "dfo_pipelines.defs.dfo.ontology.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [ @@ -371,3 +366,4 @@ def test_ingestion_ingests_the_bundle_the_export_wrote( assert kwargs["import_name"] == IMPORT_NAME assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/dfo/shared/test_assets.py b/tests/dfo/shared/test_assets.py index 1741e17..9adaec9 100644 --- a/tests/dfo/shared/test_assets.py +++ b/tests/dfo/shared/test_assets.py @@ -4,6 +4,7 @@ import dagster as dg from dagster_duckdb import DuckDBResource +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.dfo.shared.assets import ( IMPORT_NAME, dfo_shared_dc_export, @@ -13,7 +14,7 @@ KnowledgeGraphResource, UPathFileStoreResource, ) -from tests.helpers import meta +from tests.helpers import EXECUTION_NAME, meta LICENSE_TYPES_SQL = """ create or replace table dfo_shared.license_types as @@ -59,19 +60,15 @@ def test_dc_export_writes_a_node_for_each_license_type(tmp_path: Path) -> None: assert meta(result, "license_type_count", dg.IntMetadataValue).value == 1 -def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" with patch( "dfo_pipelines.defs.dfo.shared.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [dfo_shared_dc_export, dfo_shared_dc_ingestion], @@ -92,3 +89,4 @@ def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: assert kwargs["import_name"] == IMPORT_NAME assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/dfo/skills/test_assets.py b/tests/dfo/skills/test_assets.py index e42bf53..310c490 100644 --- a/tests/dfo/skills/test_assets.py +++ b/tests/dfo/skills/test_assets.py @@ -13,6 +13,7 @@ from upath import UPath from dfo_pipelines.data_commons.export import ExportSpec +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.definitions import defs from dfo_pipelines.defs.dfo.ontology.assets import ONTOLOGY_MCF from dfo_pipelines.defs.dfo.skills.assets import ( @@ -31,7 +32,7 @@ SkillsRepositoryResource, UPathFileStoreResource, ) -from tests.helpers import AutomationHarness, meta +from tests.helpers import EXECUTION_NAME, AutomationHarness, meta FIXTURES = Path(__file__).parent / "fixtures" @@ -352,19 +353,15 @@ def _stub_skills_raw_directory() -> str: return str(FIXTURES) -def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) destination = f"gs://test-bucket/ingestion/input/{IMPORT_NAME}" with patch( "dfo_pipelines.defs.dfo.skills.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [ @@ -392,6 +389,7 @@ def test_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: assert kwargs["import_name"] == IMPORT_NAME assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME def write_skill( diff --git a/tests/helpers.py b/tests/helpers.py index 10a118f..68ce8a4 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -10,6 +10,12 @@ from dfo_pipelines.defs.resources import UPathFileStoreResource +# Resource name of the workflow execution a mocked ingestion reports starting. +EXECUTION_NAME = ( + "projects/test-project/locations/test-region/workflows/test-workflow/" + "executions/test-execution" +) + def meta[T](result: dg.MaterializeResult, key: str, kind: type[T]) -> T: assert result.metadata is not None diff --git a/tests/imf/weo/test_assets.py b/tests/imf/weo/test_assets.py index bb116a1..a7ae8e1 100644 --- a/tests/imf/weo/test_assets.py +++ b/tests/imf/weo/test_assets.py @@ -13,6 +13,7 @@ from imf_reader.weo.api import FlowRef from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.imf.weo.assets import ( IMF_WEO_EXPORT_SPEC, PUBLICATION_DATE_URL, @@ -33,6 +34,7 @@ write_extraction_metadata, ) from tests.helpers import ( + EXECUTION_NAME, AutomationHarness, column_list, list_first_row, @@ -635,23 +637,19 @@ def test_imf_weo_dc_export_writes_each_property_the_statvar_sets( ) -def test_imf_weo_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_imf_weo_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) with duckdb.get_connection() as connection: connection.execute(DC_MODELS_SQL) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = IMF_WEO_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.imf.weo.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [imf_weo_dc_export, imf_weo_dc_ingestion], @@ -674,3 +672,4 @@ def test_imf_weo_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) - assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/oecd/dac1/test_assets.py b/tests/oecd/dac1/test_assets.py index c03f0ae..1e4a0e3 100644 --- a/tests/oecd/dac1/test_assets.py +++ b/tests/oecd/dac1/test_assets.py @@ -13,6 +13,7 @@ from oda_reader.download.download_tools import DAC1_FLOW_URL from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.oecd.dac1.assets import ( OECD_DAC1_EXPORT_SPEC, oecd_dac1_dc_export, @@ -32,6 +33,7 @@ write_extraction_metadata, ) from tests.helpers import ( + EXECUTION_NAME, AutomationHarness, column_list, list_parquet_columns, @@ -684,23 +686,18 @@ def test_oecd_dac1_dc_export_writes_each_property_the_statvar_sets( def test_oecd_dac1_ingestion_ingests_the_bundle_the_export_wrote( tmp_path: Path, + knowledge_graph: KnowledgeGraphResource, ) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) with duckdb.get_connection() as connection: connection.execute(DC_MODELS_SQL) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = OECD_DAC1_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.oecd.dac1.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [oecd_dac1_dc_export, oecd_dac1_dc_ingestion], @@ -723,3 +720,4 @@ def test_oecd_dac1_ingestion_ingests_the_bundle_the_export_wrote( assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/oecd/dac2a/test_assets.py b/tests/oecd/dac2a/test_assets.py index 84440a0..596409d 100644 --- a/tests/oecd/dac2a/test_assets.py +++ b/tests/oecd/dac2a/test_assets.py @@ -13,6 +13,7 @@ from oda_reader.download.download_tools import DAC2A_FLOW_URL from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.oecd.dac2a.assets import ( OECD_DAC2A_EXPORT_SPEC, oecd_dac2a_dc_export, @@ -32,6 +33,7 @@ write_extraction_metadata, ) from tests.helpers import ( + EXECUTION_NAME, AutomationHarness, column_list, list_parquet_columns, @@ -665,23 +667,18 @@ def test_oecd_dac2a_dc_export_writes_each_property_the_statvar_sets( def test_oecd_dac2a_ingestion_ingests_the_bundle_the_export_wrote( tmp_path: Path, + knowledge_graph: KnowledgeGraphResource, ) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) with duckdb.get_connection() as connection: connection.execute(DC_MODELS_SQL) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = OECD_DAC2A_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.oecd.dac2a.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [oecd_dac2a_dc_export, oecd_dac2a_dc_ingestion], @@ -704,3 +701,4 @@ def test_oecd_dac2a_ingestion_ingests_the_bundle_the_export_wrote( assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/oecd/dac2b/test_assets.py b/tests/oecd/dac2b/test_assets.py index 0c07e30..4d6f1cc 100644 --- a/tests/oecd/dac2b/test_assets.py +++ b/tests/oecd/dac2b/test_assets.py @@ -14,6 +14,7 @@ from oda_reader.download.download_tools import DAC2B_FLOW_URL from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.oecd.dac2b.assets import ( OECD_DAC2B_EXPORT_SPEC, oecd_dac2b_dc_export, @@ -33,6 +34,7 @@ write_extraction_metadata, ) from tests.helpers import ( + EXECUTION_NAME, AutomationHarness, column_list, list_parquet_columns, @@ -657,23 +659,18 @@ def test_oecd_dac2b_dc_export_writes_each_property_the_statvar_sets( def test_oecd_dac2b_ingestion_ingests_the_bundle_the_export_wrote( tmp_path: Path, + knowledge_graph: KnowledgeGraphResource, ) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) with duckdb.get_connection() as connection: connection.execute(DC_MODELS_SQL) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = OECD_DAC2B_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.oecd.dac2b.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [oecd_dac2b_dc_export, oecd_dac2b_dc_ingestion], @@ -696,3 +693,4 @@ def test_oecd_dac2b_ingestion_ingests_the_bundle_the_export_wrote( assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/oecd/dac_crs/test_assets.py b/tests/oecd/dac_crs/test_assets.py index 9224de3..b984959 100644 --- a/tests/oecd/dac_crs/test_assets.py +++ b/tests/oecd/dac_crs/test_assets.py @@ -13,6 +13,7 @@ from oda_reader.download.download_tools import CRS_FLOW_URL from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.oecd.dac_crs.assets import ( CRS_EXPORT_SPEC, oecd_dac_crs_dc_export, @@ -32,7 +33,7 @@ read_extraction_metadata, write_extraction_metadata, ) -from tests.helpers import AutomationHarness, row_count +from tests.helpers import EXECUTION_NAME, AutomationHarness, row_count FIXTURES = Path(__file__).parent / "fixtures" @@ -687,22 +688,18 @@ def test_dac_crs_dc_export_maps_the_crs_dimensions_to_ontology_properties( ) -def test_dac_crs_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_dac_crs_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) create_dc_models(duckdb) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = CRS_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.oecd.dac_crs.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [oecd_dac_crs_dc_export, oecd_dac_crs_dc_ingestion], @@ -725,3 +722,4 @@ def test_dac_crs_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) - assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME diff --git a/tests/one/conversion_tables/test_assets.py b/tests/one/conversion_tables/test_assets.py index 0d4203d..948eaef 100644 --- a/tests/one/conversion_tables/test_assets.py +++ b/tests/one/conversion_tables/test_assets.py @@ -10,6 +10,7 @@ from dagster_duckdb import DuckDBResource from dfo_pipelines.data_commons.export import DOCUMENTATION_FILES, ExportSpec +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.one.conversion_tables.assets import ( DAC_EXPORT_SPEC, IMF_EXPORT_SPEC, @@ -22,6 +23,7 @@ KnowledgeGraphResource, UPathFileStoreResource, ) +from tests.helpers import EXECUTION_NAME DOCS_DIR = Path(__file__).parents[3] / "docs" / "one" / "conversion_tables" @@ -281,7 +283,7 @@ def test_dc_export_marks_only_projected_rows_with_a_measurement_method( @pytest.mark.parametrize("table", TABLES) def test_dc_ingestion_ingests_the_bundle_the_export_wrote( - tmp_path: Path, table: str + tmp_path: Path, table: str, knowledge_graph: KnowledgeGraphResource ) -> None: """Test each ingestion is handed its own export's directory under its import name""" spec, export_asset = TABLES[table] @@ -291,17 +293,11 @@ def test_dc_ingestion_ingests_the_bundle_the_export_wrote( connection.execute(_provenance_sql(table)) connection.execute(_statvars_sql(table)) connection.execute(_observations_sql(table)) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) destination = f"gs://test-bucket/ingestion/input/{spec.import_name}" with patch( "dfo_pipelines.defs.one.conversion_tables.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [export_asset, INGESTIONS[table]], @@ -325,4 +321,5 @@ def test_dc_ingestion_ingests_the_bundle_the_export_wrote( assert ingestion.metadata == { "dagster/uri": dg.TextMetadataValue(destination), "import_name": dg.TextMetadataValue(spec.import_name), + "execution_name": dg.TextMetadataValue(EXECUTION_NAME), } diff --git a/tests/one/imputed_multilateral_sectors/test_assets.py b/tests/one/imputed_multilateral_sectors/test_assets.py index 384dc0b..ca90634 100644 --- a/tests/one/imputed_multilateral_sectors/test_assets.py +++ b/tests/one/imputed_multilateral_sectors/test_assets.py @@ -13,6 +13,7 @@ from upath import UPath from dfo_pipelines.data_commons.export import DOCUMENTATION_FILES +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.one.imputed_multilateral_sectors.assets import ( CRS_BULK_LABEL, CURRENCY, @@ -38,7 +39,7 @@ read_extraction_metadata, write_extraction_metadata, ) -from tests.helpers import column_list, list_first_row, meta, row_count +from tests.helpers import EXECUTION_NAME, column_list, list_first_row, meta, row_count ASSETS_MODULE = "dfo_pipelines.defs.one.imputed_multilateral_sectors.assets" @@ -1337,20 +1338,19 @@ def test_dc_export_writes_a_bundle_to_a_date_partitioned_path(tmp_path: Path) -> } -def test_dc_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_dc_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) _create_dc_models(duckdb) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = IMPUTED_MULTILATERAL_SECTORS_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" - with patch(f"{ASSETS_MODULE}.ingest_bundle", return_value=destination) as ingest: + with patch( + f"{ASSETS_MODULE}.ingest_bundle", + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), + ) as ingest: result = dg.materialize( [ one_imputed_multilateral_sectors_dc_export, @@ -1378,6 +1378,7 @@ def test_dc_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> Non assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME def test_dc_export_writes_the_pipeline_documentation_onto_the_provenance_node( diff --git a/tests/world_bank/ids/test_assets.py b/tests/world_bank/ids/test_assets.py index 8f22b3b..56b9ffb 100644 --- a/tests/world_bank/ids/test_assets.py +++ b/tests/world_bank/ids/test_assets.py @@ -11,6 +11,7 @@ from dagster_duckdb import DuckDBResource from upath import UPath +from dfo_pipelines.data_commons.ingestion import Ingestion from dfo_pipelines.defs.resources import ( KnowledgeGraphResource, UPathFileStoreResource, @@ -35,6 +36,7 @@ write_extraction_metadata, ) from tests.helpers import ( + EXECUTION_NAME, AutomationHarness, column_list, list_first_row, @@ -1129,22 +1131,18 @@ def test_ids_dc_export_maps_the_ids_dimensions_to_ontology_properties( ) -def test_ids_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> None: +def test_ids_ingestion_ingests_the_bundle_the_export_wrote( + tmp_path: Path, knowledge_graph: KnowledgeGraphResource +) -> None: """Test the ingestion is handed the export's directory under its import name""" duckdb = DuckDBResource(database=str(tmp_path / "database.duckdb")) create_dc_models(duckdb) - knowledge_graph = KnowledgeGraphResource( - project_id="test-project", - region="test-region", - bucket_name="test-bucket", - workflow_name="test-workflow", - ) import_name = IDS_EXPORT_SPEC.import_name destination = f"gs://test-bucket/ingestion/input/{import_name}" with patch( "dfo_pipelines.defs.world_bank.ids.assets.ingest_bundle", - return_value=destination, + return_value=Ingestion(destination=destination, execution_name=EXECUTION_NAME), ) as ingest_bundle: result = dg.materialize( [world_bank_ids_dc_export, world_bank_ids_dc_ingestion], @@ -1167,3 +1165,4 @@ def test_ids_ingestion_ingests_the_bundle_the_export_wrote(tmp_path: Path) -> No assert kwargs["import_name"] == import_name assert kwargs["settings"] == knowledge_graph.settings assert ingestion.metadata["dagster/uri"].value == destination + assert ingestion.metadata["execution_name"].value == EXECUTION_NAME From fe3a36412505d6d56aac87f7d76f6186f6a98145 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Fri, 23 Oct 2026 19:40:57 +0200 Subject: [PATCH 3/3] build: require dcp-tools 1.1.0 dcp-tools 1.1.0 starts ingestion on Data Commons Platform 1.1.6 and pulls in datacommons-admin 1.1.6, which caps setuptools at 80.0.0. resources.py imports pydantic_settings directly, so declare it. --- pyproject.toml | 3 +- uv.lock | 236 ++++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 228 insertions(+), 11 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index bb8b05e..c7dc160 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,8 @@ dependencies = [ "universal-pathlib>=0.3.10", "pandas>=3.0.5", "imf-reader>=2.2.0", - "dcp-tools>=1.0.1", + "dcp-tools>=1.1.0", + "pydantic-settings>=2.9.1,<3.0.0", "dfo-ontology", "oda-data==2.8.1", "pyyaml>=6.0", diff --git a/uv.lock b/uv.lock index 5358bef..f0eeaf4 100644 --- a/uv.lock +++ b/uv.lock @@ -576,16 +576,33 @@ wheels = [ [[package]] name = "datacommons-admin" -version = "1.1.2" +version = "1.1.6" source = { registry = "https://pypi.org/simple" } dependencies = [ + { name = "certifi" }, { name = "click" }, + { name = "datacommons-db" }, { name = "google-cloud-storage" }, { name = "pyopenssl" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/5d/2b/a98fadca468bb02befc134d04af7266217fea5ae6da4b9672ec49bf8d249/datacommons_admin-1.1.2.tar.gz", hash = "sha256:854e0a84662e75c40a6430704c3f3bd98a9ace4bbf3051940be74bd2b6268185", size = 16197, upload-time = "2026-08-12T00:02:11.359Z" } +sdist = { url = "https://files.pythonhosted.org/packages/25/d1/668cbd9b9af99f67f357bbf2617235c8b2de9907d0b3ba2f214f73bb2a7b/datacommons_admin-1.1.6.tar.gz", hash = "sha256:9f1452042ddad68f8d46fc70806c16218254a1d94dcfb38f2feedacb8ab319d0", size = 19228, upload-time = "2026-10-02T07:38:10.194Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2a/5e/7469bcbedba0b6cb4e5d6eea700ca872a12db2063ebd1c09607d0a4c244e/datacommons_admin-1.1.6-py3-none-any.whl", hash = "sha256:cbea62f1901e725d97ec60cc451d5d27b72fb3c166b468ce171ab8ea644fa014", size = 32154, upload-time = "2026-10-02T07:38:09.234Z" }, +] + +[[package]] +name = "datacommons-db" +version = "1.1.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-cloud-spanner" }, + { name = "setuptools" }, + { name = "sqlalchemy" }, + { name = "sqlalchemy-spanner" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d8/c9/3ba8e833967ac336ec084df22244af889ff5622d2eac90fd9170e6e3a4e9/datacommons_db-1.1.6.tar.gz", hash = "sha256:565c95fe99686f2cca3c8a40abd2959fe0ae3faabb7db3b88dfd0f640d1fddc1", size = 26619, upload-time = "2026-10-02T07:38:08.241Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/de/a9/601ce5ac8ce6ea4c29388d6df8be249b5d921d627f1690c28cdfc03d40e9/datacommons_admin-1.1.2-py3-none-any.whl", hash = "sha256:25f6acb195a7476437d21324bece9f20c7149b7d49b03e360b9ad9cec99eeccc", size = 17594, upload-time = "2026-08-12T00:02:10.29Z" }, + { url = "https://files.pythonhosted.org/packages/6f/43/45210e7c25d9416fefa85673492c47b5081c84c6a2b2c0d2e6f4363f310d/datacommons_db-1.1.6-py3-none-any.whl", hash = "sha256:db715184f854501c474780a2191e4a5a1f77af68357f24cdaf9778f14749e436", size = 34622, upload-time = "2026-10-02T07:38:07.197Z" }, ] [[package]] @@ -735,18 +752,19 @@ wheels = [ [[package]] name = "dcp-tools" -version = "1.0.1" +version = "1.1.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "datacommons-admin" }, + { name = "google-auth", extra = ["requests"] }, { name = "google-cloud-storage" }, { name = "pandas" }, { name = "pydantic" }, { name = "pydantic-settings" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/37/60/15df64896af5efa8adba07ec9be4df6d35a181c8aaf507c4290eea70fada/dcp_tools-1.0.1.tar.gz", hash = "sha256:d73eb5c93095b47f601a6ffd92f8dd94b2ffe090797eef5e96415f61141c1244", size = 46732, upload-time = "2026-10-13T15:28:51.69Z" } +sdist = { url = "https://files.pythonhosted.org/packages/25/64/9d85d2c297a2d0c18a8d7ca2cb209f70c052c56e725fee82995567e36d32/dcp_tools-1.1.0.tar.gz", hash = "sha256:782f6233d6b9d1e7a090846f654f83286500cf5989ec5023f3860a0d4864b241", size = 49678, upload-time = "2026-10-23T17:37:59.309Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b4/24/d17e45df5db6b176ece631d5351c3bfe04015eba3a14c745b83dff429b97/dcp_tools-1.0.1-py3-none-any.whl", hash = "sha256:1686bbfbf0edef4db88d73e4a2e5d3f9d11a28c6f37bf4307c0c7a7c6e3e96d3", size = 56802, upload-time = "2026-10-13T15:28:50.709Z" }, + { url = "https://files.pythonhosted.org/packages/5b/2a/0e3e580ef6b775f640646a2d4ab45943ccef0ef906d141681528540bff5c/dcp_tools-1.1.0-py3-none-any.whl", hash = "sha256:03aee7bcc5e5191be55c97949ee5250342fbba3539dd69f8be4593d0cffede0f", size = 60102, upload-time = "2026-10-23T17:37:58.13Z" }, ] [[package]] @@ -796,6 +814,7 @@ dependencies = [ { name = "openpyxl" }, { name = "pandas" }, { name = "pyarrow" }, + { name = "pydantic-settings" }, { name = "pyyaml" }, { name = "universal-pathlib" }, ] @@ -822,7 +841,7 @@ requires-dist = [ { name = "dagster-dbt", specifier = ">=0.29.10" }, { name = "dagster-duckdb", specifier = ">=0.29.10" }, { name = "dbt-duckdb", specifier = ">=1.10.1" }, - { name = "dcp-tools", specifier = ">=1.0.1" }, + { name = "dcp-tools", specifier = ">=1.1.0" }, { name = "dfo-ontology", git = "https://github.com/DevelopmentFinanceObservatory/ontology?rev=617bdb685e79022494132ef3614674c232b395fc" }, { name = "httpx", specifier = ">=0.28.0" }, { name = "imf-reader", specifier = ">=2.2.0" }, @@ -831,6 +850,7 @@ requires-dist = [ { name = "openpyxl", specifier = ">=3.1.0" }, { name = "pandas", specifier = ">=3.0.5" }, { name = "pyarrow", specifier = ">=23.0.0" }, + { name = "pydantic-settings", specifier = ">=2.9.1,<3.0.0" }, { name = "pyyaml", specifier = ">=6.0" }, { name = "universal-pathlib", specifier = ">=0.3.10" }, ] @@ -1015,6 +1035,12 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bc/c1/a8a92ae1bc4b1a8f804c776d7d3f0c771b78a62c3ad4df1be41b3fd8c767/google_api_core-2.34.0-py3-none-any.whl", hash = "sha256:cdf9c67e7ca2402d86ccbfde5f2503fc83e3cc3f58cc78456ae96cad24a6d2de", size = 180545, upload-time = "2026-08-06T06:22:47.502Z" }, ] +[package.optional-dependencies] +grpc = [ + { name = "grpcio" }, + { name = "grpcio-status" }, +] + [[package]] name = "google-auth" version = "2.56.3" @@ -1028,6 +1054,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bc/b3/6117b2f24065cd7e2c4f140e9a193e215f089ca8ba314cf91eb9d0b7fe0a/google_auth-2.56.3-py3-none-any.whl", hash = "sha256:8ec438808f813ad034535000261eed1067475d229d05bbf4216e78c3f2362e53", size = 259116, upload-time = "2026-08-06T06:22:51.788Z" }, ] +[package.optional-dependencies] +requests = [ + { name = "requests" }, +] + [[package]] name = "google-cloud-core" version = "2.6.1" @@ -1041,6 +1072,48 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5d/4f/37960d5255988b218c40c9b5a2b731013684294a50735d1dd1ab4894e531/google_cloud_core-2.6.1-py3-none-any.whl", hash = "sha256:2682a8a4474a32f56292fb4bca7fa7e4fb0b4af958f6abfe4bca8d195747fd45", size = 29393, upload-time = "2026-08-06T06:23:11.405Z" }, ] +[[package]] +name = "google-cloud-monitoring" +version = "2.32.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core", extra = ["grpc"] }, + { name = "google-auth" }, + { name = "grpcio" }, + { name = "proto-plus" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c8/94/68b1d25f99f8e61c8d4d4403caab3746c9e4aa629e1a8aef22f7f099c95c/google_cloud_monitoring-2.32.0.tar.gz", hash = "sha256:507501ce1c39161c071754f7ed14cd527788b4ea44a1d052bd815c115a70ea48", size = 388895, upload-time = "2026-10-01T18:17:17.249Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/89/97/86870c4f046bba0dcd5ca2099f30413653b84d0fbef611e5f39f8a94e3a8/google_cloud_monitoring-2.32.0-py3-none-any.whl", hash = "sha256:b6c026a5780330aa26c6151279194daa241369b324c96fd6c52c703137bc6b5b", size = 384477, upload-time = "2026-10-01T18:11:09.382Z" }, +] + +[[package]] +name = "google-cloud-spanner" +version = "3.71.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core", extra = ["grpc"] }, + { name = "google-auth" }, + { name = "google-cloud-core" }, + { name = "google-cloud-monitoring" }, + { name = "grpc-google-iam-v1" }, + { name = "grpc-interceptor" }, + { name = "grpcio" }, + { name = "mmh3" }, + { name = "opentelemetry-api" }, + { name = "opentelemetry-resourcedetector-gcp" }, + { name = "opentelemetry-sdk" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "proto-plus" }, + { name = "protobuf" }, + { name = "sqlparse" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/93/ed/b02289c6b4035a1dd04c6377883b5e5310e79e3eb4ce9a687aaea78da23d/google_cloud_spanner-3.71.0.tar.gz", hash = "sha256:976280a8338cd10dcbe5673788eb4b54a55c4f271b31d251c18ab4d9e1c001d3", size = 921825, upload-time = "2026-09-03T22:31:16.358Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/9a/f15b3a0fa188a550b0a7c2dde7c7ce88d792ef813c74042eec55dea036b1/google_cloud_spanner-3.71.0-py3-none-any.whl", hash = "sha256:658443c774c3967a3a5773f40f42aebf7b93cf1b63d5a8cb556926ff9fc53f9d", size = 626703, upload-time = "2026-09-03T22:30:37.359Z" }, +] + [[package]] name = "google-cloud-storage" version = "3.13.1" @@ -1095,6 +1168,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9a/51/186c02b8549b69ccda44429cf6ff5081e4b61a602ddfe6a8020d1be31d1b/googleapis_common_protos-1.75.1-py3-none-any.whl", hash = "sha256:28a1934bcd33b9c9da66ac301a0a4227e3367f095a17d0375cb98f0a09d93b79", size = 300626, upload-time = "2026-08-06T06:23:46.696Z" }, ] +[package.optional-dependencies] +grpc = [ + { name = "grpcio" }, +] + [[package]] name = "gql" version = "4.0.0" @@ -1168,6 +1246,32 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3e/e6/9160210222386b1a378ff94db846b9508ca24a121cf684991561fdb69280/greenlet-3.5.4-cp313-cp313-win_arm64.whl", hash = "sha256:7c1303791d603080cac6fc3b34df51c3b75b723739c282c8029e48a0d241672f", size = 245500, upload-time = "2026-07-22T11:40:22.185Z" }, ] +[[package]] +name = "grpc-google-iam-v1" +version = "0.14.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "googleapis-common-protos", extra = ["grpc"] }, + { name = "grpcio" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/97/24/0ef24ccdd91ce07be7d6466d39becdd4fbb33e52426ba2391fe334d24bca/grpc_google_iam_v1-0.14.6.tar.gz", hash = "sha256:fc0a683ace0a149184fcd08d667dd64d51ae65e5698e86f7136aa6002a8b5373", size = 24088, upload-time = "2026-10-22T23:43:37.377Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/61/e0/0524fbc82835c22e711ac3e7d7dfb3ee997f99f5eaa22ba6b31cc17b753d/grpc_google_iam_v1-0.14.6-py3-none-any.whl", hash = "sha256:042705a3ce306f289b9c6eafc3453b23f953bab955adcac3c36545758ccd681a", size = 33089, upload-time = "2026-10-22T23:43:12.096Z" }, +] + +[[package]] +name = "grpc-interceptor" +version = "0.15.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "grpcio" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/9f/28/57449d5567adf4c1d3e216aaca545913fbc21a915f2da6790d6734aac76e/grpc-interceptor-0.15.4.tar.gz", hash = "sha256:1f45c0bcb58b6f332f37c637632247c9b02bc6af0fdceb7ba7ce8d2ebbfb0926", size = 19322, upload-time = "2023-11-16T02:05:42.459Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/15/ac/8d53f230a7443401ce81791ec50a3b0e54924bf615ad287654fa4a2f5cdc/grpc_interceptor-0.15.4-py3-none-any.whl", hash = "sha256:0035f33228693ed3767ee49d937bac424318db173fef4d2d0170b3215f254d9d", size = 20848, upload-time = "2023-11-16T02:05:40.913Z" }, +] + [[package]] name = "grpcio" version = "1.83.0" @@ -1202,6 +1306,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6e/87/174bbfb5613794862c45b5cdf567fcfb478ad5a3a7b594e11d998656b2f9/grpcio_health_checking-1.81.1-py3-none-any.whl", hash = "sha256:cbc6a4171825ec64389de2f062d296ba129a5c27eefd0dd55fa837909184bdf9", size = 19120, upload-time = "2026-06-11T12:58:38.008Z" }, ] +[[package]] +name = "grpcio-status" +version = "1.83.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "googleapis-common-protos" }, + { name = "grpcio" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/52/fd/848dd7e009de85f8ca59999d1cc618ff8ebf7ea5636d083a47455d212d24/grpcio_status-1.83.0.tar.gz", hash = "sha256:837219c6de9afdccb6f6f72b34bc71e151a2011ef04040e3faaca746a57e54ae", size = 13965, upload-time = "2026-07-23T15:24:26.98Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d6/00/73204406228cf989bea6b0fd9fe4702fab49a8a152a0c6f90856dadb6ac7/grpcio_status-1.83.0-py3-none-any.whl", hash = "sha256:f6a838a7c5fb84ae98833ec0ef81ed438c26e11e54b2ddb8e92ad328c861de69", size = 14636, upload-time = "2026-07-23T15:23:49.044Z" }, +] + [[package]] name = "h11" version = "0.16.0" @@ -1518,6 +1636,37 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, ] +[[package]] +name = "mmh3" +version = "5.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8d/3c/eb1d82a87c504259dac5ce1c7de7587b68ffac841b55d23f8ea2c9df8422/mmh3-5.3.1.tar.gz", hash = "sha256:bd86d0c86b52332319d981d03781ff77811a29db544a69902dc06b5506bb3e19", size = 33354, upload-time = "2026-09-30T17:38:09.577Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e4/4c/c6faef1d29aa00a1f71d3a109547c86b029835b55ad19482dc625c98011c/mmh3-5.3.1-cp313-cp313-android_24_arm64_v8a.whl", hash = "sha256:4b2b6d135aafc93a666056ae87cf11dce93e11a3ee9b938d46076d93074699bf", size = 40911, upload-time = "2026-09-30T17:35:44.698Z" }, + { url = "https://files.pythonhosted.org/packages/7b/23/a35e5090c3685c3bd22f07586c4efa428710ad1404d6cf3fd47ad654e711/mmh3-5.3.1-cp313-cp313-android_24_x86_64.whl", hash = "sha256:98c6373ec81d4e74305d8d13d5de3aacf0e53e78dcb4a43dd74f6f3ff8452967", size = 42024, upload-time = "2026-09-30T17:35:45.814Z" }, + { url = "https://files.pythonhosted.org/packages/b5/59/350d214e1a37e5c2c92182750c06c671348346d52eb455bceaa861801349/mmh3-5.3.1-cp313-cp313-ios_13_0_arm64_iphoneos.whl", hash = "sha256:bf65874ed7c948281719632b6960f4eb572aa33e1a093a6a1d31bf064b0e540d", size = 38668, upload-time = "2026-09-30T17:35:46.993Z" }, + { url = "https://files.pythonhosted.org/packages/3b/c4/0a3d4e54549fd8edd6fa54cef0529dea7316666066f3bfa23c810d0c2b5e/mmh3-5.3.1-cp313-cp313-ios_13_0_arm64_iphonesimulator.whl", hash = "sha256:d1f3f35b97adfcf4545a4def9e0fb17e61eed8a06c29137a02829a67232e1588", size = 39344, upload-time = "2026-09-30T17:35:48.097Z" }, + { url = "https://files.pythonhosted.org/packages/80/b8/e96e8da1b8d52f62c15a8acb33cfd180778c18d71ed63e30a2085e35cf9c/mmh3-5.3.1-cp313-cp313-ios_13_0_x86_64_iphonesimulator.whl", hash = "sha256:39bbc0665b064e63a0e64e9efab9a97a1f0535b0e1ffd8e43e23aef82e41ca21", size = 39852, upload-time = "2026-09-30T17:35:49.227Z" }, + { url = "https://files.pythonhosted.org/packages/3e/28/c657ba46881ba84b2e1d260545c141b0794bb579c981fd71a0f7e5c15a73/mmh3-5.3.1-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:5cc32468caf0071882c3682b9ab04f45d756231059b4e36cccc94eb972f8c192", size = 55334, upload-time = "2026-09-30T17:35:50.704Z" }, + { url = "https://files.pythonhosted.org/packages/12/5b/cbff42a3248d0869a940eebef0eefe7feb948f6eeed2f242f0098e2892a0/mmh3-5.3.1-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8cb9941e2613b22ed4901faa29338c134194b2dec501023e6433b7e62161e329", size = 39946, upload-time = "2026-09-30T17:35:51.944Z" }, + { url = "https://files.pythonhosted.org/packages/5a/0a/67d5082ad1fe184c4c928ca6be61d775947590863d52de0a9d9aa7d525b2/mmh3-5.3.1-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:c25a6d4b6ff31d801ff6f1ad5ce003271bceabf21c3e9ffcf04a47774354e956", size = 39561, upload-time = "2026-09-30T17:35:53.331Z" }, + { url = "https://files.pythonhosted.org/packages/de/2c/948789af3824e81621f01183c1a2a017229bd7884e243641628682ea9ea5/mmh3-5.3.1-cp313-cp313-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:9ba38fef5eeed0668a27f8b5a002a5e30c789dd11fa058495b307f76226a4662", size = 97120, upload-time = "2026-09-30T17:35:54.528Z" }, + { url = "https://files.pythonhosted.org/packages/7f/46/88420e1561f1a5cda72e23581b9cbbde336cc4a4a8deb1259559bf57b8d2/mmh3-5.3.1-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:931d9d86c66f306e91414e95509af05e5c79bfcbac78218c2ee55c9734000053", size = 103129, upload-time = "2026-09-30T17:35:55.78Z" }, + { url = "https://files.pythonhosted.org/packages/bb/97/064d5c9eed7afe9b2087c164ab4b11a9d4cd0cb1d8d826804df72d7e0e17/mmh3-5.3.1-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ae367d0cf6cb40f3ec60ebdb572022f3cc875bcf4c661d345f3dbf24571e7aa3", size = 106044, upload-time = "2026-09-30T17:35:57.277Z" }, + { url = "https://files.pythonhosted.org/packages/39/b4/c4e968be21d55aead9ef78b6ac6fc0e4a455cfbbaff9ef62bcdaee40b26a/mmh3-5.3.1-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:59dab80b8124998406c168ddc9d6cbcede1c117dd0aed3db80e16e43ad71ef82", size = 112890, upload-time = "2026-09-30T17:35:58.596Z" }, + { url = "https://files.pythonhosted.org/packages/17/e9/b3f3da18b38bd08d39eb24c142ba3e9217b8975d3ca8468143dd3c63aafc/mmh3-5.3.1-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:803ba415427118e00ffccefbacc41b03df8ac60403cd9cf2dfd461ef56072002", size = 120520, upload-time = "2026-09-30T17:35:59.924Z" }, + { url = "https://files.pythonhosted.org/packages/fe/03/c7dc6eb152425dd2fba09b299a186be37bd53910a28531ab12f475d9bf99/mmh3-5.3.1-cp313-cp313-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:5d856a44ef94204820338b0e3312c02a6a4df8040ad06102c050d5005dbc601c", size = 103089, upload-time = "2026-09-30T17:36:01.266Z" }, + { url = "https://files.pythonhosted.org/packages/25/c5/1192cf2db35390b0ca1f54eae2699c62235fce57992eda605e675af06b9f/mmh3-5.3.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:bf1fa41b7587477c86ffe4e69b854ef688e243f9feb97eebc08666031bde71e0", size = 98905, upload-time = "2026-09-30T17:36:02.545Z" }, + { url = "https://files.pythonhosted.org/packages/18/3a/9af0d1f08e3e03cd8b52e5d53fd3be74345993c0d6a7cf61b02e454c4daa/mmh3-5.3.1-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:86c1593ebec4bd8a7b1e0f28fce5f220e5bc0b2d5f9ba48c34224c04d9f63b8f", size = 98457, upload-time = "2026-09-30T17:36:04.006Z" }, + { url = "https://files.pythonhosted.org/packages/b1/66/ab879d60e7f2cd69e69a7f46613108d9d904c29acaaa1adb345a3a479fcd/mmh3-5.3.1-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:427f2ba51baf54ce25f32beb6edd2db70bdc95ac9746067eea0a2b2ca484fd10", size = 106276, upload-time = "2026-09-30T17:36:05.366Z" }, + { url = "https://files.pythonhosted.org/packages/cc/58/cd805eabd1fc01ad36861d3cbf4eb25df822bb0e72c8ee8b3ffd47c71225/mmh3-5.3.1-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:01159489255615d4be76a9ebb07cb0c9b0345f544197aa64ab18a7cfa5579a28", size = 100971, upload-time = "2026-09-30T17:36:06.74Z" }, + { url = "https://files.pythonhosted.org/packages/c2/d0/20d98b665deca070ce5e19df678d476ddab9652462f1e5bf636fc82265a2/mmh3-5.3.1-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:24627cb76ff1e7870a07d7520cf5f3099b1767390236e46d38656dbea5cc6ad0", size = 109795, upload-time = "2026-09-30T17:36:08.14Z" }, + { url = "https://files.pythonhosted.org/packages/09/51/be441d264a38c390582b3b3382f629e66e10847cbcaa60f560d517237b1b/mmh3-5.3.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:8d83f27143c8ae4e78781306ce002ee466219d334346609d0d7f675c8664aef4", size = 97326, upload-time = "2026-09-30T17:36:09.548Z" }, + { url = "https://files.pythonhosted.org/packages/e0/c8/240446abf409338e93d9c8c2e31b47133a08306cbd4346abb0064736f20c/mmh3-5.3.1-cp313-cp313-win32.whl", hash = "sha256:4836a024fe923605d85049f887aacca98add969c8d4932aed5d0d3884cdaa682", size = 40331, upload-time = "2026-09-30T17:36:10.898Z" }, + { url = "https://files.pythonhosted.org/packages/06/5b/b63154d3d8d3dab42a6713df71da40c2952c4e640973a70ebd897df5508c/mmh3-5.3.1-cp313-cp313-win_amd64.whl", hash = "sha256:6759c43a90729a963ab5503779e07cd000c372ebd2d80196f78da2bf2d4101f1", size = 41754, upload-time = "2026-09-30T17:36:12.084Z" }, + { url = "https://files.pythonhosted.org/packages/be/67/b03f7b39d5f22cbfe72e6d73374c820829a69813a7489ba2a4a8d252391b/mmh3-5.3.1-cp313-cp313-win_arm64.whl", hash = "sha256:78219f6b1cf27872295dd4548e862f317b48ef1e1b2c9e0143069ac3a8b822d7", size = 40103, upload-time = "2026-09-30T17:36:13.425Z" }, +] + [[package]] name = "more-itertools" version = "10.8.0" @@ -1732,6 +1881,59 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.45.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2e/02/6e0ae9cc61bd3169d401077b507b3ebc344745171e1051ab430be012dcd9/opentelemetry_api-1.45.1.tar.gz", hash = "sha256:aa38ed19bcc084ba42782a73255b3582283eced7ad6dddbd6695189e69adfb75", size = 72804, upload-time = "2026-10-06T17:32:58.133Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1e/41/f7dcf80b81ee8e71c1a2b59f14208bc723edbd89ed027a73b175abf6348e/opentelemetry_api-1.45.1-py3-none-any.whl", hash = "sha256:b31553efa588ae44bc306f863c785c5333a9ecc091248c6ee68b4b6c87fdedfb", size = 60256, upload-time = "2026-10-06T17:32:33.506Z" }, +] + +[[package]] +name = "opentelemetry-resourcedetector-gcp" +version = "1.14.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-sdk" }, + { name = "requests" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/54/0a95db467d25d81abc16961f5c4538b65f5707e62628a7bb9cdac2849486/opentelemetry_resourcedetector_gcp-1.14.0.tar.gz", hash = "sha256:10b41802acf815838a85c9fe1dae02f5d27ebc17527b77441111c010bead95bc", size = 14831, upload-time = "2026-07-24T19:08:37.901Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/26/75/81a3035969eb44ff4c6a18abc26c5cbff73ba5e4bde6be94298004c3fbb2/opentelemetry_resourcedetector_gcp-1.14.0-py3-none-any.whl", hash = "sha256:cc4818570933b5c651aef80fbf6e8fbe26d63fe6fbc24ce22ae37e78744d30eb", size = 15561, upload-time = "2026-07-24T19:08:36.843Z" }, +] + +[[package]] +name = "opentelemetry-sdk" +version = "1.45.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a1/79/7392e21a1c8f0c61d90b223e31c7e48cb9d452e91a6b820ad24cca5f23c4/opentelemetry_sdk-1.45.1.tar.gz", hash = "sha256:63d24a6ca645019a631e6a51999c73e93adcac1196ca640b8ae78a7cc4762bf3", size = 218324, upload-time = "2026-10-06T17:33:13.26Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/95/3c/87c42b4bd6dd297536f04cd9383d212ac557ecd49f2cbdcd46da1c9ef5c8/opentelemetry_sdk-1.45.1-py3-none-any.whl", hash = "sha256:c604c11dc429810812348989115fa44bd558772a3d7442afc43d024f2c250ca4", size = 140063, upload-time = "2026-10-06T17:32:55.04Z" }, +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.66b1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/46/e4/dbbfb2a010c4db2224a5114638acede6fe563d33cc20fb1752cebcbe6298/opentelemetry_semantic_conventions-0.66b1.tar.gz", hash = "sha256:497ca63bf383723411e8eaf60c8779e9877633c936bb641080adab59d0eb6ec8", size = 150250, upload-time = "2026-10-06T17:33:14.073Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/bc/14/67f8aa798857f8cf686f515bf93d9bb877ce952ddc8efae0fa25b45ce0d6/opentelemetry_semantic_conventions-0.66b1-py3-none-any.whl", hash = "sha256:d4cddeb4315490b35213f55e2bdc9ac54bb1e4d318927475bed62b35545e581b", size = 206279, upload-time = "2026-10-06T17:32:56.103Z" }, +] + [[package]] name = "orderly-set" version = "5.5.0" @@ -2583,11 +2785,11 @@ wheels = [ [[package]] name = "setuptools" -version = "83.0.0" +version = "80.0.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/34/26/f5d29e25ffdb535afef2d35cdb55b325298f96debd670da4c325e08d70f4/setuptools-83.0.0.tar.gz", hash = "sha256:025bccbbf0fa05b6192bc64ae1e7b16e001fd6d6d4d5de03c97b1c1ade523bef", size = 1154254, upload-time = "2026-07-04T15:31:22.699Z" } +sdist = { url = "https://files.pythonhosted.org/packages/44/80/97e25f0f1e4067677806084b7382a6ff9979f3d15119375c475c288db9d7/setuptools-80.0.0.tar.gz", hash = "sha256:c40a5b3729d58dd749c0f08f1a07d134fb8a0a3d7f87dc33e7c5e1f762138650", size = 1354221, upload-time = "2025-04-27T17:21:10.806Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/5d/40/e1e72872c6354b306daef1703549e8e83b4d43cfea356311bf722a043752/setuptools-83.0.0-py3-none-any.whl", hash = "sha256:29b23c360f22f414dc7336bb39178cc7bcbf6021ed2733cde173f09dba19abb3", size = 1008090, upload-time = "2026-07-04T15:31:20.885Z" }, + { url = "https://files.pythonhosted.org/packages/23/63/5517029d6696ddf2bd378d46f63f479be001c31b462303170a1da57650cb/setuptools-80.0.0-py3-none-any.whl", hash = "sha256:a38f898dcd6e5380f4da4381a87ec90bd0a7eec23d204a5552e80ee3cab6bd27", size = 1240907, upload-time = "2025-04-27T17:21:09.175Z" }, ] [[package]] @@ -2659,6 +2861,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e2/22/dbf013a12ec759e54a34a119e9e217435b3f71b2dd5c61a7ade0a25dae87/sqlalchemy-2.0.51-py3-none-any.whl", hash = "sha256:bb024d8b621d0be75f4f44ecc7c950450026e76d66dc8f791bb5331d7fed59d5", size = 1944334, upload-time = "2026-06-15T16:09:22.418Z" }, ] +[[package]] +name = "sqlalchemy-spanner" +version = "1.20.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "alembic" }, + { name = "google-cloud-spanner" }, + { name = "sqlalchemy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/bd/12799120e3e23d1c825ecd9a1afcb8c612b00bcae18b4aed5f7104e14f7c/sqlalchemy_spanner-1.20.1.tar.gz", hash = "sha256:73537add78fe258752545d1e2d3e13be081ba278b0612cbd671b26f6c24fad64", size = 83277, upload-time = "2026-09-03T22:31:23.33Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/30/d7/ddde98fc154aa2c09f19c7734afc4f2247ea41457de51431a217973e289f/sqlalchemy_spanner-1.20.1-py3-none-any.whl", hash = "sha256:bf72778b440e74240453cf755b071469ef54c0972d4a2cc3eb6b6240b9a4a275", size = 32177, upload-time = "2026-09-03T22:30:46.705Z" }, +] + [[package]] name = "sqlfluff" version = "4.2.2"