"""Application-owned Modelo 100 borrador snapshot lifecycle.
Borrador100 is the proof-of-concept consumer of the shared snapshot lifecycle
abstraction. The application owns the payload, identity, state machine, and
repository capability; executable composition supplies the encrypted persistence
implementation through :class:`Borrador100SnapshotRepository`.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime
from decimal import Decimal
from typing import Annotated, Protocol, override
from pydantic import BaseModel, Field, StringConstraints, model_validator
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.filing_year import FilingYear
from ...core.hashing import content_hash_hex
from ...core.identity.bucket import BucketId
from ...core.identity.hex_ids import SnapshotId
from ...core.modelo import Modelo
from ...core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core.period import Period
from ...domain.calculations.registry.authority import PinnedAuthorityOperation, bundled_indexed_authority
from ...domain.calculations.registry.errors import RegistrySnapshotError
from ...domain.calculations.registry.ids import BindingId
from ...domain.calculations.registry.schema_references import RegistrySnapshotRef
from ..calculations.revision_carry_gate import revision_carry_outcome
from .errors import LiveApplicationInputError
from .snapshot_base import (
SnapshotLifecycleState,
SnapshotNotFoundError,
SnapshotRepository,
SnapshotService,
enforce_snapshot_state_invariants,
)
type _BorradorValue = Decimal | str
BorradorSourceUrl = Annotated[str, StringConstraints(min_length=1, max_length=2048)]
"""The sede address a borrador snapshot was captured from."""
[docs]
class BorradorSnapshotNotFoundError(SnapshotNotFoundError):
"""Raised when a Modelo 100 borrador snapshot lookup misses by id.
:class:`SnapshotNotFoundError` inherits ``CadrumoError`` first, so MRO
routes ``__init__`` through the structured constructor (accepts
``context=`` / ``translated_message=`` kwargs) rather than
:class:`KeyError`'s C-level constructor. Listing ``CadrumoError``
explicitly here would violate C3 linearization.
"""
[docs]
class Borrador100Snapshot(BaseModel):
"""Captured Modelo 100 borrador values available to application consumers."""
model_config = _STRICT_FROZEN
snapshot_id: SnapshotId
bucket_id: BucketId
modelo: str = Field(pattern=f"^{Modelo('100').value}$")
filing_year: FilingYear
period: Period
registry_snapshot_ref: RegistrySnapshotRef
captured_at: datetime
source_url: BorradorSourceUrl
state: SnapshotLifecycleState
binding_values: Mapping[BindingId, _BorradorValue] = Field(default_factory=dict)
superseded_by_snapshot_id: SnapshotId | None = None
discarded_at: datetime | None = None
discarded_by: str = Field(default="", max_length=128)
discard_reason: str = Field(default="", max_length=500)
@model_validator(mode="after")
@pydantic_validation_boundary
def _enforce_state_payload(self) -> Borrador100Snapshot:
expected_ref = RegistrySnapshotRef(
modelo=self.modelo,
revision_id=self.registry_snapshot_ref.revision_id,
modelo_year=self.filing_year,
period=self.period.registry_token,
)
if self.registry_snapshot_ref != expected_ref:
raise ValueError("borrador registry_snapshot_ref must match modelo, filing_year, and period")
enforce_snapshot_state_invariants(
state=self.state,
has_supersession_pointer=self.superseded_by_snapshot_id is not None,
discarded_at=self.discarded_at,
discarded_by=self.discarded_by,
discard_reason=self.discard_reason,
)
blank_keys = sorted(key for key in self.binding_values if not key.strip())
if blank_keys:
raise LiveApplicationInputError(
translated_message="application.live.borrador.errors.binding_value_key_blank",
context={"blank_key_count": len(blank_keys)},
)
return self
[docs]
def require_borrador_100_snapshot_coordinates_current(
snapshot: Borrador100Snapshot,
*,
operation: PinnedAuthorityOperation,
) -> Borrador100Snapshot:
"""Return a persisted borrador only when its producing coordinate re-confirms."""
outcome = revision_carry_outcome(snapshot.registry_snapshot_ref, operation=operation)
if outcome.refused:
raise RegistrySnapshotError(
"borrador registry coordinate cannot be re-confirmed: "
f"{snapshot.registry_snapshot_ref.revision_id}: {outcome.detail}"
)
return snapshot
[docs]
def borrador_100_snapshot_object_key(bucket_id: str, snapshot_id: str) -> str:
"""Return the secure-object key for one bucket's Modelo 100 borrador snapshot."""
trimmed_bucket = bucket_id.strip()
trimmed_snapshot = snapshot_id.strip()
if not trimmed_bucket:
raise LiveApplicationInputError(
translated_message="application.live.borrador.errors.bucket_id_blank",
)
if not trimmed_snapshot:
raise LiveApplicationInputError(
translated_message="application.live.borrador.errors.snapshot_id_blank",
)
return f"modelo-100-borrador-snapshot:{trimmed_bucket}:{trimmed_snapshot}"
[docs]
def derive_borrador_100_snapshot_id(
*,
filing_year: int,
period: Period,
registry_snapshot_ref: RegistrySnapshotRef,
captured_at: datetime,
source_url: str,
binding_values: Mapping[BindingId, _BorradorValue],
) -> str:
"""Return the content-addressed id for one Modelo 100 borrador capture.
The canonical-JSON shape is preserved exactly so existing on-disk
snapshot ids remain valid: the core canonical JSON hash does not change
the hashed bytes.
"""
if (
registry_snapshot_ref.modelo != Modelo("100").value
or registry_snapshot_ref.modelo_year != filing_year
or registry_snapshot_ref.period != period.registry_token
):
raise ValueError("borrador snapshot identity coordinate must match registry_snapshot_ref")
return content_hash_hex(
{
"modelo": Modelo("100").value,
"filing_year": filing_year,
"period": period.registry_token,
"registry_snapshot_ref": registry_snapshot_ref.model_dump(mode="json"),
"captured_at": captured_at.isoformat(),
"source_url": source_url,
"binding_values": {
key: format(value, "f") if isinstance(value, Decimal) else value
for key, value in sorted(binding_values.items())
},
},
)
[docs]
class Borrador100SnapshotRepository(SnapshotRepository[Borrador100Snapshot], Protocol):
"""Application persistence capability for bucket-scoped borrador snapshots.
Encrypted secure-object storage, namespaces, object keys, and persistence
errors belong to the outer adapter. Implementations satisfy this structural
contract without inheriting application code.
"""
[docs]
class Borrador100SnapshotRepositoryFactory(Protocol):
"""Construct the required borrador repository for one profile bucket."""
def __call__(self, *, bucket_id: str) -> Borrador100SnapshotRepository:
"""Return the application repository capability for ``bucket_id``."""
...
class _Borrador100CaptureRequest(BaseModel):
model_config = _STRICT_FROZEN
filing_year: int
period: Period
registry_snapshot_ref: RegistrySnapshotRef
captured_at: datetime
source_url: str
binding_values: Mapping[BindingId, _BorradorValue]
[docs]
class Borrador100SnapshotService(SnapshotService[Borrador100Snapshot, _Borrador100CaptureRequest]):
"""Canonical backend service for bucket-scoped Modelo 100 borrador snapshots."""
def __init__(
self,
*,
bucket_id: str,
repository: Borrador100SnapshotRepository,
) -> None:
"""Initialize this public contract."""
super().__init__(bucket_id=bucket_id, repository=repository)
[docs]
def capture(
self,
*,
filing_year: int,
period: Period,
captured_at: datetime,
source_url: str,
binding_values: Mapping[BindingId, _BorradorValue],
) -> Borrador100Snapshot:
"""Execute this public contract operation."""
with bundled_indexed_authority().operation() as operation:
snapshot_ref = operation.snapshot(
Modelo("100").value,
filing_year=filing_year,
period=period.registry_token,
).snapshot_ref
return self._capture_with_lifecycle(
_Borrador100CaptureRequest(
filing_year=filing_year,
period=period,
registry_snapshot_ref=snapshot_ref,
captured_at=captured_at,
source_url=source_url,
binding_values=binding_values,
),
)
[docs]
@override
# TYPE-IGNORE-RATIONALE-OVERRIDE-COVARIANT-RETURN:
# Subclass returns a narrower snapshot type and adds optional filter params;
# base-class signature widening would ripple to N subclasses.
def list_snapshots(
self,
*,
filing_year: int | None = None,
state: SnapshotLifecycleState | None = SnapshotLifecycleState.ACTIVE,
) -> tuple[Borrador100Snapshot, ...]:
with bundled_indexed_authority().operation() as operation:
return self._list_snapshots_current(
operation=operation,
filing_year=filing_year,
state=state,
)
def _list_snapshots_current(
self,
*,
operation: PinnedAuthorityOperation,
filing_year: int | None = None,
state: SnapshotLifecycleState | None = SnapshotLifecycleState.ACTIVE,
) -> tuple[Borrador100Snapshot, ...]:
snapshots = tuple(
require_borrador_100_snapshot_coordinates_current(snapshot, operation=operation)
for snapshot in super().list_snapshots()
)
if filing_year is not None:
snapshots = tuple(snapshot for snapshot in snapshots if snapshot.filing_year == filing_year)
if state is not None:
snapshots = tuple(snapshot for snapshot in snapshots if snapshot.state is state)
return snapshots
[docs]
def show(self, snapshot_id: str, *, operation: PinnedAuthorityOperation) -> Borrador100Snapshot:
"""Execute this public contract operation."""
return require_borrador_100_snapshot_coordinates_current(
self.resolve_snapshot(snapshot_id), operation=operation
)
[docs]
def latest_for_year(
self,
*,
filing_year: int,
operation: PinnedAuthorityOperation,
period: Period | None = None,
) -> Borrador100Snapshot | None:
"""Execute this public contract operation."""
snapshots = [
snapshot
for snapshot in self._list_snapshots_current(filing_year=filing_year, operation=operation)
if period is None or snapshot.period == period
]
if not snapshots:
return None
return max(snapshots, key=lambda snapshot: snapshot.captured_at)
# ---- SnapshotService[Borrador100Snapshot] hooks ----------------------
@override
def _derive_snapshot_id(self, capture: _Borrador100CaptureRequest) -> str:
return derive_borrador_100_snapshot_id(
filing_year=capture.filing_year,
period=capture.period,
registry_snapshot_ref=capture.registry_snapshot_ref,
captured_at=capture.captured_at,
source_url=capture.source_url,
binding_values=capture.binding_values,
)
@override
def _build_active_payload(
self,
*,
snapshot_id: str,
capture: _Borrador100CaptureRequest,
) -> Borrador100Snapshot:
return Borrador100Snapshot(
snapshot_id=snapshot_id,
bucket_id=self._repository.bucket_id,
modelo=Modelo("100").value,
filing_year=capture.filing_year,
period=capture.period,
registry_snapshot_ref=capture.registry_snapshot_ref,
captured_at=capture.captured_at,
source_url=capture.source_url,
state=SnapshotLifecycleState.ACTIVE,
binding_values=dict(capture.binding_values),
)
@override
def _payload_axis_key(self, payload: Borrador100Snapshot) -> tuple[object, ...]:
return (payload.modelo, payload.filing_year, payload.period)
@override
def _payload_captured_at(self, payload: Borrador100Snapshot) -> datetime:
return payload.captured_at
@override
def _payload_snapshot_id(self, payload: Borrador100Snapshot) -> str:
return payload.snapshot_id
@override
def _payload_state(self, payload: Borrador100Snapshot) -> SnapshotLifecycleState:
return payload.state
@override
def _demote_to_superseded(self, payload: Borrador100Snapshot, *, superseded_by: str) -> Borrador100Snapshot:
return payload.model_copy(
update={
"state": SnapshotLifecycleState.SUPERSEDED,
"superseded_by_snapshot_id": superseded_by,
},
)
__all__ = [
"Borrador100Snapshot",
"Borrador100SnapshotRepository",
"Borrador100SnapshotRepositoryFactory",
"Borrador100SnapshotService",
"BorradorSnapshotNotFoundError",
"borrador_100_snapshot_object_key",
"derive_borrador_100_snapshot_id",
"require_borrador_100_snapshot_coordinates_current",
]