Split source-family ingestion-run and source-document upserts into helpers

src/carbonfactor_parser/persistence/postgresql_source_family_repository.py

@@ -10,13 +10,6 @@
 from carbonfactor_parser.persistence.parsed_factor_persistence_writer import (
     persist_parsed_factor_records,
 )
-from carbonfactor_parser.persistence.postgresql_schema_catalog import (
-    source_family_postgresql_value,
-)
-from carbonfactor_parser.persistence.postgresql_source_family_ids import (
-    ingestion_run_uuid,
-    source_document_uuid,
-)
 from carbonfactor_parser.persistence.postgresql_source_family_parameters import (
     detail_parameters,
     master_parameters,
@@ -25,6 +18,10 @@
     detail_insert_sql,
     master_insert_sql,
 )
+from carbonfactor_parser.persistence.postgresql_source_family_upserts import (
+    ensure_ingestion_run,
+    ensure_source_document,
+)
 from carbonfactor_parser.persistence.source_family_repository import (
     SourceFamilyDetailRecord,
     SourceFamilyMasterRecord,
@@ -145,8 +142,8 @@ def persist_source_family_records(
         inserted_details = 0
         try:
             for master in master_records:
-                self._ensure_ingestion_run(master)
-                self._ensure_source_document(master)
+                ensure_ingestion_run(self._connection, master)
+                ensure_source_document(self._connection, master)
                 if _fetchone(
                     _execute(
                         self._connection,
@@ -192,54 +189,6 @@ def persist_source_family_records(
             skipped_detail_count=len(detail_records) - inserted_details,
         )
 
-    def _ensure_ingestion_run(self, master: SourceFamilyMasterRecord) -> None:
-        ingestion_run_id = ingestion_run_uuid(master)
-        if ingestion_run_id is None:
-            return
-        _execute(
-            self._connection,
-            """
-            INSERT INTO ingestion_runs (
-                ingestion_run_id,
-                run_status,
-                created_at,
-                updated_at
-            )
-            VALUES (%s, %s, NOW(), NOW())
-            ON CONFLICT (ingestion_run_id) DO NOTHING
-            """,
-            (str(ingestion_run_id), "completed"),
-        )
-
-    def _ensure_source_document(self, master: SourceFamilyMasterRecord) -> None:
-        _execute(
-            self._connection,
-            """
-            INSERT INTO source_documents (
-                source_document_id,
-                ingestion_run_id,
-                source_family,
-                source_document_uri,
-                source_checksum_sha256,
-                acquisition_status,
-                acquired_at,
-                created_at,
-                updated_at
-            )
-            VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW())
-            ON CONFLICT (source_family, source_document_uri, source_checksum_sha256)
-            DO NOTHING
-            """,
-            (
-                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",
-                "downloaded",
-            ),
-        )
-
 
 def _execute(
     connection: object,

src/carbonfactor_parser/persistence/postgresql_source_family_upserts.py

@@ -0,0 +1,85 @@
+"""PostgreSQL source-family source-document and ingestion-run upserts."""
+
+from __future__ import annotations
+
+from carbonfactor_parser.persistence.postgresql_schema_catalog import (
+    source_family_postgresql_value,
+)
+from carbonfactor_parser.persistence.postgresql_source_family_ids import (
+    ingestion_run_uuid,
+    source_document_uuid,
+)
+from carbonfactor_parser.persistence.source_family_repository import (
+    SourceFamilyMasterRecord,
+)
+
+
+def ensure_ingestion_run(connection: object, master: SourceFamilyMasterRecord) -> None:
+    """Ensure the source-family ingestion run row exists."""
+
+    ingestion_run_id = ingestion_run_uuid(master)
+    if ingestion_run_id is None:
+        return
+    _execute(
+        connection,
+        """
+        INSERT INTO ingestion_runs (
+            ingestion_run_id,
+            run_status,
+            created_at,
+            updated_at
+        )
+        VALUES (%s, %s, NOW(), NOW())
+        ON CONFLICT (ingestion_run_id) DO NOTHING
+        """,
+        (str(ingestion_run_id), "completed"),
+    )
+
+
+def ensure_source_document(connection: object, master: SourceFamilyMasterRecord) -> None:
+    """Ensure the source-family source document row exists."""
+
+    _execute(
+        connection,
+        """
+        INSERT INTO source_documents (
+            source_document_id,
+            ingestion_run_id,
+            source_family,
+            source_document_uri,
+            source_checksum_sha256,
+            acquisition_status,
+            acquired_at,
+            created_at,
+            updated_at
+        )
+        VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW())
+        ON CONFLICT (source_family, source_document_uri, source_checksum_sha256)
+        DO NOTHING
+        """,
+        (
+            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",
+            "downloaded",
+        ),
+    )
+
+
+def _execute(
+    connection: object,
+    statement: str,
+    parameters: object | None = None,
+) -> object:
+    execute = getattr(connection, "execute")
+    if parameters is None:
+        return execute(statement)
+    return execute(statement, parameters)
+
+
+__all__ = (
+    "ensure_ingestion_run",
+    "ensure_source_document",
+)

tests/test_postgresql_source_family_repository.py

@@ -39,6 +39,9 @@
 from carbonfactor_parser.persistence.postgresql_runtime_schema_bootstrap import (
     bootstrap_postgresql_phase1_schema,
 )
+from carbonfactor_parser.persistence.postgresql_schema_catalog import (
+    source_family_postgresql_value,
+)
 from carbonfactor_parser.persistence.postgresql_source_family_sql import (
     detail_insert_sql,
     master_insert_sql,
@@ -59,6 +62,10 @@
     PostgreSQLSourceFamilyRuntimeRepository,
     PostgreSQLSourceSpecificFactorInsertStatus,
 )
+from carbonfactor_parser.persistence.postgresql_source_family_upserts import (
+    ensure_ingestion_run,
+    ensure_source_document,
+)
 
 
 class _FakeCursor:
@@ -156,6 +163,90 @@ def test_postgresql_source_family_ingestion_run_uuid_fallback_is_deterministic()
     assert ingestion_run_uuid(master) == ingestion_run_uuid(master)
 
 
+def test_postgresql_source_family_ensure_ingestion_run_upsert_compatibility() -> None:
+    connection = _FakeConnection()
+    command = build_parsed_factor_persistence_command(_payload("ipcc_efdb"))
+    master = replace(
+        command.master_records[0],
+        source_year=2025,
+        source_version="ipcc-2025",
+        source_release="release-a",
+        ingestion_run_id=None,
+        run_id="cycle-run-001",
+    )
+
+    ensure_ingestion_run(connection, master)
+
+    assert len(connection.statements) == 1
+    statement, parameters = connection.statements[0]
+    sql = _normalized_sql(statement)
+    assert "INSERT INTO ingestion_runs" in sql
+    assert "ingestion_run_id" in sql
+    assert "run_status" in sql
+    assert "VALUES (%s, %s, NOW(), NOW())" in sql
+    assert "ON CONFLICT (ingestion_run_id) DO NOTHING" in sql
+    assert parameters == (str(ingestion_run_uuid(master)), "completed")
+    assert ingestion_run_uuid(master) == _legacy_stable_uuid(
+        "ingestion_run",
+        source_family_postgresql_value(master.source_family),
+        master.run_id,
+    )
+    assert master.source_year == 2025
+    assert master.source_version == "ipcc-2025"
+    assert master.source_release == "release-a"
+
+
+def test_postgresql_source_family_ensure_source_document_upsert_compatibility() -> None:
+    connection = _FakeConnection()
+    command = build_parsed_factor_persistence_command(_payload("ghg_protocol"))
+    master = replace(
+        command.master_records[0],
+        artifact_reference=None,
+        artifact_checksum_sha256=None,
+    )
+
+    ensure_source_document(connection, master)
+
+    assert len(connection.statements) == 1
+    statement, parameters = connection.statements[0]
+    sql = _normalized_sql(statement)
+    assert "INSERT INTO source_documents" in sql
+    assert "source_document_id" in sql
+    assert "ingestion_run_id" in sql
+    assert "source_family" in sql
+    assert "source_document_uri" in sql
+    assert "source_checksum_sha256" in sql
+    assert "acquisition_status" in sql
+    assert "VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW())" in sql
+    assert (
+        "ON CONFLICT (source_family, source_document_uri, source_checksum_sha256)"
+        in sql
+    )
+    assert "DO NOTHING" in sql
+    assert parameters == (
+        str(source_document_uuid(master)),
+        str(ingestion_run_uuid(master)),
+        source_family_postgresql_value(master.source_family),
+        master.source_document_id,
+        "checksum-unavailable",
+        "downloaded",
+    )
+
+
+def test_postgresql_source_family_ensure_source_document_uses_artifact_values() -> None:
+    connection = _FakeConnection()
+    command = build_parsed_factor_persistence_command(_payload("defra_desnz"))
+    master = command.master_records[0]
+
+    ensure_source_document(connection, master)
+
+    _statement, parameters = connection.statements[0]
+    assert isinstance(parameters, tuple)
+    assert parameters[3] == master.artifact_reference
+    assert parameters[4] == master.artifact_checksum_sha256
+    assert parameters[2] == source_family_postgresql_value(master.source_family)
+
+
 @pytest.mark.parametrize(
     ("source_family", "master_table", "master_id"),
     (