From da90954b8c9697df79cc90db4d6a642159ed1c7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?K=C3=BCr=C5=9Fat=20Alpay?= Date: Tue, 2 Jun 2026 15:39:49 +0300 Subject: [PATCH] OPS-REF-001B split configured cycle dependencies --- .../pipeline/configured_cycle_dependencies.py | 170 ++++++++++++++++++ .../pipeline/configured_cycle_runner.py | 155 ++-------------- tests/test_configured_cycle_runner.py | 98 ++++++++++ 3 files changed, 278 insertions(+), 145 deletions(-) create mode 100644 src/carbonfactor_parser/pipeline/configured_cycle_dependencies.py diff --git a/src/carbonfactor_parser/pipeline/configured_cycle_dependencies.py b/src/carbonfactor_parser/pipeline/configured_cycle_dependencies.py new file mode 100644 index 0000000..e7c6de7 --- /dev/null +++ b/src/carbonfactor_parser/pipeline/configured_cycle_dependencies.py @@ -0,0 +1,170 @@ +"""Configured cycle dependency composition. + +This module wires configured-cycle runtime dependencies for the production +E2E year orchestrator without owning the runner loop, startup summary, cycle +summary output, or run-history persistence behavior. +""" + +from __future__ import annotations + +from typing import Mapping + +from carbonfactor_parser.persistence.postgresql_runtime import ( + PostgreSQLRuntimeStartupResult, +) +from carbonfactor_parser.persistence.postgresql_source_family_repository import ( + PostgreSQLSourceFamilyRuntimeRepository, +) +from carbonfactor_parser.pipeline.configured_cycle_config import ( + ConfiguredCycleRunnerConfig, + ConfiguredSourceYearArtifact, +) +from carbonfactor_parser.pipeline.defra_desnz_production_e2e import ( + DEFRA_DESNZ_SOURCE_FAMILY, + DefraDesnzPhase2ValidationBoundary, + DefraDesnzProductionParserBoundary, + DefraDesnzProductionSourceAdapter, + DefraDesnzSourceYear, +) +from carbonfactor_parser.pipeline.ghg_protocol_production_e2e import ( + GHG_PROTOCOL_SOURCE_FAMILY, + GHGProtocolPhase2ValidationBoundary, + GHGProtocolProductionParserBoundary, + GHGProtocolProductionSourceAdapter, + GHGProtocolSourceYear, +) +from carbonfactor_parser.pipeline.ipcc_efdb_production_e2e import ( + IPCC_EFDB_SOURCE_FAMILY, + IpccEfdbPhase2ValidationBoundary, + IpccEfdbProductionParserBoundary, + IpccEfdbProductionSourceAdapter, + IpccEfdbSourceYear, +) +from carbonfactor_parser.pipeline.production_e2e_year_orchestrator import ( + ProductionE2EValidationResult, + ProductionE2EYearOrchestratorDependencies, +) +from carbonfactor_parser.pipeline.source_artifact_transport import ( + build_configured_artifact_transport, +) + + +class ConfiguredCycleValidationBoundary: + """Route validation to the source-family-specific validation boundary.""" + + def __init__(self) -> None: + self._boundaries = { + GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolPhase2ValidationBoundary(), + DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzPhase2ValidationBoundary(), + IPCC_EFDB_SOURCE_FAMILY: IpccEfdbPhase2ValidationBoundary(), + } + + def validate(self, batch: object) -> ProductionE2EValidationResult: + rows = tuple(getattr(batch, "rows", ())) + source_family = rows[0].source_family if rows else GHG_PROTOCOL_SOURCE_FAMILY + boundary = self._boundaries.get(source_family) + if boundary is None: + boundary = GHGProtocolPhase2ValidationBoundary() + return boundary.validate(batch) + + +def build_configured_cycle_dependencies( + config: ConfiguredCycleRunnerConfig, + runtime: PostgreSQLRuntimeStartupResult, +) -> ProductionE2EYearOrchestratorDependencies: + """Build production E2E dependencies for the configured cycle runner.""" + + transport = build_configured_artifact_transport( + allow_live_source_access=config.allow_live_source_access, + ) + source_years = config.source_years or {} + return ProductionE2EYearOrchestratorDependencies( + year_state_repository=runtime.year_state_repository, + source_adapters={ + GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolProductionSourceAdapter( + target_root=config.archive_root, + source_years=_ghg_source_years( + source_years.get(GHG_PROTOCOL_SOURCE_FAMILY, {}), + ), + transport=transport, + ), + DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzProductionSourceAdapter( + target_root=config.archive_root, + source_years=_defra_source_years( + source_years.get(DEFRA_DESNZ_SOURCE_FAMILY, {}), + ), + transport=transport, + ), + IPCC_EFDB_SOURCE_FAMILY: IpccEfdbProductionSourceAdapter( + target_root=config.archive_root, + source_years=_ipcc_source_years( + source_years.get(IPCC_EFDB_SOURCE_FAMILY, {}), + ), + transport=transport, + ), + }, + parser_boundaries={ + GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolProductionParserBoundary(), + DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzProductionParserBoundary(), + IPCC_EFDB_SOURCE_FAMILY: IpccEfdbProductionParserBoundary(), + }, + validation_boundary=ConfiguredCycleValidationBoundary(), + insert_repository=PostgreSQLSourceFamilyRuntimeRepository(runtime.connection), + ) + + +def _ghg_source_years( + values: Mapping[int, ConfiguredSourceYearArtifact], +) -> Mapping[int, GHGProtocolSourceYear]: + return { + year: GHGProtocolSourceYear( + year=entry.year, + publication_url=entry.publication_url, + artifact_url=entry.artifact_url, + title=entry.title, + version_label=entry.version_label, + content_type=entry.content_type, + format_hint=entry.format_hint, + ) + for year, entry in values.items() + } + + +def _defra_source_years( + values: Mapping[int, ConfiguredSourceYearArtifact], +) -> Mapping[int, DefraDesnzSourceYear]: + return { + year: DefraDesnzSourceYear( + year=entry.year, + publication_url=entry.publication_url, + artifact_url=entry.artifact_url, + title=entry.title, + version_label=entry.version_label, + content_type=entry.content_type, + format_hint=entry.format_hint, + ) + for year, entry in values.items() + } + + +def _ipcc_source_years( + values: Mapping[int, ConfiguredSourceYearArtifact], +) -> Mapping[int, IpccEfdbSourceYear]: + return { + year: IpccEfdbSourceYear( + year=entry.year, + publication_url=entry.publication_url, + artifact_url=entry.artifact_url, + title=entry.title, + version_label=entry.version_label, + content_type=entry.content_type, + format_hint=entry.format_hint, + ) + for year, entry in values.items() + } + + +__all__ = ( + "ConfiguredCycleValidationBoundary", + "build_configured_cycle_dependencies", +) diff --git a/src/carbonfactor_parser/pipeline/configured_cycle_runner.py b/src/carbonfactor_parser/pipeline/configured_cycle_runner.py index 4ba9c80..ad231f5 100644 --- a/src/carbonfactor_parser/pipeline/configured_cycle_runner.py +++ b/src/carbonfactor_parser/pipeline/configured_cycle_runner.py @@ -12,7 +12,7 @@ from enum import Enum import time import uuid -from typing import Callable, Mapping +from typing import Callable from carbonfactor_parser.diagnostics.redaction import redact_sensitive_text @@ -30,47 +30,22 @@ PostgreSQLRuntimeStartupResult, start_postgresql_runtime, ) -from carbonfactor_parser.persistence.postgresql_source_family_repository import ( - PostgreSQLSourceFamilyRuntimeRepository, -) from carbonfactor_parser.pipeline.configured_cycle_config import ( CONFIGURED_CYCLE_SOURCE_FAMILIES, ConfiguredCycleRunnerConfig, ConfiguredSourceYearArtifact, load_configured_cycle_runner_config, ) -from carbonfactor_parser.pipeline.defra_desnz_production_e2e import ( - DEFRA_DESNZ_SOURCE_FAMILY, - DefraDesnzPhase2ValidationBoundary, - DefraDesnzProductionParserBoundary, - DefraDesnzProductionSourceAdapter, - DefraDesnzSourceYear, -) -from carbonfactor_parser.pipeline.ghg_protocol_production_e2e import ( - GHG_PROTOCOL_SOURCE_FAMILY, - GHGProtocolPhase2ValidationBoundary, - GHGProtocolProductionParserBoundary, - GHGProtocolProductionSourceAdapter, - GHGProtocolSourceYear, -) -from carbonfactor_parser.pipeline.ipcc_efdb_production_e2e import ( - IPCC_EFDB_SOURCE_FAMILY, - IpccEfdbPhase2ValidationBoundary, - IpccEfdbProductionParserBoundary, - IpccEfdbProductionSourceAdapter, - IpccEfdbSourceYear, +from carbonfactor_parser.pipeline.configured_cycle_dependencies import ( + ConfiguredCycleValidationBoundary, + build_configured_cycle_dependencies, ) from carbonfactor_parser.pipeline.production_e2e_year_orchestrator import ( - ProductionE2EValidationResult, - ProductionE2EYearOrchestratorDependencies, ProductionE2EYearOrchestratorRequest, ProductionE2EYearOrchestratorResult, ProductionE2EYearRunStatus, run_production_e2e_year_orchestrator, ) -from carbonfactor_parser.pipeline.source_artifact_transport import ( - build_configured_artifact_transport, -) class ConfiguredCycleRunnerStatus(str, Enum): @@ -101,25 +76,6 @@ class ConfiguredCycleRunnerResult: schema_missing_table_names: tuple[str, ...] -class ConfiguredCycleValidationBoundary: - """Route validation to the source-family-specific validation boundary.""" - - def __init__(self) -> None: - self._boundaries = { - GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolPhase2ValidationBoundary(), - DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzPhase2ValidationBoundary(), - IPCC_EFDB_SOURCE_FAMILY: IpccEfdbPhase2ValidationBoundary(), - } - - def validate(self, batch: object) -> ProductionE2EValidationResult: - rows = tuple(getattr(batch, "rows", ())) - source_family = rows[0].source_family if rows else GHG_PROTOCOL_SOURCE_FAMILY - boundary = self._boundaries.get(source_family) - if boundary is None: - boundary = GHGProtocolPhase2ValidationBoundary() - return boundary.validate(batch) - - def run_configured_cycle_runner( config: ConfiguredCycleRunnerConfig, *, @@ -137,7 +93,7 @@ def run_configured_cycle_runner( if emit is not None: _emit_startup_summary(config, runtime, emit) - dependencies = _build_dependencies(config, runtime) + dependencies = build_configured_cycle_dependencies(config, runtime) history_repository = run_history_repository if history_repository is None: history_repository_factory = ( @@ -196,7 +152,6 @@ def run_configured_cycle_runner( ) - def _persist_configured_cycle_history( cycle: ConfiguredCycleResult, *, @@ -262,6 +217,7 @@ def _persist_configured_cycle_history( history_persistence_issue_count=issue_count, ) + def emit_configured_cycle_summary( cycle: ConfiguredCycleResult, *, @@ -308,49 +264,6 @@ def emit_configured_cycle_summary( ) -def _build_dependencies( - config: ConfiguredCycleRunnerConfig, - runtime: PostgreSQLRuntimeStartupResult, -) -> ProductionE2EYearOrchestratorDependencies: - transport = build_configured_artifact_transport( - allow_live_source_access=config.allow_live_source_access, - ) - source_years = config.source_years or {} - return ProductionE2EYearOrchestratorDependencies( - year_state_repository=runtime.year_state_repository, - source_adapters={ - GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolProductionSourceAdapter( - target_root=config.archive_root, - source_years=_ghg_source_years( - source_years.get(GHG_PROTOCOL_SOURCE_FAMILY, {}), - ), - transport=transport, - ), - DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzProductionSourceAdapter( - target_root=config.archive_root, - source_years=_defra_source_years( - source_years.get(DEFRA_DESNZ_SOURCE_FAMILY, {}), - ), - transport=transport, - ), - IPCC_EFDB_SOURCE_FAMILY: IpccEfdbProductionSourceAdapter( - target_root=config.archive_root, - source_years=_ipcc_source_years( - source_years.get(IPCC_EFDB_SOURCE_FAMILY, {}), - ), - transport=transport, - ), - }, - parser_boundaries={ - GHG_PROTOCOL_SOURCE_FAMILY: GHGProtocolProductionParserBoundary(), - DEFRA_DESNZ_SOURCE_FAMILY: DefraDesnzProductionParserBoundary(), - IPCC_EFDB_SOURCE_FAMILY: IpccEfdbProductionParserBoundary(), - }, - validation_boundary=ConfiguredCycleValidationBoundary(), - insert_repository=PostgreSQLSourceFamilyRuntimeRepository(runtime.connection), - ) - - def _emit_startup_summary( config: ConfiguredCycleRunnerConfig, runtime: PostgreSQLRuntimeStartupResult, @@ -374,57 +287,6 @@ def _redact_sensitive_text(text: str) -> str: return redact_sensitive_text(text) -def _ghg_source_years( - values: Mapping[int, ConfiguredSourceYearArtifact], -) -> Mapping[int, GHGProtocolSourceYear]: - return { - year: GHGProtocolSourceYear( - year=entry.year, - publication_url=entry.publication_url, - artifact_url=entry.artifact_url, - title=entry.title, - version_label=entry.version_label, - content_type=entry.content_type, - format_hint=entry.format_hint, - ) - for year, entry in values.items() - } - - -def _defra_source_years( - values: Mapping[int, ConfiguredSourceYearArtifact], -) -> Mapping[int, DefraDesnzSourceYear]: - return { - year: DefraDesnzSourceYear( - year=entry.year, - publication_url=entry.publication_url, - artifact_url=entry.artifact_url, - title=entry.title, - version_label=entry.version_label, - content_type=entry.content_type, - format_hint=entry.format_hint, - ) - for year, entry in values.items() - } - - -def _ipcc_source_years( - values: Mapping[int, ConfiguredSourceYearArtifact], -) -> Mapping[int, IpccEfdbSourceYear]: - return { - year: IpccEfdbSourceYear( - year=entry.year, - publication_url=entry.publication_url, - artifact_url=entry.artifact_url, - title=entry.title, - version_label=entry.version_label, - content_type=entry.content_type, - format_hint=entry.format_hint, - ) - for year, entry in values.items() - } - - def _download_status_value(download_result: object | None) -> str: if download_result is None: return "not_run" @@ -438,7 +300,10 @@ def _parse_status_value(family: object) -> str: if any(getattr(failure, "stage", "") == "parser" for failure in failures): return "failed" download_result = getattr(family, "download_result", None) - if download_result is None or _download_status_value(download_result) != "downloaded": + if ( + download_result is None + or _download_status_value(download_result) != "downloaded" + ): return "not_run" return "no_rows" diff --git a/tests/test_configured_cycle_runner.py b/tests/test_configured_cycle_runner.py index 2c110ee..51fffa6 100644 --- a/tests/test_configured_cycle_runner.py +++ b/tests/test_configured_cycle_runner.py @@ -17,19 +17,42 @@ from carbonfactor_parser.persistence.postgresql_runtime import ( PostgreSQLRuntimeStartupResult, ) +from carbonfactor_parser.persistence.postgresql_source_family_repository import ( + PostgreSQLSourceFamilyRuntimeRepository, +) from carbonfactor_parser.pipeline.configured_cycle_config import ( ConfiguredCycleRunnerConfig as ExtractedConfiguredCycleRunnerConfig, ConfiguredSourceYearArtifact as ExtractedConfiguredSourceYearArtifact, load_configured_cycle_runner_config as extracted_load_configured_cycle_runner_config, ) +from carbonfactor_parser.pipeline.configured_cycle_dependencies import ( + ConfiguredCycleValidationBoundary as ExtractedConfiguredCycleValidationBoundary, + build_configured_cycle_dependencies, +) from carbonfactor_parser.pipeline.configured_cycle_runner import ( ConfiguredCycleRunnerConfig, ConfiguredCycleRunnerStatus, + ConfiguredCycleValidationBoundary, ConfiguredSourceYearArtifact, load_configured_cycle_runner_config, run_configured_cycle_runner, ) from carbonfactor_parser.pipeline import source_artifact_transport +from carbonfactor_parser.pipeline.defra_desnz_production_e2e import ( + DefraDesnzProductionParserBoundary, + DefraDesnzProductionSourceAdapter, +) +from carbonfactor_parser.pipeline.ghg_protocol_production_e2e import ( + GHGProtocolProductionParserBoundary, + GHGProtocolProductionSourceAdapter, +) +from carbonfactor_parser.pipeline.ipcc_efdb_production_e2e import ( + IpccEfdbProductionParserBoundary, + IpccEfdbProductionSourceAdapter, +) +from carbonfactor_parser.pipeline.production_e2e_year_orchestrator import ( + ProductionE2EYearOrchestratorDependencies, +) from carbonfactor_parser.pipeline.source_artifact_transport import ( build_configured_artifact_transport, ) @@ -39,6 +62,81 @@ def test_configured_cycle_runner_imports_remain_backward_compatible() -> None: assert ConfiguredCycleRunnerConfig is ExtractedConfiguredCycleRunnerConfig assert ConfiguredSourceYearArtifact is ExtractedConfiguredSourceYearArtifact assert load_configured_cycle_runner_config is extracted_load_configured_cycle_runner_config + assert ConfiguredCycleValidationBoundary is ExtractedConfiguredCycleValidationBoundary + + +def test_build_configured_cycle_dependencies_wires_configured_runtime( + tmp_path: Path, +) -> None: + connection = _FakeConnection() + startup = _startup(connection) + config = ConfiguredCycleRunnerConfig( + postgresql_config_result=load_postgresql_runtime_config( + {"CARBONOPS_POSTGRESQL_DSN": "postgresql://user:pass@localhost/db"}, + ), + archive_root=tmp_path / "archive", + enabled_source_families=("ghg_protocol", "defra_desnz", "ipcc_efdb"), + initial_year=2024, + cycle_interval_seconds=0, + max_cycles=1, + source_years=_source_years(tmp_path), + allow_live_source_access=False, + ) + + dependencies = build_configured_cycle_dependencies(config, startup) + + assert isinstance(dependencies, ProductionE2EYearOrchestratorDependencies) + assert dependencies.year_state_repository is startup.year_state_repository + assert dependencies.source_adapters.keys() == { + "ghg_protocol", + "defra_desnz", + "ipcc_efdb", + } + assert isinstance( + dependencies.source_adapters["ghg_protocol"], + GHGProtocolProductionSourceAdapter, + ) + assert isinstance( + dependencies.source_adapters["defra_desnz"], + DefraDesnzProductionSourceAdapter, + ) + assert isinstance( + dependencies.source_adapters["ipcc_efdb"], + IpccEfdbProductionSourceAdapter, + ) + assert dependencies.parser_boundaries.keys() == { + "ghg_protocol", + "defra_desnz", + "ipcc_efdb", + } + assert isinstance( + dependencies.parser_boundaries["ghg_protocol"], + GHGProtocolProductionParserBoundary, + ) + assert isinstance( + dependencies.parser_boundaries["defra_desnz"], + DefraDesnzProductionParserBoundary, + ) + assert isinstance( + dependencies.parser_boundaries["ipcc_efdb"], + IpccEfdbProductionParserBoundary, + ) + assert isinstance( + dependencies.validation_boundary, + ExtractedConfiguredCycleValidationBoundary, + ) + assert isinstance( + dependencies.insert_repository, + PostgreSQLSourceFamilyRuntimeRepository, + ) + assert dependencies.insert_repository._connection is connection + assert ( + dependencies.source_adapters["ghg_protocol"]._source_years[2024].year == 2024 + ) + assert ( + dependencies.source_adapters["defra_desnz"]._source_years[2024].year == 2024 + ) + assert dependencies.source_adapters["ipcc_efdb"]._source_years[2024].year == 2024 def test_source_artifact_transport_reads_local_file_path(tmp_path: Path) -> None: