OPS-REF-002B: Split PostgreSQL source-family parameter mapping

src/carbonfactor_parser/persistence/postgresql_source_family_parameters.py

@@ -0,0 +1,95 @@
+"""PostgreSQL source-family master/detail parameter mapping helpers."""
+
+from __future__ import annotations
+
+from decimal import Decimal
+import json
+from typing import Mapping
+
+from carbonfactor_parser.persistence.postgresql_schema_catalog import (
+    source_family_postgresql_value,
+)
+from carbonfactor_parser.persistence.postgresql_source_family_ids import (
+    detail_uuid,
+    ingestion_run_uuid,
+    master_uuid,
+    source_document_uuid,
+)
+from carbonfactor_parser.persistence.source_family_repository import (
+    SourceFamilyDetailRecord,
+    SourceFamilyMasterRecord,
+)
+
+
+def master_parameters(record: SourceFamilyMasterRecord) -> tuple[object, ...]:
+    """Return PostgreSQL parameters for a source-family master row."""
+
+    ingestion_id = ingestion_run_uuid(record)
+    return (
+        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_id) if ingestion_id else None,
+        record.run_id,
+        record.master_external_key,
+        record.status,
+        record.artifact_reference,
+        record.artifact_checksum_sha256,
+        record.archive_reference,
+        record.archive_checksum_sha256,
+        record.effective_from,
+        record.effective_to,
+        record.record_checksum_sha256,
+        json_payload(record.metadata),
+    )
+
+
+def detail_parameters(record: SourceFamilyDetailRecord) -> tuple[object, ...]:
+    """Return PostgreSQL parameters for a source-family detail row."""
+
+    return (
+        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,
+        record.factor_name,
+        str(Decimal(str(record.factor_value))),
+        record.factor_unit,
+        record.status,
+        record.record_checksum_sha256,
+        json_payload(record.raw_fields),
+        json_payload(record.normalized_fields),
+    )
+
+
+def json_payload(value: Mapping[str, object]) -> str:
+    """Return compact, sorted JSON for PostgreSQL JSONB parameters."""
+
+    return json.dumps(json_safe(value), sort_keys=True, separators=(",", ":"))
+
+
+def json_safe(value: object) -> object:
+    """Convert values that are not JSON-native while preserving payload shape."""
+
+    if isinstance(value, Decimal):
+        return str(value)
+    if isinstance(value, Mapping):
+        return {
+            str(key): json_safe(item)
+            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
+        }
+    if isinstance(value, tuple | list):
+        return [json_safe(item) for item in value]
+    return value
+
+
+__all__ = (
+    "detail_parameters",
+    "json_payload",
+    "json_safe",
+    "master_parameters",
+)

src/carbonfactor_parser/persistence/postgresql_source_family_repository.py

@@ -3,10 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from decimal import Decimal
 from enum import Enum
-import json
-from typing import Mapping
 
 from carbonfactor_parser.diagnostics.redaction import redact_sensitive_text
 
@@ -17,11 +14,13 @@
     source_family_postgresql_value,
 )
 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_parameters import (
+    detail_parameters,
+    master_parameters,
+)
 from carbonfactor_parser.persistence.postgresql_source_family_sql import (
     detail_insert_sql,
     master_insert_sql,
@@ -152,7 +151,7 @@ def persist_source_family_records(
                     _execute(
                         self._connection,
                         master_insert_sql(master.source_family),
-                        _master_parameters(master),
+                        master_parameters(master),
                     )
                 ) is not None:
                     inserted_masters += 1
@@ -162,7 +161,7 @@ def persist_source_family_records(
                     _execute(
                         self._connection,
                         detail_insert_sql(detail.source_family),
-                        _detail_parameters(detail),
+                        detail_parameters(detail),
                     )
                 ) is not None:
                     inserted_details += 1
@@ -242,63 +241,6 @@ def _ensure_source_document(self, master: SourceFamilyMasterRecord) -> None:
         )
 
 
-def _master_parameters(record: SourceFamilyMasterRecord) -> tuple[object, ...]:
-    return (
-        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,
-        record.run_id,
-        record.master_external_key,
-        record.status,
-        record.artifact_reference,
-        record.artifact_checksum_sha256,
-        record.archive_reference,
-        record.archive_checksum_sha256,
-        record.effective_from,
-        record.effective_to,
-        record.record_checksum_sha256,
-        _json_payload(record.metadata),
-    )
-
-
-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)),
-        record.detail_external_key,
-        record.source_row_number,
-        record.factor_id,
-        record.factor_name,
-        str(Decimal(str(record.factor_value))),
-        record.factor_unit,
-        record.status,
-        record.record_checksum_sha256,
-        _json_payload(record.raw_fields),
-        _json_payload(record.normalized_fields),
-    )
-
-
-def _json_payload(value: Mapping[str, object]) -> str:
-    return json.dumps(_json_safe(value), sort_keys=True, separators=(",", ":"))
-
-
-def _json_safe(value: object) -> object:
-    if isinstance(value, Decimal):
-        return str(value)
-    if isinstance(value, Mapping):
-        return {
-            str(key): _json_safe(item)
-            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
-        }
-    if isinstance(value, tuple | list):
-        return [_json_safe(item) for item in value]
-    return value
-
-
 def _execute(
     connection: object,
     statement: str,

tests/test_postgresql_source_family_repository.py

@@ -49,6 +49,12 @@
     master_uuid,
     source_document_uuid,
 )
+from carbonfactor_parser.persistence.postgresql_source_family_parameters import (
+    detail_parameters,
+    json_payload,
+    json_safe,
+    master_parameters,
+)
 from carbonfactor_parser.persistence.postgresql_source_family_repository import (
     PostgreSQLSourceFamilyRuntimeRepository,
     PostgreSQLSourceSpecificFactorInsertStatus,
@@ -213,6 +219,80 @@ def test_postgresql_source_family_detail_insert_sql_compatibility(
     assert "normalized_fields" in sql
     assert sql.count("%s::jsonb") == 2
 
+
+def test_postgresql_source_family_master_parameters_compatibility() -> None:
+    command = build_parsed_factor_persistence_command(_payload("ghg_protocol"))
+    master = command.master_records[0]
+
+    parameters = master_parameters(master)
+
+    assert len(parameters) == 18
+    assert parameters[0] == str(
+        master_uuid(master.source_family, master.source_family_master_id)
+    )
+    assert parameters[1] == "ghg_protocol"
+    assert parameters[2] == master.source_year
+    assert parameters[3] == master.source_version
+    assert parameters[5] == str(source_document_uuid(master))
+    assert parameters[6] == str(ingestion_run_uuid(master))
+    assert parameters[8] == master.master_external_key
+    assert parameters[16] == master.record_checksum_sha256
+    assert parameters[17] == json.dumps(
+        json_safe(master.metadata), sort_keys=True, separators=(",", ":")
+    )
+    assert ": " not in str(parameters[17])
+    assert ", " not in str(parameters[17])
+
+
+def test_postgresql_source_family_detail_parameters_compatibility() -> None:
+    command = build_parsed_factor_persistence_command(_payload("defra_desnz"))
+    detail = command.detail_records[0]
+
+    parameters = detail_parameters(detail)
+
+    assert len(parameters) == 12
+    assert parameters[0] == str(
+        detail_uuid(detail.source_family, detail.source_family_detail_id)
+    )
+    assert parameters[1] == str(
+        master_uuid(detail.source_family, detail.source_family_master_id)
+    )
+    assert parameters[2] == detail.detail_external_key
+    assert parameters[3] == detail.source_row_number
+    assert parameters[4] == detail.factor_id
+    assert parameters[6] == str(Decimal(str(detail.factor_value)))
+    assert parameters[9] == detail.record_checksum_sha256
+    assert parameters[10] == json.dumps(
+        json_safe(detail.raw_fields), sort_keys=True, separators=(",", ":")
+    )
+    assert parameters[11] == json.dumps(
+        json_safe(detail.normalized_fields), sort_keys=True, separators=(",", ":")
+    )
+    assert ": " not in str(parameters[10])
+    assert ", " not in str(parameters[10])
+    assert ": " not in str(parameters[11])
+    assert ", " not in str(parameters[11])
+
+
+def test_postgresql_source_family_json_helpers_preserve_legacy_behavior() -> None:
+    payload = {
+        2: (Decimal("1.20"), [Decimal("3.40"), {"b": 2, "a": Decimal("5")}]),
+        "10": "ten",
+        "a": None,
+    }
+
+    safe_payload = json_safe(payload)
+
+    assert safe_payload == {
+        "10": "ten",
+        "2": ["1.20", ["3.40", {"a": "5", "b": 2}]],
+        "a": None,
+    }
+    assert json_payload(payload) == (
+        '{"10":"ten","2":["1.20",["3.40",{"a":"5","b":2}]],"a":null}'
+    )
+
+
 def test_postgresql_source_family_repository_inserts_and_skips_idempotently() -> None:
     connection = _FakeConnection()
     repository = PostgreSQLSourceFamilyRuntimeRepository(connection)