From 52e64789e66d44b5c51fca3ba980708d94c504eb Mon Sep 17 00:00:00 2001 From: citarf <8773429+citarf@users.noreply.github.com> Date: Fri, 7 Aug 2026 19:20:19 +0200 Subject: [PATCH 1/2] fix(lineage): align station contracts and run evidence --- contracts/bronze-ref-station.odcs.yaml | 24 ++-- contracts/silver-ref-station.odcs.yaml | 40 ++++++- contracts/station-reference.schema.json | 98 ++++++++++++++++ lineage/namespaces.md | 5 + tools/README.md | 7 ++ tools/export_station_reference_schema.py | 54 +++++++++ tools/lineage_run.py | 107 +++++++++++++++++- tools/tests/test_lineage_run.py | 82 ++++++++++++++ .../test_station_reference_governance.py | 38 +++++++ 9 files changed, 444 insertions(+), 11 deletions(-) create mode 100644 contracts/station-reference.schema.json create mode 100644 tools/export_station_reference_schema.py create mode 100644 tools/tests/test_lineage_run.py diff --git a/contracts/bronze-ref-station.odcs.yaml b/contracts/bronze-ref-station.odcs.yaml index d23342b..4f919bc 100644 --- a/contracts/bronze-ref-station.odcs.yaml +++ b/contracts/bronze-ref-station.odcs.yaml @@ -2,7 +2,7 @@ apiVersion: v3.0.2 kind: DataContract id: urn:infoclimat:contract:bronze-ref-station name: Bronze référentiel stations — source legacy immuable -version: 0.1.0 +version: 0.2.0 status: active domain: referentiel dataProduct: ref-station @@ -44,6 +44,18 @@ schema: erronées de 67128/-19.467/55.483, qui restent des valeurs source immuables et auditables. properties: + - name: source_station_id + physicalType: TEXT + required: true + description: Copie textuelle de ic_id utilisée comme clé source stable du pipeline. + - name: source_catalog + physicalType: TEXT + required: true + description: Catalogue physique lu, actuellement warehouse. + - name: source_relation + physicalType: TEXT + required: true + description: Relation physique lue, actuellement dim.station. - name: source_snapshot_id physicalType: TEXT required: true @@ -54,6 +66,10 @@ schema: physicalType: TEXT required: true description: Identifiant du run qui a écrit la capture Bronze. + - name: ingested_at + physicalType: TIMESTAMP + required: true + description: Instant UTC d'archivage de la capture. - name: ic_id physicalType: TEXT required: true @@ -66,12 +82,6 @@ schema: - name: longitude physicalType: DOUBLE description: Longitude WGS84 brute, sans correction ; 67128 peut contenir sa mauvaise coordonnée source. - - name: station_name - physicalType: TEXT - description: Libellé station brut issu de dim.station. - - name: source_payload - physicalType: JSON - description: Enregistrement source complet, conservé pour audit et replay. quality: - rule: uniqueness description: Une capture ne contient qu'une ligne par couple (source_snapshot_id, ic_id). diff --git a/contracts/silver-ref-station.odcs.yaml b/contracts/silver-ref-station.odcs.yaml index d4fdf08..ba49b61 100644 --- a/contracts/silver-ref-station.odcs.yaml +++ b/contracts/silver-ref-station.odcs.yaml @@ -2,7 +2,7 @@ apiVersion: v3.0.2 kind: DataContract id: urn:infoclimat:contract:silver-ref-station name: Silver référentiel stations — canonisation et alias -version: 0.1.0 +version: 0.2.0 status: active domain: referentiel dataProduct: ref-station @@ -49,6 +49,13 @@ schema: primaryKey: true primaryKeyPosition: 1 description: Identifiant de cette ligne de référentiel station. + - name: source_station_id + physicalType: TEXT + required: true + description: Identifiant textuel de la ligne Bronze ayant produit cette décision. + - name: station_uid_source + physicalType: TEXT + description: UID interne de la ligne source, conservé pour les jointures historiques. - name: source_latitude physicalType: DOUBLE description: Latitude WGS84 observée dans Bronze, y compris une valeur erronée de 67128. @@ -85,6 +92,12 @@ schema: mustBeGreaterOrEqualTo: -180 mustBeLessOrEqualTo: 180 severity: error + - name: canonical_ic_id + physicalType: TEXT + description: Identifiant canonique résolu ; nul uniquement pour une ligne quarantined. + - name: canonical_station_uid + physicalType: TEXT + description: UID interne de la station canonique résolue. - name: quality_status physicalType: TEXT required: true @@ -103,6 +116,9 @@ schema: - name: quality_flags physicalType: ARRAY description: Drapeaux explicites motivant la décision qualité. + - name: aliases + physicalType: ARRAY + description: Identifiants obsolètes rattachés à cette ligne lorsqu'elle est canonique. - name: correction_rule_id physicalType: TEXT description: Identifiant versionné de la règle de correction ou de quarantaine appliquée. @@ -120,6 +136,18 @@ schema: physicalType: TEXT required: true description: Run ayant produit ou réconcilié la ligne Silver. + - name: source_catalog + physicalType: TEXT + required: true + description: Catalogue Bronze d'origine. + - name: source_relation + physicalType: TEXT + required: true + description: Relation Bronze d'origine. + - name: ingested_at + physicalType: TIMESTAMP + required: true + description: Instant UTC d'archivage Bronze conservé en Silver. - name: station_alias physicalName: silver_ref.station_alias @@ -134,10 +162,18 @@ schema: primaryKey: true primaryKeyPosition: 1 description: Identifiant station alias, obsolète ou mis en quarantaine. + - name: alias_station_uid + physicalType: TEXT + required: true + description: UID interne historique de l'alias. - name: canonical_ic_id physicalType: TEXT required: true description: Identifiant station canonique à utiliser après résolution. + - name: canonical_station_uid + physicalType: TEXT + required: true + description: UID interne de la station canonique cible. - name: relation physicalType: TEXT required: true @@ -156,9 +192,11 @@ schema: description: Identifiant versionné de la règle ayant créé la relation. - name: source_snapshot_id physicalType: TEXT + required: true description: Capture Bronze à l'origine de la relation. - name: ingest_run_id physicalType: TEXT + required: true description: Run ayant créé la relation d'alias. team: diff --git a/contracts/station-reference.schema.json b/contracts/station-reference.schema.json new file mode 100644 index 0000000..76d0b3c --- /dev/null +++ b/contracts/station-reference.schema.json @@ -0,0 +1,98 @@ +{ + "contentSha256": "sha256:d02750d2bc9f78e1ff63c0f811e1f26fc9d0b86cf8597acc5578e4183c207ac4", + "manifestVersion": "station-reference.schema/v1", + "sourceContracts": { + "bronze-ref-station.odcs.yaml": "0.2.0", + "gold-ref.odcs.yaml": "0.2.0", + "silver-ref-station.odcs.yaml": "0.2.0" + }, + "tables": { + "bronze_ref.station_source": { + "fields": { + "ic_id": {"physicalType": "TEXT", "required": true}, + "ingest_run_id": {"physicalType": "TEXT", "required": true}, + "ingested_at": {"physicalType": "TIMESTAMP", "required": true}, + "latitude": {"physicalType": "DOUBLE", "required": false}, + "longitude": {"physicalType": "DOUBLE", "required": false}, + "source_catalog": {"physicalType": "TEXT", "required": true}, + "source_relation": {"physicalType": "TEXT", "required": true}, + "source_snapshot_id": {"physicalType": "TEXT", "required": true}, + "source_station_id": {"physicalType": "TEXT", "required": true} + } + }, + "gold_ref.station": { + "fields": { + "aliases": {"physicalType": "ARRAY", "required": false}, + "altitude": {"physicalType": "INTEGER", "required": false}, + "departement": {"physicalType": "TEXT", "required": false}, + "derniere_activite": {"physicalType": "TIMESTAMP", "required": false}, + "dh_ouverture": {"physicalType": "TIMESTAMP", "required": false}, + "genre": {"physicalType": "TEXT", "required": false}, + "ic_id": {"physicalType": "TEXT", "required": true}, + "latitude": {"physicalType": "DOUBLE", "required": true}, + "libelle": {"physicalType": "TEXT", "required": false}, + "libelle_source": {"physicalType": "TEXT", "required": false}, + "longitude": {"physicalType": "DOUBLE", "required": true}, + "mfid": {"physicalType": "TEXT", "required": false}, + "ouverte": {"physicalType": "BOOLEAN", "required": false}, + "parametres_mesures": {"physicalType": "ARRAY", "required": false}, + "pays": {"physicalType": "TEXT", "required": false}, + "publiable": {"physicalType": "BOOLEAN", "required": true}, + "station_uid": {"physicalType": "TEXT", "required": true}, + "type_poste": {"physicalType": "TEXT", "required": false} + } + }, + "gold_ref.station_alias": { + "fields": { + "alias_ic_id": {"physicalType": "TEXT", "required": true}, + "canonical_ic_id": {"physicalType": "TEXT", "required": true}, + "relation": {"physicalType": "TEXT", "required": true} + } + }, + "gold_ref.station_parametre": { + "fields": { + "derniere_obs": {"physicalType": "TIMESTAMP", "required": false}, + "n_obs": {"physicalType": "BIGINT", "required": false}, + "parametre": {"physicalType": "TEXT", "required": true}, + "premiere_obs": {"physicalType": "TIMESTAMP", "required": false}, + "station_uid": {"physicalType": "TEXT", "required": true} + } + }, + "silver_ref.station": { + "fields": { + "aliases": {"physicalType": "ARRAY", "required": false}, + "authority_checked_at": {"physicalType": "TIMESTAMP", "required": false}, + "authority_uri": {"physicalType": "TEXT", "required": false}, + "canonical_ic_id": {"physicalType": "TEXT", "required": false}, + "canonical_latitude": {"physicalType": "DOUBLE", "required": false}, + "canonical_longitude": {"physicalType": "DOUBLE", "required": false}, + "canonical_station_uid": {"physicalType": "TEXT", "required": false}, + "correction_rule_id": {"physicalType": "TEXT", "required": false}, + "ic_id": {"physicalType": "TEXT", "required": true}, + "ingest_run_id": {"physicalType": "TEXT", "required": true}, + "ingested_at": {"physicalType": "TIMESTAMP", "required": true}, + "quality_flags": {"physicalType": "ARRAY", "required": false}, + "quality_status": {"physicalType": "TEXT", "required": true}, + "source_catalog": {"physicalType": "TEXT", "required": true}, + "source_latitude": {"physicalType": "DOUBLE", "required": false}, + "source_longitude": {"physicalType": "DOUBLE", "required": false}, + "source_relation": {"physicalType": "TEXT", "required": true}, + "source_snapshot_id": {"physicalType": "TEXT", "required": true}, + "source_station_id": {"physicalType": "TEXT", "required": true}, + "station_uid_source": {"physicalType": "TEXT", "required": false} + } + }, + "silver_ref.station_alias": { + "fields": { + "alias_ic_id": {"physicalType": "TEXT", "required": true}, + "alias_station_uid": {"physicalType": "TEXT", "required": true}, + "canonical_ic_id": {"physicalType": "TEXT", "required": true}, + "canonical_station_uid": {"physicalType": "TEXT", "required": true}, + "correction_rule_id": {"physicalType": "TEXT", "required": true}, + "ingest_run_id": {"physicalType": "TEXT", "required": true}, + "relation": {"physicalType": "TEXT", "required": true}, + "source_snapshot_id": {"physicalType": "TEXT", "required": true} + } + } + } +} diff --git a/lineage/namespaces.md b/lineage/namespaces.md index 3a0966c..d9ed30b 100644 --- a/lineage/namespaces.md +++ b/lineage/namespaces.md @@ -51,6 +51,11 @@ jamais bloquante) + forwarder `tools/lineage_forward.py` → backend Marquez sel 4. Le champ `producer` vaut `https://github.com/infoclimat/site-infoclimat` (versionné). 5. Les pipelines contractualisés référencent leur job via la customProperty `lineageJob` du contrat ODCS. +6. Les cinq jobs `station_ref` peuvent joindre au seul événement terminal le facet + custom `infoclimat_station_reference`. Son JSON `1.0`, borné à 64 KiB, contient + uniquement les snapshot IDs d'entrée/sortie, l'`ingest_run_id`, les comptes de + statuts/flags et la version content-addressed du registre. Les lignes brutes, + chemins de secrets et credentials y sont interdits. ## Pipelines prioritaires à instrumenter diff --git a/tools/README.md b/tools/README.md index 11eaea7..60f28e4 100644 --- a/tools/README.md +++ b/tools/README.md @@ -30,6 +30,13 @@ récurrent (cron / Kestra) qui publiera son résultat. | `check_dual_source.py [--check]` | Couverture/comptes MariaDB ↔ TimescaleDB par station/jour (cibles `audits/dual-source-targets.yaml`) | 2 agrégats bornés par fenêtre, colonnes indexées des deux côtés | | `volumetrie_audit.py` | Inventaire de volumétrie des bases | lecture `information_schema` | +`lineage_run.py` accepte aussi `--evidence ` (ou +`LINEAGE_EVIDENCE_PATH`). Après la commande, il valide un JSON borné à 64 KiB +contenant uniquement les snapshot IDs, l'`ingest_run_id`, les comptes de statuts +et flags et la version du registre. Cette preuve est jointe au facet terminal +`infoclimat_station_reference`; une preuve absente ou invalide reste non +bloquante et aucune ligne brute ou secret n'est admis. + Prérequis Python : `pip install pymysql 'psycopg[binary]' pyyaml 'datacontract-cli[postgres,mysql]'`. ### Pièges connus diff --git a/tools/export_station_reference_schema.py b/tools/export_station_reference_schema.py new file mode 100644 index 0000000..30d6a30 --- /dev/null +++ b/tools/export_station_reference_schema.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +"""Exporte le manifeste JSON station-reference depuis les contrats ODCS.""" + +import hashlib +import json +from pathlib import Path +import sys + +import yaml + + +CONTRACTS = ( + "bronze-ref-station.odcs.yaml", + "silver-ref-station.odcs.yaml", + "gold-ref.odcs.yaml", +) +MANIFEST_VERSION = "station-reference.schema/v1" + + +def build_manifest(root: Path) -> dict: + source_contracts = {} + tables = {} + for name in CONTRACTS: + document = yaml.safe_load((root / "contracts" / name).read_text()) + source_contracts[name] = document["version"] + for table in document["schema"]: + tables[table["physicalName"]] = { + "fields": { + prop["name"]: { + "physicalType": prop["physicalType"], + "required": bool(prop.get("required", False)), + } + for prop in table["properties"] + } + } + payload = { + "manifestVersion": MANIFEST_VERSION, + "sourceContracts": source_contracts, + "tables": tables, + } + digest = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + return {**payload, "contentSha256": f"sha256:{digest}"} + + +def main() -> int: + root = Path(__file__).resolve().parents[1] + print(json.dumps(build_manifest(root), indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/lineage_run.py b/tools/lineage_run.py index 4bfa422..63b8adb 100644 --- a/tools/lineage_run.py +++ b/tools/lineage_run.py @@ -28,6 +28,20 @@ ERROR_FACET_SCHEMA = "https://openlineage.io/spec/facets/1-0-0/ErrorMessageRunFacet.json" DEFAULT_NAMESPACE = "cron://infoclimat" DEFAULT_SPOOL = "/var/spool/lineage/events.jsonl" +DEFAULT_EVIDENCE = "" +EVIDENCE_ENV = "LINEAGE_EVIDENCE_PATH" +EVIDENCE_FACET = "infoclimat_station_reference" +EVIDENCE_FACET_SCHEMA = "https://openlineage.io/spec/2-0-2/OpenLineage.json#/definitions/BaseFacet" +MAX_EVIDENCE_BYTES = 64 * 1024 +EVIDENCE_KEYS = { + "schema_version", + "input_snapshot_ids", + "output_snapshot_ids", + "ingest_run_id", + "status_counts", + "flag_counts", + "registry_version", +} HELP_TEXT = f"""\ Usage: python3 lineage_run.py --job [options] -- [args…] @@ -39,6 +53,7 @@ --job Nom du job (cf. data-platform/lineage/jobs.yaml), requis --namespace Namespace du job (défaut : {DEFAULT_NAMESPACE}) --spool Fichier JSONL de spool (défaut : {DEFAULT_SPOOL}) + --evidence JSON métier borné, lu après la commande (ou {EVIDENCE_ENV}) --help Affiche cette aide Tout ce qui suit `--` est la commande à exécuter, lancée avec les stdio @@ -51,8 +66,19 @@ def parse_cli_args(argv: list) -> tuple: """Sépare argv en (options, commande) autour du premier `--`.""" - options = {"job": "", "namespace": DEFAULT_NAMESPACE, "spool": DEFAULT_SPOOL, "help": False} - flags_with_value = {"--job": "job", "--namespace": "namespace", "--spool": "spool"} + options = { + "job": "", + "namespace": DEFAULT_NAMESPACE, + "spool": DEFAULT_SPOOL, + "evidence": os.environ.get(EVIDENCE_ENV, DEFAULT_EVIDENCE), + "help": False, + } + flags_with_value = { + "--job": "job", + "--namespace": "namespace", + "--spool": "spool", + "--evidence": "evidence", + } command = [] index = 0 while index < len(argv): @@ -125,6 +151,77 @@ def process_facet(command: list, exit_code: int = None, duration: float = None) return facet +def _bounded_string(value, field: str, limit: int = 512) -> str: + if not isinstance(value, str) or not value or len(value) > limit: + raise ValueError(f"{field} must be a non-empty string of at most {limit} characters") + return value + + +def _bounded_string_map(value, field: str, integer_values: bool = False) -> dict: + if not isinstance(value, dict) or len(value) > 32: + raise ValueError(f"{field} must be an object with at most 32 entries") + result = {} + for key, item in value.items(): + key = _bounded_string(key, f"{field} key", 256) + if integer_values: + if not isinstance(item, int) or isinstance(item, bool) or item < 0: + raise ValueError(f"{field}.{key} must be a non-negative integer") + else: + item = _bounded_string(item, f"{field}.{key}", 256) + result[key] = item + return result + + +def load_evidence(path: str) -> dict: + """Charge la preuve métier bornée sans jamais exposer de lignes ou secrets.""" + if not path: + return {} + evidence_path = Path(path) + if evidence_path.stat().st_size > MAX_EVIDENCE_BYTES: + raise ValueError(f"evidence exceeds {MAX_EVIDENCE_BYTES} bytes") + document = json.loads(evidence_path.read_text(encoding="utf-8")) + if not isinstance(document, dict) or set(document) != EVIDENCE_KEYS: + raise ValueError("evidence must contain only the documented station-reference keys") + if document["schema_version"] != "1.0": + raise ValueError("unsupported evidence schema_version") + return { + "schema_version": "1.0", + "input_snapshot_ids": _bounded_string_map( + document["input_snapshot_ids"], "input_snapshot_ids" + ), + "output_snapshot_ids": _bounded_string_map( + document["output_snapshot_ids"], "output_snapshot_ids" + ), + "ingest_run_id": _bounded_string(document["ingest_run_id"], "ingest_run_id"), + "status_counts": _bounded_string_map( + document["status_counts"], "status_counts", integer_values=True + ), + "flag_counts": _bounded_string_map( + document["flag_counts"], "flag_counts", integer_values=True + ), + "registry_version": _bounded_string( + document["registry_version"], "registry_version" + ), + } + + +def evidence_facet(path: str) -> dict: + if not path: + return {} + try: + evidence = load_evidence(path) + except Exception as error: # noqa: BLE001 — preuve non bloquante comme le spool + print(f"lineage_run: preuve métier ignorée ({error})", file=sys.stderr) + return {} + return { + EVIDENCE_FACET: { + "_producer": PRODUCER, + "_schemaURL": EVIDENCE_FACET_SCHEMA, + **evidence, + } + } + + def main(argv: list) -> int: try: options, command = parse_cli_args(argv) @@ -147,7 +244,10 @@ def main(argv: list) -> int: started = time.monotonic() error_message = "" try: - exit_code = subprocess.run(command).returncode # stdio hérités du cron + command_env = os.environ.copy() + if options["evidence"]: + command_env[EVIDENCE_ENV] = options["evidence"] + exit_code = subprocess.run(command, env=command_env).returncode # stdio hérités du cron except OSError as error: # commande introuvable / non exécutable exit_code = 127 error_message = str(error) @@ -155,6 +255,7 @@ def main(argv: list) -> int: duration = time.monotonic() - started facets = {"infoclimat_process": process_facet(command, exit_code, duration)} + facets.update(evidence_facet(options["evidence"])) if exit_code == 0: emit(spool, build_event("COMPLETE", run_id, namespace, job, facets)) else: diff --git a/tools/tests/test_lineage_run.py b/tools/tests/test_lineage_run.py new file mode 100644 index 0000000..9edcc6d --- /dev/null +++ b/tools/tests/test_lineage_run.py @@ -0,0 +1,82 @@ +import json +import os +from pathlib import Path +import sys + +from tools import lineage_run + + +EVIDENCE = { + "schema_version": "1.0", + "input_snapshot_ids": {"iceberg://warehouse/dim.station": "42"}, + "output_snapshot_ids": { + "iceberg://diffusion/bronze_ref.station_source": "101" + }, + "ingest_run_id": "station-ref:42", + "status_counts": {"raw": 3, "canonical": 1}, + "flag_counts": {"outside_declared_territory": 2}, + "registry_version": "station-reference-corrections/v1@sha256:abc", +} + + +def _events(path): + return [json.loads(line) for line in path.read_text().splitlines()] + + +def _writer_command(evidence, exit_code=0): + script = ( + "import json,os,pathlib,sys;" + "p=pathlib.Path(os.environ['LINEAGE_EVIDENCE_PATH']);" + f"p.write_text(json.dumps({evidence!r}));" + f"sys.exit({exit_code})" + ) + return [sys.executable, "-c", script] + + +def test_terminal_complete_event_ingests_namespaced_bounded_evidence(tmp_path): + spool = tmp_path / "events.jsonl" + evidence_path = tmp_path / "run-evidence.json" + result = lineage_run.main([ + "--job", "batch.station_ref_bronze", + "--spool", str(spool), + "--evidence", str(evidence_path), + "--", *_writer_command(EVIDENCE), + ]) + + assert result == 0 + start, complete = _events(spool) + assert "infoclimat_station_reference" not in start["run"]["facets"] + facet = complete["run"]["facets"]["infoclimat_station_reference"] + assert {k: v for k, v in facet.items() if not k.startswith("_")} == EVIDENCE + + +def test_terminal_fail_event_keeps_business_exit_code_and_evidence(tmp_path): + spool = tmp_path / "events.jsonl" + evidence_path = tmp_path / "run-evidence.json" + result = lineage_run.main([ + "--job=batch.station_ref_silver", + f"--spool={spool}", + f"--evidence={evidence_path}", + "--", *_writer_command(EVIDENCE, exit_code=9), + ]) + + assert result == 9 + fail = _events(spool)[-1] + assert fail["eventType"] == "FAIL" + assert fail["run"]["facets"]["infoclimat_station_reference"]["ingest_run_id"] == "station-ref:42" + + +def test_invalid_or_oversized_evidence_is_nonblocking_and_not_attached(tmp_path): + spool = tmp_path / "events.jsonl" + evidence_path = tmp_path / "run-evidence.json" + invalid = {**EVIDENCE, "raw_rows": ["secret"]} + result = lineage_run.main([ + "--job", "batch.station_ref_dim", + "--spool", str(spool), + "--evidence", str(evidence_path), + "--", *_writer_command(invalid), + ]) + + assert result == 0 + complete = _events(spool)[-1] + assert "infoclimat_station_reference" not in complete["run"]["facets"] diff --git a/tools/tests/test_station_reference_governance.py b/tools/tests/test_station_reference_governance.py index 0de6520..f93b3c6 100644 --- a/tools/tests/test_station_reference_governance.py +++ b/tools/tests/test_station_reference_governance.py @@ -1,4 +1,5 @@ import json +import hashlib from pathlib import Path import yaml @@ -159,6 +160,43 @@ def test_station_medallion_contracts_expose_provenance_and_aliases(): for value in ("canonical", "alias_obsolete", "quarantined")) +def test_station_reference_manifest_is_a_versioned_export_of_odcs_contracts(): + manifest_path = ROOT / "contracts" / "station-reference.schema.json" + manifest = json.loads(manifest_path.read_text()) + assert manifest["manifestVersion"] == "station-reference.schema/v1" + assert manifest["sourceContracts"] == { + "bronze-ref-station.odcs.yaml": "0.2.0", + "silver-ref-station.odcs.yaml": "0.2.0", + "gold-ref.odcs.yaml": "0.2.0", + } + + exported = {} + for contract_name in manifest["sourceContracts"]: + document = contract(contract_name) + assert document["version"] == manifest["sourceContracts"][contract_name] + for table_name, table in tables(document).items(): + if table_name in manifest["tables"]: + exported[table_name] = { + prop["name"]: { + "physicalType": prop["physicalType"], + "required": bool(prop.get("required", False)), + } + for prop in table["properties"] + } + assert exported == { + table_name: table["fields"] + for table_name, table in manifest["tables"].items() + } + payload = { + key: manifest[key] + for key in ("manifestVersion", "sourceContracts", "tables") + } + digest = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert manifest["contentSha256"] == f"sha256:{digest}" + + def test_station_67128_governance_is_immutable_canonical_and_not_proximity_based(): bronze_document = contract("bronze-ref-station.odcs.yaml") silver_document = contract("silver-ref-station.odcs.yaml") From 2d7c24594e544945a59028d369a85d2c8a94d6de Mon Sep 17 00:00:00 2001 From: citarf <8773429+citarf@users.noreply.github.com> Date: Fri, 7 Aug 2026 19:56:35 +0200 Subject: [PATCH 2/2] fix(governance): mark station alias crosswalk internal --- contracts/gold-ref.odcs.yaml | 11 +++++++---- contracts/station-reference.schema.json | 7 ++++--- tools/export_station_reference_schema.py | 13 ++++++++++++- .../tests/test_station_reference_governance.py | 17 +++++++++++++++-- 4 files changed, 38 insertions(+), 10 deletions(-) diff --git a/contracts/gold-ref.odcs.yaml b/contracts/gold-ref.odcs.yaml index c204ee8..70e389c 100644 --- a/contracts/gold-ref.odcs.yaml +++ b/contracts/gold-ref.odcs.yaml @@ -2,7 +2,7 @@ apiVersion: v3.0.2 kind: DataContract id: urn:infoclimat:contract:gold-ref name: Gold référentiel stations (annuaire géolocalisé + réseau + activité par paramètre) -version: 0.2.0 +version: 0.2.1 status: draft domain: referentiel dataProduct: gold-ref @@ -121,9 +121,12 @@ schema: physicalName: gold_ref.station_alias physicalType: table description: >- - Relations d'identifiants historiques vers leur station Gold canonique. - Les alias permettent de résoudre les anciens codes sans diffuser les - lignes Silver quarantined. + Crosswalk interne d'identifiants historiques vers leur station Gold + canonique. Il alimente la résolution d'alias de /stations mais n'est ni un + dataset public ni un endpoint /ref/station-alias. + customProperties: + - property: served + value: false properties: - name: alias_ic_id physicalType: TEXT diff --git a/contracts/station-reference.schema.json b/contracts/station-reference.schema.json index 76d0b3c..83ae625 100644 --- a/contracts/station-reference.schema.json +++ b/contracts/station-reference.schema.json @@ -1,9 +1,9 @@ { - "contentSha256": "sha256:d02750d2bc9f78e1ff63c0f811e1f26fc9d0b86cf8597acc5578e4183c207ac4", + "contentSha256": "sha256:5910c785e5eab04313aa4c5dbc8e73124f2e8f44229372678a648c6253a8da66", "manifestVersion": "station-reference.schema/v1", "sourceContracts": { "bronze-ref-station.odcs.yaml": "0.2.0", - "gold-ref.odcs.yaml": "0.2.0", + "gold-ref.odcs.yaml": "0.2.1", "silver-ref-station.odcs.yaml": "0.2.0" }, "tables": { @@ -47,7 +47,8 @@ "alias_ic_id": {"physicalType": "TEXT", "required": true}, "canonical_ic_id": {"physicalType": "TEXT", "required": true}, "relation": {"physicalType": "TEXT", "required": true} - } + }, + "served": false }, "gold_ref.station_parametre": { "fields": { diff --git a/tools/export_station_reference_schema.py b/tools/export_station_reference_schema.py index 30d6a30..5ec3132 100644 --- a/tools/export_station_reference_schema.py +++ b/tools/export_station_reference_schema.py @@ -17,6 +17,13 @@ MANIFEST_VERSION = "station-reference.schema/v1" +def custom_property(item: dict, name: str): + for prop in item.get("customProperties", []): + if prop.get("property") == name: + return prop.get("value") + return None + + def build_manifest(root: Path) -> dict: source_contracts = {} tables = {} @@ -24,7 +31,7 @@ def build_manifest(root: Path) -> dict: document = yaml.safe_load((root / "contracts" / name).read_text()) source_contracts[name] = document["version"] for table in document["schema"]: - tables[table["physicalName"]] = { + exported_table = { "fields": { prop["name"]: { "physicalType": prop["physicalType"], @@ -33,6 +40,10 @@ def build_manifest(root: Path) -> dict: for prop in table["properties"] } } + served = custom_property(table, "served") + if served is not None: + exported_table["served"] = served + tables[table["physicalName"]] = exported_table payload = { "manifestVersion": MANIFEST_VERSION, "sourceContracts": source_contracts, diff --git a/tools/tests/test_station_reference_governance.py b/tools/tests/test_station_reference_governance.py index f93b3c6..b9304b1 100644 --- a/tools/tests/test_station_reference_governance.py +++ b/tools/tests/test_station_reference_governance.py @@ -5,6 +5,7 @@ import yaml from tools.lineage_forward import load_declared_datasets +from tools.export_station_reference_schema import build_manifest ROOT = Path(__file__).parents[2] @@ -26,6 +27,10 @@ def custom_properties(document): return {item["property"]: item["value"] for item in document["customProperties"]} +def table_custom_properties(table): + return {item["property"]: item["value"] for item in table.get("customProperties", [])} + + def dataset_pairs(job, direction): datasets = [ (dataset["namespace"], dataset["name"]) @@ -167,8 +172,10 @@ def test_station_reference_manifest_is_a_versioned_export_of_odcs_contracts(): assert manifest["sourceContracts"] == { "bronze-ref-station.odcs.yaml": "0.2.0", "silver-ref-station.odcs.yaml": "0.2.0", - "gold-ref.odcs.yaml": "0.2.0", + "gold-ref.odcs.yaml": "0.2.1", } + assert manifest == build_manifest(ROOT) + assert manifest["tables"]["gold_ref.station_alias"]["served"] is False exported = {} for contract_name in manifest["sourceContracts"]: @@ -304,7 +311,7 @@ def test_gold_station_contract_exposes_canonical_aliases_and_corrected_lineage() gold = tables(gold_document) gold_properties = custom_properties(gold_document) - assert gold_document["version"] == "0.2.0" + assert gold_document["version"] == "0.2.1" assert properties(gold["gold_ref.station"])["aliases"]["physicalType"] == "ARRAY" alias = properties(gold["gold_ref.station_alias"]) assert {"alias_ic_id", "canonical_ic_id", "relation"} <= alias.keys() @@ -324,3 +331,9 @@ def test_gold_station_contract_exposes_canonical_aliases_and_corrected_lineage() "excludedQualityStatuses": ["alias_obsolete", "quarantined"], "publishedQualityStatuses": ["canonical"], } + + +def test_gold_station_alias_is_governed_as_an_internal_non_served_crosswalk(): + gold = tables(contract("gold-ref.odcs.yaml")) + + assert table_custom_properties(gold["gold_ref.station_alias"])["served"] is False