From d5f8202e3f7c47d5aa380f9d89d62c2dfd5a3446 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?K=C3=BCr=C5=9Fat=20Alpay?= Date: Tue, 2 Jun 2026 16:47:35 +0300 Subject: [PATCH] OPS-REF-002A split source-family ID and SQL helpers --- .../postgresql_source_family_ids.py | 66 +++++++++ .../postgresql_source_family_repository.py | 140 +++--------------- .../postgresql_source_family_sql.py | 88 +++++++++++ ...est_postgresql_source_family_repository.py | 138 +++++++++++++++++ 4 files changed, 312 insertions(+), 120 deletions(-) create mode 100644 src/carbonfactor_parser/persistence/postgresql_source_family_ids.py create mode 100644 src/carbonfactor_parser/persistence/postgresql_source_family_sql.py diff --git a/src/carbonfactor_parser/persistence/postgresql_source_family_ids.py b/src/carbonfactor_parser/persistence/postgresql_source_family_ids.py new file mode 100644 index 0000000..f121742 --- /dev/null +++ b/src/carbonfactor_parser/persistence/postgresql_source_family_ids.py @@ -0,0 +1,66 @@ +"""Stable UUID helpers for PostgreSQL source-family persistence.""" + +from __future__ import annotations + +import json +import uuid + +from carbonfactor_parser.persistence.postgresql_schema_catalog import ( + SourceFamily, + source_family_postgresql_value, +) +from carbonfactor_parser.persistence.source_family_repository import ( + SourceFamilyMasterRecord, +) + + +def source_document_uuid(record: SourceFamilyMasterRecord) -> uuid.UUID: + """Return the stable source document UUID for a master record.""" + + return _stable_uuid( + "source_document", + source_family_postgresql_value(record.source_family), + record.source_document_id, + ) + + +def ingestion_run_uuid(record: SourceFamilyMasterRecord) -> uuid.UUID | None: + """Return the stable ingestion run UUID for a master record.""" + + source = record.ingestion_run_id or record.run_id + if source is None: + source = ( + f"{source_family_postgresql_value(record.source_family)}:" + f"{record.source_year}:" + f"{record.source_version}" + ) + return _stable_uuid( + "ingestion_run", + source_family_postgresql_value(record.source_family), + source, + ) + + +def master_uuid(source_family: SourceFamily, master_id: str) -> uuid.UUID: + """Return the stable source-family master UUID.""" + + return _stable_uuid("master", source_family_postgresql_value(source_family), master_id) + + +def detail_uuid(source_family: SourceFamily, detail_id: str) -> uuid.UUID: + """Return the stable source-family detail UUID.""" + + return _stable_uuid("detail", source_family_postgresql_value(source_family), detail_id) + + +def _stable_uuid(*values: object) -> uuid.UUID: + payload = json.dumps(tuple(str(value) for value in values), separators=(",", ":")) + return uuid.uuid5(uuid.NAMESPACE_URL, payload) + + +__all__ = ( + "detail_uuid", + "ingestion_run_uuid", + "master_uuid", + "source_document_uuid", +) diff --git a/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py b/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py index 522f9b2..3408b67 100644 --- a/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py +++ b/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py @@ -6,7 +6,6 @@ from decimal import Decimal from enum import Enum import json -import uuid from typing import Mapping from carbonfactor_parser.diagnostics.redaction import redact_sensitive_text @@ -15,9 +14,17 @@ persist_parsed_factor_records, ) from carbonfactor_parser.persistence.postgresql_schema_catalog import ( - SourceFamily, source_family_postgresql_value, - source_family_table_prefix, +) +from carbonfactor_parser.persistence.postgresql_source_family_ids import ( + detail_uuid, + ingestion_run_uuid, + master_uuid, + source_document_uuid, +) +from carbonfactor_parser.persistence.postgresql_source_family_sql import ( + detail_insert_sql, + master_insert_sql, ) from carbonfactor_parser.persistence.source_family_repository import ( SourceFamilyDetailRecord, @@ -25,7 +32,6 @@ SourceFamilyRepositoryIssue, SourceFamilyRepositoryPersistResult, SourceFamilyRepositoryPersistStatus, - source_family_repository_table_names, validate_source_family_repository_inputs, ) @@ -145,7 +151,7 @@ def persist_source_family_records( if _fetchone( _execute( self._connection, - _master_insert_sql(master.source_family), + master_insert_sql(master.source_family), _master_parameters(master), ) ) is not None: @@ -155,7 +161,7 @@ def persist_source_family_records( if _fetchone( _execute( self._connection, - _detail_insert_sql(detail.source_family), + detail_insert_sql(detail.source_family), _detail_parameters(detail), ) ) is not None: @@ -188,7 +194,7 @@ def persist_source_family_records( ) def _ensure_ingestion_run(self, master: SourceFamilyMasterRecord) -> None: - ingestion_run_id = _ingestion_run_uuid(master) + ingestion_run_id = ingestion_run_uuid(master) if ingestion_run_id is None: return _execute( @@ -226,8 +232,8 @@ def _ensure_source_document(self, master: SourceFamilyMasterRecord) -> None: DO NOTHING """, ( - str(_source_document_uuid(master)), - str(_ingestion_run_uuid(master)), + str(source_document_uuid(master)), + str(ingestion_run_uuid(master)), source_family_postgresql_value(master.source_family), master.artifact_reference or master.source_document_id, master.artifact_checksum_sha256 or "checksum-unavailable", @@ -236,85 +242,15 @@ def _ensure_source_document(self, master: SourceFamilyMasterRecord) -> None: ) -def _master_insert_sql(source_family: SourceFamily) -> str: - master_table, _detail_table = source_family_repository_table_names(source_family) - family_prefix = source_family_table_prefix(source_family) - master_id = f"{family_prefix}_emission_factor_master_id" - return f""" - INSERT INTO {master_table} ( - {master_id}, - source_family, - source_year, - source_version, - source_release, - source_document_id, - ingestion_run_id, - run_id, - master_external_key, - status, - artifact_reference, - artifact_checksum_sha256, - archive_reference, - archive_checksum_sha256, - effective_from, - effective_to, - record_checksum_sha256, - metadata, - created_at, - updated_at - ) - VALUES ( - %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, - %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NOW(), NOW() - ) - ON CONFLICT (source_family, source_year, source_version, master_external_key) - DO NOTHING - RETURNING {master_id} - """ - - -def _detail_insert_sql(source_family: SourceFamily) -> str: - master_table, detail_table = source_family_repository_table_names(source_family) - del master_table - family_prefix = source_family_table_prefix(source_family) - master_id = f"{family_prefix}_emission_factor_master_id" - detail_id = f"{family_prefix}_emission_factor_detail_id" - return f""" - INSERT INTO {detail_table} ( - {detail_id}, - {master_id}, - detail_external_key, - source_row_number, - factor_id, - factor_name, - factor_value, - factor_unit, - status, - record_checksum_sha256, - raw_fields, - normalized_fields, - created_at, - updated_at - ) - VALUES ( - %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, - %s::jsonb, %s::jsonb, NOW(), NOW() - ) - ON CONFLICT ({master_id}, detail_external_key) - DO NOTHING - RETURNING {detail_id} - """ - - def _master_parameters(record: SourceFamilyMasterRecord) -> tuple[object, ...]: return ( - str(_master_uuid(record.source_family, record.source_family_master_id)), + str(master_uuid(record.source_family, record.source_family_master_id)), source_family_postgresql_value(record.source_family), record.source_year, record.source_version, record.source_release, - str(_source_document_uuid(record)), - str(_ingestion_run_uuid(record)) if _ingestion_run_uuid(record) else None, + str(source_document_uuid(record)), + str(ingestion_run_uuid(record)) if ingestion_run_uuid(record) else None, record.run_id, record.master_external_key, record.status, @@ -331,8 +267,8 @@ def _master_parameters(record: SourceFamilyMasterRecord) -> tuple[object, ...]: def _detail_parameters(record: SourceFamilyDetailRecord) -> tuple[object, ...]: return ( - str(_detail_uuid(record.source_family, record.source_family_detail_id)), - str(_master_uuid(record.source_family, record.source_family_master_id)), + str(detail_uuid(record.source_family, record.source_family_detail_id)), + str(master_uuid(record.source_family, record.source_family_master_id)), record.detail_external_key, record.source_row_number, record.factor_id, @@ -346,42 +282,6 @@ def _detail_parameters(record: SourceFamilyDetailRecord) -> tuple[object, ...]: ) -def _source_document_uuid(record: SourceFamilyMasterRecord) -> uuid.UUID: - return _stable_uuid( - "source_document", - source_family_postgresql_value(record.source_family), - record.source_document_id, - ) - - -def _ingestion_run_uuid(record: SourceFamilyMasterRecord) -> uuid.UUID | None: - source = record.ingestion_run_id or record.run_id - if source is None: - source = ( - f"{source_family_postgresql_value(record.source_family)}:" - f"{record.source_year}:" - f"{record.source_version}" - ) - return _stable_uuid( - "ingestion_run", - source_family_postgresql_value(record.source_family), - source, - ) - - -def _master_uuid(source_family: SourceFamily, master_id: str) -> uuid.UUID: - return _stable_uuid("master", source_family_postgresql_value(source_family), master_id) - - -def _detail_uuid(source_family: SourceFamily, detail_id: str) -> uuid.UUID: - return _stable_uuid("detail", source_family_postgresql_value(source_family), detail_id) - - -def _stable_uuid(*values: object) -> uuid.UUID: - payload = json.dumps(tuple(str(value) for value in values), separators=(",", ":")) - return uuid.uuid5(uuid.NAMESPACE_URL, payload) - - def _json_payload(value: Mapping[str, object]) -> str: return json.dumps(_json_safe(value), sort_keys=True, separators=(",", ":")) diff --git a/src/carbonfactor_parser/persistence/postgresql_source_family_sql.py b/src/carbonfactor_parser/persistence/postgresql_source_family_sql.py new file mode 100644 index 0000000..67b0dfd --- /dev/null +++ b/src/carbonfactor_parser/persistence/postgresql_source_family_sql.py @@ -0,0 +1,88 @@ +"""SQL builders for PostgreSQL source-family master/detail inserts.""" + +from __future__ import annotations + +from carbonfactor_parser.persistence.postgresql_schema_catalog import ( + SourceFamily, + source_family_table_prefix, +) +from carbonfactor_parser.persistence.source_family_repository import ( + source_family_repository_table_names, +) + + +def master_insert_sql(source_family: SourceFamily) -> str: + """Return the source-family master INSERT statement.""" + + master_table, _detail_table = source_family_repository_table_names(source_family) + family_prefix = source_family_table_prefix(source_family) + master_id = f"{family_prefix}_emission_factor_master_id" + return f""" + INSERT INTO {master_table} ( + {master_id}, + source_family, + source_year, + source_version, + source_release, + source_document_id, + ingestion_run_id, + run_id, + master_external_key, + status, + artifact_reference, + artifact_checksum_sha256, + archive_reference, + archive_checksum_sha256, + effective_from, + effective_to, + record_checksum_sha256, + metadata, + created_at, + updated_at + ) + VALUES ( + %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s, %s, %s, %s, %s, %s, %s, %s::jsonb, NOW(), NOW() + ) + ON CONFLICT (source_family, source_year, source_version, master_external_key) + DO NOTHING + RETURNING {master_id} + """ + + +def detail_insert_sql(source_family: SourceFamily) -> str: + """Return the source-family detail INSERT statement.""" + + master_table, detail_table = source_family_repository_table_names(source_family) + del master_table + family_prefix = source_family_table_prefix(source_family) + master_id = f"{family_prefix}_emission_factor_master_id" + detail_id = f"{family_prefix}_emission_factor_detail_id" + return f""" + INSERT INTO {detail_table} ( + {detail_id}, + {master_id}, + detail_external_key, + source_row_number, + factor_id, + factor_name, + factor_value, + factor_unit, + status, + record_checksum_sha256, + raw_fields, + normalized_fields, + created_at, + updated_at + ) + VALUES ( + %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s::jsonb, %s::jsonb, NOW(), NOW() + ) + ON CONFLICT ({master_id}, detail_external_key) + DO NOTHING + RETURNING {detail_id} + """ + + +__all__ = ("detail_insert_sql", "master_insert_sql") diff --git a/tests/test_postgresql_source_family_repository.py b/tests/test_postgresql_source_family_repository.py index 7ad0331..8fe87d6 100644 --- a/tests/test_postgresql_source_family_repository.py +++ b/tests/test_postgresql_source_family_repository.py @@ -1,6 +1,8 @@ from __future__ import annotations +from dataclasses import replace from decimal import Decimal +import json import os from pathlib import Path import uuid @@ -37,6 +39,16 @@ from carbonfactor_parser.persistence.postgresql_runtime_schema_bootstrap import ( bootstrap_postgresql_phase1_schema, ) +from carbonfactor_parser.persistence.postgresql_source_family_sql import ( + detail_insert_sql, + master_insert_sql, +) +from carbonfactor_parser.persistence.postgresql_source_family_ids import ( + detail_uuid, + ingestion_run_uuid, + master_uuid, + source_document_uuid, +) from carbonfactor_parser.persistence.postgresql_source_family_repository import ( PostgreSQLSourceFamilyRuntimeRepository, PostgreSQLSourceSpecificFactorInsertStatus, @@ -85,6 +97,122 @@ def rollback(self) -> None: self.rollback_count += 1 + +def test_postgresql_source_family_stable_uuid_helpers_match_legacy_payloads() -> None: + command = build_parsed_factor_persistence_command(_payload("ghg_protocol")) + master = command.master_records[0] + detail = command.detail_records[0] + + assert source_document_uuid(master) == _legacy_stable_uuid( + "source_document", + "ghg_protocol", + master.source_document_id, + ) + ingestion_source = master.ingestion_run_id or master.run_id + if ingestion_source is None: + ingestion_source = ( + f"ghg_protocol:{master.source_year}:{master.source_version}" + ) + assert ingestion_run_uuid(master) == _legacy_stable_uuid( + "ingestion_run", + "ghg_protocol", + ingestion_source, + ) + assert master_uuid(master.source_family, master.source_family_master_id) == ( + _legacy_stable_uuid( + "master", + "ghg_protocol", + master.source_family_master_id, + ) + ) + assert detail_uuid(detail.source_family, detail.source_family_detail_id) == ( + _legacy_stable_uuid( + "detail", + "ghg_protocol", + detail.source_family_detail_id, + ) + ) + assert source_document_uuid(master) == source_document_uuid(master) + assert ingestion_run_uuid(master) == ingestion_run_uuid(master) + + +def test_postgresql_source_family_ingestion_run_uuid_fallback_is_deterministic() -> None: + command = build_parsed_factor_persistence_command(_payload("defra_desnz")) + master = replace(command.master_records[0], ingestion_run_id=None, run_id=None) + + expected = _legacy_stable_uuid( + "ingestion_run", + "defra_desnz", + f"defra_desnz:{master.source_year}:{master.source_version}", + ) + + assert ingestion_run_uuid(master) == expected + assert ingestion_run_uuid(master) == ingestion_run_uuid(master) + + +@pytest.mark.parametrize( + ("source_family", "master_table", "master_id"), + ( + ("ghg_protocol", "ghg_emission_factor_masters", "ghg_emission_factor_master_id"), + ("defra_desnz", "defra_emission_factor_masters", "defra_emission_factor_master_id"), + ("ipcc_efdb", "ipcc_emission_factor_masters", "ipcc_emission_factor_master_id"), + ), +) +def test_postgresql_source_family_master_insert_sql_compatibility( + source_family: str, + master_table: str, + master_id: str, +) -> None: + sql = _normalized_sql(master_insert_sql(source_family)) + + assert f"INSERT INTO {master_table}" in sql + assert ( + "ON CONFLICT (source_family, source_year, source_version, master_external_key)" + in sql + ) + assert f"RETURNING {master_id}" in sql + assert "metadata" in sql + assert "%s::jsonb" in sql + + +@pytest.mark.parametrize( + ("source_family", "detail_table", "master_id", "detail_id"), + ( + ( + "ghg_protocol", + "ghg_emission_factor_details", + "ghg_emission_factor_master_id", + "ghg_emission_factor_detail_id", + ), + ( + "defra_desnz", + "defra_emission_factor_details", + "defra_emission_factor_master_id", + "defra_emission_factor_detail_id", + ), + ( + "ipcc_efdb", + "ipcc_emission_factor_details", + "ipcc_emission_factor_master_id", + "ipcc_emission_factor_detail_id", + ), + ), +) +def test_postgresql_source_family_detail_insert_sql_compatibility( + source_family: str, + detail_table: str, + master_id: str, + detail_id: str, +) -> None: + sql = _normalized_sql(detail_insert_sql(source_family)) + + assert f"INSERT INTO {detail_table}" in sql + assert f"ON CONFLICT ({master_id}, detail_external_key)" in sql + assert f"RETURNING {detail_id}" in sql + assert "raw_fields" in sql + assert "normalized_fields" in sql + assert sql.count("%s::jsonb") == 2 + def test_postgresql_source_family_repository_inserts_and_skips_idempotently() -> None: connection = _FakeConnection() repository = PostgreSQLSourceFamilyRuntimeRepository(connection) @@ -305,3 +433,13 @@ def __init__(self, exc: Exception) -> None: def execute(self, statement: str, parameters: object | None = None) -> _FakeCursor: raise self._exc + + + +def _legacy_stable_uuid(*values: object) -> uuid.UUID: + payload = json.dumps(tuple(str(value) for value in values), separators=(",", ":")) + return uuid.uuid5(uuid.NAMESPACE_URL, payload) + + +def _normalized_sql(statement: str) -> str: + return " ".join(statement.split())