"""Encrypted SQL repository for the calculation-revision catalogue.
:class:`CalculationRevisionCatalogueRepository` persists and loads
:class:`~domain.modelos.CalculationRevision` records in a
:class:`~domain.modelos.CalculationRevisionCatalogue` via
:class:`~adapters.persistence.storage.SecureObjectRepository` at
``FINANCIAL`` :class:`~adapters.persistence.storage.SensitivityClass`.
Each catalogue is wrapped in
:class:`~adapters.persistence.storage.Envelope` before being written to
the encrypted BLOB per profile bucket.
This concrete repository is the persistence adapter behind the read-side
:class:`~domain.modelos.CalculationRevisionCatalogueRepositoryProtocol`. It
lives in the persistence adapter (not in :mod:`~domain.modelos`) because its
secure-object coupling is SQL/crypto-bound; the domain package owns only the
typed :class:`~domain.modelos.CalculationRevisionCatalogue` model, the pure
:func:`~domain.modelos.upsert_calculation_revision` mutator, and the
:class:`~domain.modelos.CalculationRevisionPersistenceError` boundary error.
The namespace/version constants are redeclared here as the persisted-envelope
contract; the strings are preserved to avoid orphaning persisted envelopes.
See Also:
:mod:`~adapters.persistence.profile._modelo_runtime`
Bucket-id resolution and runtime secure-object factory shared by modelo
persistence adapters.
:class:`~domain.modelos.CalculationRevisionCatalogue`
Domain catalogue payload encrypted by this repository.
:class:`~domain.modelos.CalculationRevisionCatalogueRepositoryProtocol`
Domain port this concrete persistence adapter implements.
:data:`~adapters.persistence.storage.MODELO_CALCULATION_REVISION_CATALOGUE_NAMESPACE`
Central namespace, sensitivity, schema-version, and singleton-key
contract for these secure objects.
:class:`~adapters.persistence.storage.SecureObjectRepository`
Runtime-created encrypted storage boundary used for load/save.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from ....core.external_constants import UTF_8_ENCODING
from ....core.logging import get_logger
from ....core.time import now
from ....domain.modelos import (
CalculationRevisionCatalogue,
CalculationRevisionPersistenceError,
assert_revision_snapshot_evidence_coverage,
raise_catalogue_integrity_error,
)
from ._modelo_runtime import resolve_modelo_repository_bucket_id, secure_objects_for_modelo_bucket
if TYPE_CHECKING: # pragma: no cover — import-cycle guard
from ..storage import SecureObjectRepository, SecureObjectWrite
_LOGGER = get_logger(__name__)
# namespace string preserved across rename to avoid orphaning persisted envelopes
_CALCULATION_NAMESPACE = "aeat.domain.modelos.calculation_revisions"
_CALCULATION_OBJECT_KEY = "catalogue"
_CALCULATION_CATALOGUE_VERSION = 1
_CALCULATION_PERSISTENCE_MESSAGE = "errors.fail.fail_modelo_calculation_revision_persistence"
[docs]
class CalculationRevisionCatalogueRepository:
"""Repository over encrypted SQL-backed calculation-revision catalogue storage.
:data:`~adapters.persistence.storage.MODELO_CALCULATION_REVISION_CATALOGUE_NAMESPACE`
is the central namespace, schema-version, sensitivity, and singleton-key
contract for the encrypted :class:`CalculationRevisionCatalogue` row. The
catalogue payload is wrapped in
:class:`~adapters.persistence.storage.Envelope` before
:class:`~adapters.persistence.storage.SecureObjectRepository`
persists it, and the same envelope can be returned as a
:class:`~adapters.persistence.storage.SecureObjectWrite` for atomic
co-writes. The class exposes the concrete load/save implementation behind
:class:`~domain.modelos.CalculationRevisionCatalogueRepositoryProtocol`.
"""
def __init__(self, *, bucket_id: str | None = None, objects: SecureObjectRepository | None = None) -> None:
"""Bind the repository to a bucket id and/or an explicit secure-object store."""
self._bucket_id = bucket_id.strip() if bucket_id is not None else None
if objects is not None:
self._objects = objects
return
self._bucket_id = resolve_modelo_repository_bucket_id(
bucket_id,
error_type=CalculationRevisionPersistenceError,
)
self._objects = secure_objects_for_modelo_bucket(self._bucket_id)
@property
def bucket_id(self) -> str | None:
"""Identifier of the per-profile storage bucket this repository reads and writes.
A modelo (an AEAT tax form or declaration) carries calculation revisions
per filing profile, and each profile owns its own encrypted bucket. This
property exposes the resolved bucket identifier, or ``None`` when the
repository was constructed against a caller-supplied
:class:`~adapters.persistence.storage.SecureObjectRepository`
rather than a resolved bucket.
Returns:
The trimmed bucket identifier, or ``None`` when no bucket was resolved.
"""
return self._bucket_id
[docs]
def exists(self) -> bool:
"""Report whether a calculation-revision catalogue has been persisted.
Checks the encrypted store for an object under this repository's namespace
and key without decrypting or validating it, so a ``True`` result attests
to presence only, not integrity.
Returns:
``True`` when a stored catalogue object exists, ``False`` otherwise.
"""
return self._objects.exists(_CALCULATION_NAMESPACE, _CALCULATION_OBJECT_KEY)
[docs]
def load(self) -> CalculationRevisionCatalogue:
"""Load and decrypt the persisted calculation-revision catalogue.
A calculation revision is a dated, computed version of a modelo's casilla
values (a casilla is a numbered box on an AEAT form); the catalogue is the
keyed collection of those revisions. The stored record is decrypted, its
:class:`~adapters.persistence.storage.Envelope` parsed, and its
sensitivity classification and schema version checked before the payload
is returned. When nothing has been persisted yet, an empty
:class:`CalculationRevisionCatalogue` is returned rather than raising.
Returns:
The persisted :class:`CalculationRevisionCatalogue`, or an empty one when no
record exists.
Raises:
:class:`~domain.modelos.CalculationRevisionPersistenceError`: If
the stored record fails the FINANCIAL classification check, or its
envelope schema version exceeds the version this consumer
supports, or an integrity error surfaces while decrypting and
decoding the record.
"""
from ..storage import (
ClassificationError,
Envelope,
EnvelopeVersionError,
SensitivityClass,
)
try:
record = self._objects.load(
_CALCULATION_NAMESPACE,
_CALCULATION_OBJECT_KEY,
expected_class=SensitivityClass.FINANCIAL,
max_supported_version=_CALCULATION_CATALOGUE_VERSION,
)
except (ClassificationError, EnvelopeVersionError) as exc:
raise_catalogue_integrity_error(
exc,
error_cls=CalculationRevisionPersistenceError,
label="calculation-revision",
translated_message=_CALCULATION_PERSISTENCE_MESSAGE,
logger=_LOGGER,
)
if record is None:
return CalculationRevisionCatalogue()
envelope = Envelope[CalculationRevisionCatalogue].model_validate_json(record.payload.decode(UTF_8_ENCODING))
if envelope.classification is not SensitivityClass.FINANCIAL:
_LOGGER.error(
"calculation-revision catalogue classification mismatch",
extra={
"expected_classification": SensitivityClass.FINANCIAL.value,
"actual_classification": envelope.classification.value,
},
)
raise CalculationRevisionPersistenceError(
"calculation-revision catalogue classification mismatch",
translated_message=_CALCULATION_PERSISTENCE_MESSAGE,
context={
"reason": "classification_mismatch",
"expected_classification": SensitivityClass.FINANCIAL.value,
"actual_classification": envelope.classification.value,
},
)
if envelope.schema_version > _CALCULATION_CATALOGUE_VERSION:
_LOGGER.error(
"calculation-revision catalogue envelope version unsupported",
extra={
"stored_schema_version": envelope.schema_version,
"max_supported_version": _CALCULATION_CATALOGUE_VERSION,
},
)
raise CalculationRevisionPersistenceError(
"calculation-revision catalogue envelope version unsupported",
translated_message=_CALCULATION_PERSISTENCE_MESSAGE,
context={
"reason": "unsupported_envelope_version",
"stored_schema_version": envelope.schema_version,
"max_supported_version": _CALCULATION_CATALOGUE_VERSION,
},
)
# Post-roundtrip coverage gate: every loaded revision's bundled
# ledger evidence must cover the same contributor set as its
# fingerprint snapshot. A row silently dropped after persistence
# surfaces here on load rather than shipping an unexplainable casilla.
for revision in envelope.payload.values():
assert_revision_snapshot_evidence_coverage(revision)
return envelope.payload
[docs]
def save(self, catalogue: CalculationRevisionCatalogue) -> None:
"""Persist the calculation-revision catalogue to encrypted storage.
Wraps the catalogue (the keyed collection of a modelo's dated
calculation revisions) in an
:class:`~adapters.persistence.storage.Envelope` stamped with the
current schema version, write time, and ``FINANCIAL`` sensitivity
classification, then writes the serialised envelope to the encrypted
store under this repository's namespace and key. An existing catalogue
object at that location is overwritten.
Args:
catalogue: The :class:`CalculationRevisionCatalogue` to serialise and
store.
"""
from ..storage import Envelope, SensitivityClass
envelope = Envelope[CalculationRevisionCatalogue](
schema_version=_CALCULATION_CATALOGUE_VERSION,
written_at=now(),
classification=SensitivityClass.FINANCIAL,
payload=catalogue,
)
self._objects.save(
namespace=_CALCULATION_NAMESPACE,
object_key=_CALCULATION_OBJECT_KEY,
classification=SensitivityClass.FINANCIAL,
schema_version=_CALCULATION_CATALOGUE_VERSION,
written_at=envelope.written_at,
payload=envelope.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def to_secure_object_write(self, catalogue: CalculationRevisionCatalogue) -> SecureObjectWrite:
"""Return the secure-object upsert for ``catalogue`` without committing it.
The returned :class:`~adapters.persistence.storage.SecureObjectWrite`
carries the same :class:`~adapters.persistence.storage.Envelope`
and :class:`~adapters.persistence.storage.SensitivityClass`
classification that :meth:`save` would persist directly. It can be
co-emitted with related secure objects (e.g. the participation index) in
one :meth:`save_with_secure_object_writes` unit of work.
"""
from ..storage import Envelope, SecureObjectWrite, SensitivityClass
envelope = Envelope[CalculationRevisionCatalogue](
schema_version=_CALCULATION_CATALOGUE_VERSION,
written_at=now(),
classification=SensitivityClass.FINANCIAL,
payload=catalogue,
)
return SecureObjectWrite(
namespace=_CALCULATION_NAMESPACE,
object_key=_CALCULATION_OBJECT_KEY,
classification=SensitivityClass.FINANCIAL,
schema_version=_CALCULATION_CATALOGUE_VERSION,
written_at=envelope.written_at,
payload=envelope.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def save_with_secure_object_writes(
self,
catalogue: CalculationRevisionCatalogue,
extra_writes: tuple[SecureObjectWrite, ...],
) -> None:
"""Persist ``catalogue`` plus related secure objects in one unit of work.
The catalogue save and every extra write land or fail together in a
single SQL transaction, so the participation index co-emitted here can
never drift from the calculation revision it indexes (per the
composition-service single-writer discipline).
Args:
catalogue: The :class:`CalculationRevisionCatalogue` to persist.
extra_writes: Additional
:class:`~adapters.persistence.storage.SecureObjectWrite`
objects to commit atomically with the catalogue.
"""
self._objects.save_many((self.to_secure_object_write(catalogue), *extra_writes))
__all__ = [
"CalculationRevisionCatalogueRepository",
]