Skip to content

Split configured cycle summary and history helpers #649

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ktalpay merged 1 commit into from
Jun 2, 2026
Merged

Split configured cycle summary and history helpers #649

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/carbonfactor_parser/persistence/ingestion_run_history_mapping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
ParserIngestionSourceResultRecord,
)
if TYPE_CHECKING:
from carbonfactor_parser.pipeline.configured_cycle_runner import ConfiguredCycleResult
from carbonfactor_parser.pipeline.configured_cycle_models import ConfiguredCycleResult


def build_ingestion_run_history_command_from_configured_cycle(
Expand Down
87 changes: 87 additions & 0 deletions src/carbonfactor_parser/pipeline/configured_cycle_history.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""Configured ingestion cycle run-history persistence helpers."""

from __future__ import annotations

from datetime import datetime
from typing import Callable

from carbonfactor_parser.diagnostics.redaction import redact_sensitive_text
from carbonfactor_parser.persistence.ingestion_run_history import (
ParserIngestionRunHistoryRepository,
ParserIngestionRunHistoryStatus,
)
from carbonfactor_parser.persistence.ingestion_run_history_mapping import (
build_ingestion_run_history_command_from_configured_cycle,
)
from carbonfactor_parser.pipeline.configured_cycle_models import ConfiguredCycleResult


def persist_configured_cycle_history(
cycle: ConfiguredCycleResult,
*,
history_repository: ParserIngestionRunHistoryRepository,
started_at: datetime,
finished_at: datetime,
emit: Callable[[str], None] | None,
) -> ConfiguredCycleResult:
"""Persist run-history for a configured cycle without affecting ingestion result."""

command = build_ingestion_run_history_command_from_configured_cycle(
cycle,
started_at=started_at,
finished_at=finished_at,
)
try:
persist_result = history_repository.persist_ingestion_run_history(command)
except Exception as exc: # pragma: no cover - defensive boundary protection
safe_message = redact_sensitive_text(str(exc))
if emit is not None:
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
"issue code=INGESTION_RUN_HISTORY_PERSISTENCE_EXCEPTION "
f"message={safe_message}"
)
return ConfiguredCycleResult(
cycle_number=cycle.cycle_number,
run_id=cycle.run_id,
result=cycle.result,
history_persistence_status="failed",
history_persistence_issue_count=1,
)

issue_count = len(persist_result.issues)
if persist_result.status is ParserIngestionRunHistoryStatus.DECLARED:
if emit is not None:
emit(f"history_persistence status=declared run_id={cycle.run_id}")
else:
if emit is not None:
for issue in persist_result.issues or ():
safe_message = redact_sensitive_text(str(issue.message))
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
f"issue code={issue.code} message={safe_message}"
)
if not persist_result.issues:
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
"issue code=INGESTION_RUN_HISTORY_PERSISTENCE_FAILED "
"message=run history persistence failed"
)
issue_count = 1
return ConfiguredCycleResult(
cycle_number=cycle.cycle_number,
run_id=cycle.run_id,
result=cycle.result,
history_persistence_status=(
"declared"
if persist_result.status is ParserIngestionRunHistoryStatus.DECLARED
else "failed"
),
history_persistence_issue_count=issue_count,
)


__all__ = ("persist_configured_cycle_history",)
36 changes: 36 additions & 0 deletions src/carbonfactor_parser/pipeline/configured_cycle_models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""Configured ingestion cycle result models."""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

from carbonfactor_parser.pipeline.production_e2e_year_orchestrator import (
ProductionE2EYearOrchestratorResult,
)

if TYPE_CHECKING:
from carbonfactor_parser.pipeline.configured_cycle_runner import (
ConfiguredCycleRunnerStatus,
)


@dataclass(frozen=True)
class ConfiguredCycleResult:
"""One completed application cycle."""

cycle_number: int
run_id: str
result: ProductionE2EYearOrchestratorResult
history_persistence_status: str | None = None
history_persistence_issue_count: int = 0


@dataclass(frozen=True)
class ConfiguredCycleRunnerResult:
"""All cycles run by one application invocation."""

status: ConfiguredCycleRunnerStatus

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep runner status resolvable at runtime

When ConfiguredCycleRunnerResult is imported from the new models module, the status annotation is stored as the string ConfiguredCycleRunnerStatus, but that name only exists under TYPE_CHECKING. Runtime consumers that introspect this public result model with typing.get_type_hints(ConfiguredCycleRunnerResult) now get NameError: name 'ConfiguredCycleRunnerStatus' is not defined; this worked before because the enum and dataclass lived in the same module. This affects serializers/schema builders or other API tooling that resolves annotations, so the enum should be moved/re-exported into a runtime-resolvable namespace instead of being only type-checking imported.

Useful? React with 👍 / 👎.

cycles: tuple[ConfiguredCycleResult, ...]
schema_created_table_names: tuple[str, ...]
schema_missing_table_names: tuple[str, ...]
172 changes: 11 additions & 161 deletions src/carbonfactor_parser/pipeline/configured_cycle_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
import time
Expand All @@ -18,10 +17,6 @@

from carbonfactor_parser.persistence.ingestion_run_history import (
ParserIngestionRunHistoryRepository,
ParserIngestionRunHistoryStatus,
)
from carbonfactor_parser.persistence.ingestion_run_history_mapping import (
build_ingestion_run_history_command_from_configured_cycle,
)
from carbonfactor_parser.persistence.postgresql_ingestion_run_history_repository import (
PostgreSQLIngestionRunHistoryRepository,
Expand All @@ -40,9 +35,18 @@
ConfiguredCycleValidationBoundary,
build_configured_cycle_dependencies,
)
from carbonfactor_parser.pipeline.configured_cycle_history import (
persist_configured_cycle_history,
)
from carbonfactor_parser.pipeline.configured_cycle_models import (
ConfiguredCycleResult,
ConfiguredCycleRunnerResult,
)
from carbonfactor_parser.pipeline.configured_cycle_summary import (
emit_configured_cycle_summary,
)
from carbonfactor_parser.pipeline.production_e2e_year_orchestrator import (
ProductionE2EYearOrchestratorRequest,
ProductionE2EYearOrchestratorResult,
ProductionE2EYearRunStatus,
run_production_e2e_year_orchestrator,
)
Expand All @@ -55,27 +59,6 @@ class ConfiguredCycleRunnerStatus(str, Enum):
COMPLETED_WITH_FAILURES = "completed_with_failures"


@dataclass(frozen=True)
class ConfiguredCycleResult:
"""One completed application cycle."""

cycle_number: int
run_id: str
result: ProductionE2EYearOrchestratorResult
history_persistence_status: str | None = None
history_persistence_issue_count: int = 0


@dataclass(frozen=True)
class ConfiguredCycleRunnerResult:
"""All cycles run by one application invocation."""

status: ConfiguredCycleRunnerStatus
cycles: tuple[ConfiguredCycleResult, ...]
schema_created_table_names: tuple[str, ...]
schema_missing_table_names: tuple[str, ...]


def run_configured_cycle_runner(
config: ConfiguredCycleRunnerConfig,
*,
Expand Down Expand Up @@ -121,7 +104,7 @@ def run_configured_cycle_runner(
)
if emit is not None:
emit_configured_cycle_summary(cycle, emit=emit)
cycle = _persist_configured_cycle_history(
cycle = persist_configured_cycle_history(
cycle,
history_repository=history_repository,
started_at=started_at,
Expand Down Expand Up @@ -152,118 +135,6 @@ def run_configured_cycle_runner(
)


def _persist_configured_cycle_history(
cycle: ConfiguredCycleResult,
*,
history_repository: ParserIngestionRunHistoryRepository,
started_at: datetime,
finished_at: datetime,
emit: Callable[[str], None] | None,
) -> ConfiguredCycleResult:
command = build_ingestion_run_history_command_from_configured_cycle(
cycle,
started_at=started_at,
finished_at=finished_at,
)
try:
persist_result = history_repository.persist_ingestion_run_history(command)
except Exception as exc: # pragma: no cover - defensive boundary protection
safe_message = _redact_sensitive_text(str(exc))
if emit is not None:
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
"issue code=INGESTION_RUN_HISTORY_PERSISTENCE_EXCEPTION "
f"message={safe_message}"
)
return ConfiguredCycleResult(
cycle_number=cycle.cycle_number,
run_id=cycle.run_id,
result=cycle.result,
history_persistence_status="failed",
history_persistence_issue_count=1,
)

issue_count = len(persist_result.issues)
if persist_result.status is ParserIngestionRunHistoryStatus.DECLARED:
if emit is not None:
emit(f"history_persistence status=declared run_id={cycle.run_id}")
else:
if emit is not None:
for issue in persist_result.issues or ():
safe_message = _redact_sensitive_text(str(issue.message))
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
f"issue code={issue.code} message={safe_message}"
)
if not persist_result.issues:
emit(
"history_persistence "
f"status=failed run_id={cycle.run_id} "
"issue code=INGESTION_RUN_HISTORY_PERSISTENCE_FAILED "
"message=run history persistence failed"
)
issue_count = 1
return ConfiguredCycleResult(
cycle_number=cycle.cycle_number,
run_id=cycle.run_id,
result=cycle.result,
history_persistence_status=(
"declared"
if persist_result.status is ParserIngestionRunHistoryStatus.DECLARED
else "failed"
),
history_persistence_issue_count=issue_count,
)


def emit_configured_cycle_summary(
cycle: ConfiguredCycleResult,
*,
emit: Callable[[str], None] = print,
) -> None:
"""Print user-readable summary output for one cycle."""

summary = cycle.result.summary
emit(
"cycle="
f"{cycle.cycle_number} run_id={cycle.run_id} status={cycle.result.status.value}"
)
emit(
"summary "
f"completed={summary.completed_family_count} "
f"no_available_source_year={summary.no_available_source_year_count} "
f"failed={summary.failed_family_count} "
f"parsed_rows={summary.parsed_row_count} "
f"inserted={summary.inserted_count} "
f"skipped_duplicates={summary.skipped_duplicate_count}"
)
for family in cycle.result.family_results:
insert_summary = family.insert_summary
emit(
"source "
f"family={family.source_family} "
f"target_year={family.year_state.target_year} "
f"latest_year={family.year_state.latest_year} "
f"status={family.status.value} "
f"download_status={_download_status_value(family.download_result)} "
f"parse_status={_parse_status_value(family)} "
f"parsed_rows={family.parsed_row_count} "
f"master_inserted={getattr(insert_summary, 'master_inserted', 0)} "
f"master_skipped={getattr(insert_summary, 'master_skipped', 0)} "
f"detail_inserted={getattr(insert_summary, 'detail_inserted', 0)} "
f"detail_skipped={getattr(insert_summary, 'detail_skipped', 0)}"
)
for failure in family.failures:
safe_message = _redact_sensitive_text(str(failure.message))
emit(
"issue "
f"family={failure.source_family} stage={failure.stage} "
f"code={failure.code} message={safe_message}"
)


def _emit_startup_summary(
config: ConfiguredCycleRunnerConfig,
runtime: PostgreSQLRuntimeStartupResult,
Expand All @@ -287,27 +158,6 @@ def _redact_sensitive_text(text: str) -> str:
return redact_sensitive_text(text)


def _download_status_value(download_result: object | None) -> str:
if download_result is None:
return "not_run"
return str(getattr(getattr(download_result, "status", None), "value", "unknown"))


def _parse_status_value(family: object) -> str:
if getattr(family, "parsed_row_count", 0) > 0:
return "parsed"
failures = tuple(getattr(family, "failures", ()))
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"
):
return "not_run"
return "no_rows"


__all__ = (
"CONFIGURED_CYCLE_SOURCE_FAMILIES",
"ConfiguredCycleResult",
Expand Down
Loading
Loading