"""Encrypted SQL persistence for the capital-goods IVA regularización register.
The :class:`domain.bienes_inversion.BienesInversionIvaRegister` document is
stored as a ``FINANCIAL``
:class:`adapters.persistence.storage.SensitivityClass` secure object in the
primary database through
:class:`adapters.persistence.storage.SecureObjectRepository`. The singleton
namespace, default object key, schema version, and custody contracts come from
:data:`adapters.persistence.storage.PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE`.
The register is authoritative primary state (the operator declares each tracked
capital good), not a rebuildable cache; it therefore carries a strict
save/load/equality roundtrip plus an anti-tautology proof.
See Also:
:mod:`domain.bienes_inversion`
Typed register payload models persisted here.
:mod:`adapters.persistence.profile.assets`
Sibling profile-local secure-object adapter whose shape this mirrors.
"""
from __future__ import annotations
from pathlib import Path
from ....core.errors import AeatError
from ....core.external_constants import UTF_8_ENCODING
from ....core.logging import get_logger
from ....core.time import now
from ....domain.bienes_inversion import (
BienesInversionIvaRegister,
BienInversionIvaRecord,
BienInversionRecordError,
)
from ..storage import (
PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE,
SecureObjectRepository,
SensitivityClass,
secure_object_logical_path,
secure_object_repository_for_active_bucket,
secure_object_repository_for_bucket,
)
_log = get_logger(__name__)
BIENES_INVERSION_REGISTER_FILENAME = "bienes-inversion-iva-register.secure-object"
_REGISTER_SECURE_OBJECT_VERSION = PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE.schema_version
_REGISTER_NAMESPACE = PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE.namespace
_REGISTER_OBJECT_KEY = PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE.require_default_object_key()
def _secure_object_marker(namespace: str, filename: str) -> Path:
return secure_object_logical_path(namespace, filename)
[docs]
def load_bienes_inversion_register() -> BienesInversionIvaRegister:
"""Load the register, returning an empty register when absent.
Returns:
Persisted :class:`BienesInversionIvaRegister`, or an empty one when no
envelope exists.
"""
return BienesInversionIvaRegisterRepository().load()
[docs]
def save_bienes_inversion_register(register: BienesInversionIvaRegister) -> Path:
"""Persist ``register`` as a governed FINANCIAL-class encrypted envelope.
Args:
register: Register document to encrypt and write.
Returns:
Logical secure-object marker for the persisted register.
"""
repository = BienesInversionIvaRegisterRepository()
repository.save(register)
return repository.envelope_path
[docs]
def declare_bien_inversion(record: BienInversionIvaRecord) -> BienesInversionIvaRegister:
"""Atomically add ``record`` to the encrypted register.
Args:
record: Capital-good record to insert.
Returns:
The updated :class:`BienesInversionIvaRegister` including the new record.
"""
return BienesInversionIvaRegisterRepository().add(record)
[docs]
class BienesInversionIvaRegisterRepository:
"""Governed repository for the encrypted register singleton.
The singleton row is owned by
:data:`adapters.persistence.storage.PROFILE_BIENES_INVERSION_IVA_REGISTER_NAMESPACE`
and persisted through
:class:`adapters.persistence.storage.SecureObjectRepository`.
"""
def __init__(
self,
*,
bucket_id: str | None = None,
objects: SecureObjectRepository | None = None,
) -> None:
"""Initialise the repository.
Args:
bucket_id: Explicit bucket to bind to, resolved through
:func:`~adapters.persistence.storage.secure_object_repository_for_bucket`.
Lets a caller that already knows the target bucket (e.g. the
calculate-path advisory collector, which receives ``bucket_id``
from its context rather than the process-global active-profile
pointer) load the register for that bucket explicitly. Ignored
when ``objects`` is supplied.
objects: Explicit :class:`SecureObjectRepository` override (tests).
When neither ``objects`` nor ``bucket_id`` is supplied, defaults
to the active-bucket secure object store.
"""
if objects is not None:
self._objects = objects
elif bucket_id is not None:
self._objects = secure_object_repository_for_bucket(bucket_id)
else:
self._objects = secure_object_repository_for_active_bucket()
@property
def envelope_path(self) -> Path:
"""Logical path retained for callers that display the storage target."""
return _secure_object_marker(_REGISTER_NAMESPACE, BIENES_INVERSION_REGISTER_FILENAME)
[docs]
def load(self) -> BienesInversionIvaRegister:
"""Load the register, returning an empty document when absent.
Returns:
Decrypted :class:`BienesInversionIvaRegister`.
Raises:
BienInversionRecordError: When the envelope exists but cannot be
loaded or decrypted.
"""
try:
record = self._objects.load(
_REGISTER_NAMESPACE,
self._object_key,
expected_class=SensitivityClass.FINANCIAL,
max_supported_version=_REGISTER_SECURE_OBJECT_VERSION,
)
if record is None:
return BienesInversionIvaRegister()
return BienesInversionIvaRegister.model_validate_json(record.payload.decode(UTF_8_ENCODING))
except (OSError, AeatError) as exc:
_log.debug(
"bienes inversion register load failed",
extra={
"namespace": _REGISTER_NAMESPACE,
"object_key": self._object_key,
"error_type": type(exc).__name__,
},
)
raise BienInversionRecordError(
f"unable to load bienes inversion register: {self._object_key}",
context={"namespace": _REGISTER_NAMESPACE, "object_key": self._object_key},
translated_message="adapters.persistence.profile.bienes_inversion.errors.load_register_failed",
) from exc
[docs]
def save(self, register: BienesInversionIvaRegister) -> None:
"""Persist ``register`` as FINANCIAL-class ciphertext.
Args:
register: Register document to encrypt and write.
"""
self._save_unlocked(register)
_log.info(
"saved %d bienes inversion records to secure object %s",
len(register.records),
self._object_key,
)
[docs]
def add(self, record: BienInversionIvaRecord) -> BienesInversionIvaRegister:
"""Atomically add ``record`` and refuse duplicate identifiers.
Args:
record: Capital-good record to insert.
Returns:
The :class:`BienesInversionIvaRegister` including the new record.
Raises:
BienInversionRecordError: When a record with the same identifier
already exists.
"""
current = self.load()
if any(existing.identifier == record.identifier for existing in current.records):
raise BienInversionRecordError(
f"bien de inversion {record.identifier!r} already exists",
context={"record_id": record.identifier},
suggestion=None,
translated_message="adapters.persistence.profile.bienes_inversion.errors.record_already_exists",
)
updated = BienesInversionIvaRegister(records=(*current.records, record))
self._save_unlocked(updated)
return updated
def _save_unlocked(self, register: BienesInversionIvaRegister) -> None:
self._objects.save(
namespace=_REGISTER_NAMESPACE,
object_key=self._object_key,
classification=SensitivityClass.FINANCIAL,
schema_version=_REGISTER_SECURE_OBJECT_VERSION,
written_at=now(),
payload=register.model_dump_json().encode(UTF_8_ENCODING),
)
@property
def _object_key(self) -> str:
return _REGISTER_OBJECT_KEY
__all__ = [
"BienesInversionIvaRegisterRepository",
"declare_bien_inversion",
"load_bienes_inversion_register",
"save_bienes_inversion_register",
]