diff --git a/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py b/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py index 349e8b6..f26e3f4 100644 --- a/src/carbonfactor_parser/persistence/postgresql_source_family_repository.py +++ b/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, diff --git a/src/carbonfactor_parser/persistence/postgresql_source_family_upserts.py b/src/carbonfactor_parser/persistence/postgresql_source_family_upserts.py new file mode 100644 index 0000000..1661943 --- /dev/null +++ b/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", +) diff --git a/tests/test_postgresql_source_family_repository.py b/tests/test_postgresql_source_family_repository.py index e637a42..92c5c8b 100644 --- a/tests/test_postgresql_source_family_repository.py +++ b/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"), (