"""Portable profile-bundle serialisation for bucket export/import.
This module composes
:class:`~aeat.domain.user_profile.UserProfilePortableExport` payloads at
the application boundary. A v3 bundle contains the profile record plus
the four bucket-local history categories that must move with it: work
units, ledger transactions, calculation revisions, and filing records.
The v3 shape additionally carries the generic secure-object custody
schema and coverage manifest, default-empty until the transport-aware
phases populate them.
The ledger category is loaded as a
:class:`~aeat.domain.transactions.TransactionCatalogue` through
:class:`~aeat.domain.transactions.TransactionCatalogueRepository`.
Bundles carry typed domain-model payloads, not encrypted blobs, key
material, or raw secure-storage rows. Export reads domain records from
their owning repositories; import saves those records through the target
bucket's repository save paths so the target bucket re-encrypts them
under its own data-encryption key.
The bundle version gate is a ceiling with a durability floor: a version
above :data:`BUNDLE_SCHEMA_VERSION` was written by a newer application
and is refused; a version at or above :data:`BUNDLE_DURABILITY_FLOOR` is
readable exactly when the per-hop chain in
:data:`BUNDLE_PAYLOAD_UPGRADERS` reaches the current version. The floor
starts at the current version (no released bundles exist below it) and
moves forward only through a superseding accepted ADR
(``2026-07-08-released-data-durability-adr``). Callers must provision and
collision-check the target bucket and hold the appropriate bucket
session before deserialising; this module performs schema-version
validation and typed repository writes.
"""
from __future__ import annotations
import json
from collections.abc import Callable, Mapping
from typing import TYPE_CHECKING, Final
from ...adapters.persistence.storage import STORAGE_NAMESPACE_REGISTRY, StorageCustodyProfile
from ...core.errors import AeatError
if TYPE_CHECKING:
from ...domain.user_profile import (
CarriedSecureObject,
CoverageManifest,
UserProfilePortableExport,
)
#: Current bundle write version. Every export stamps this.
BUNDLE_SCHEMA_VERSION: Final[int] = 3
#: Oldest bundle version the import path keeps readable. Starts at the
#: current version (no released bundles exist below it); moves forward only
#: through a superseding accepted ADR.
BUNDLE_DURABILITY_FLOOR: Final[int] = 3
#: One-hop raw-payload upgraders keyed by ``from_version``: each transforms
#: the parsed JSON mapping of a version-N bundle into the version-N+1 shape
#: (including restamping ``bundle_schema_version``) BEFORE strict pydantic
#: validation — the raw mapping is the one sanctioned pre-validation
#: boundary. Empty while the floor equals the current version; a version
#: bump MUST land its hop here in the same change or the lineage gate fails.
BUNDLE_PAYLOAD_UPGRADERS: Mapping[int, Callable[[dict[str, object]], dict[str, object]]] = {}
#: Versions the import path accepts: the complete floor-to-current range.
SUPPORTED_BUNDLE_SCHEMA_VERSIONS: frozenset[int] = frozenset(
range(BUNDLE_DURABILITY_FLOOR, BUNDLE_SCHEMA_VERSION + 1),
)
[docs]
def validate_bundle_payload(
raw_json: bytes | str,
*,
expected_written_version: int | None = None,
) -> UserProfilePortableExport:
"""Parse, chain-upgrade, and strictly validate a serialized bundle payload.
Reads the payload's own ``bundle_schema_version``, refuses a future
version (above :data:`BUNDLE_SCHEMA_VERSION`) or one below
:data:`BUNDLE_DURABILITY_FLOOR`, chain-upgrades an older supported
payload hop by hop through :data:`BUNDLE_PAYLOAD_UPGRADERS`, and
validates the result against the current strict
:class:`~aeat.domain.user_profile.UserProfilePortableExport` model.
Args:
raw_json: The serialized bundle payload (decrypted transport bytes
or the plaintext export text).
expected_written_version: When set, the version a transport envelope
declared for this payload; a payload whose own stamped version
differs is refused before any upgrade runs.
Raises:
UnsupportedBundleSchemaVersionError: When the payload does not carry
an integer ``bundle_schema_version``, the version is outside the
floor-to-current range, an upgrade hop is unregistered, or the
stamped version contradicts ``expected_written_version``.
"""
from ...domain.user_profile import UserProfilePortableExport
payload = json.loads(raw_json)
if not isinstance(payload, dict):
raise UnsupportedBundleSchemaVersionError("bundle payload is not a JSON object")
written_version = payload.get("bundle_schema_version")
if not isinstance(written_version, int) or isinstance(written_version, bool):
raise UnsupportedBundleSchemaVersionError(
f"bundle payload carries no integer bundle_schema_version (got {written_version!r})",
)
if expected_written_version is not None and written_version != expected_written_version:
raise UnsupportedBundleSchemaVersionError(
f"bundle payload is stamped bundle_schema_version {written_version} but its "
f"transport envelope declares {expected_written_version}",
)
supported = ",".join(str(version) for version in sorted(SUPPORTED_BUNDLE_SCHEMA_VERSIONS))
if written_version > BUNDLE_SCHEMA_VERSION:
raise UnsupportedBundleSchemaVersionError(
f"bundle_schema_version {written_version} was written by a newer application; "
f"this application reads up to version {BUNDLE_SCHEMA_VERSION}",
context={
"bundle_schema_version": str(written_version),
"supported_versions": supported,
},
translated_message="application.user_profile.errors.unsupported_bundle_schema_version",
)
if written_version < BUNDLE_DURABILITY_FLOOR:
raise UnsupportedBundleSchemaVersionError(
f"bundle_schema_version {written_version!r} is not supported; "
f"supported versions: {sorted(SUPPORTED_BUNDLE_SCHEMA_VERSIONS)}",
context={
"bundle_schema_version": str(written_version),
"supported_versions": supported,
},
translated_message="application.user_profile.errors.unsupported_bundle_schema_version",
)
for hop in range(written_version, BUNDLE_SCHEMA_VERSION):
upgrader = BUNDLE_PAYLOAD_UPGRADERS.get(hop)
if upgrader is None:
raise UnsupportedBundleSchemaVersionError(
f"bundle_schema_version {written_version} has no registered upgrade "
f"from version {hop}; the payload is supported but this build cannot upgrade it",
context={
"bundle_schema_version": str(written_version),
"missing_from_version": str(hop),
},
translated_message="application.user_profile.errors.unsupported_bundle_schema_version",
)
payload = upgrader(payload)
# JSON-mode validation: the strict model accepts JSON arrays as tuple
# fields only on the json path, so the (possibly upgraded) mapping is
# re-serialized rather than validated as python objects.
return UserProfilePortableExport.model_validate_json(json.dumps(payload))
# ---------------------------------------------------------------------------
# Serialiser — S105
# ---------------------------------------------------------------------------
[docs]
def serialize_profile_bundle(
*,
bucket_id: str,
custody_profile: StorageCustodyProfile | str = StorageCustodyProfile.STRUCTURED,
) -> UserProfilePortableExport:
"""Build a v3 :class:`~aeat.domain.user_profile.UserProfilePortableExport`.
Reads the profile record and all four financial-history categories
from ``bucket_id``'s encrypted repositories and assembles them into
one portable payload. The caller is responsible for ensuring a live
bucket session is active for ``bucket_id``.
Args:
bucket_id: Profile bucket whose domain repositories are exported.
custody_profile: Secure-object custody scope to apply, as a
:class:`~aeat.adapters.persistence.storage.StorageCustodyProfile`
or one of its string values.
The bundle carries only decrypted pydantic domain-model payloads
(no encrypted envelopes or key material). The recipient re-encrypts
each object under its own bucket data-encryption key through the
standard repository save paths on import.
"""
from ...adapters.persistence.profile.modelos_calculation import CalculationRevisionCatalogueRepository
from ...adapters.persistence.profile.modelos_filing import ModeloRecordCatalogueRepository
from ...adapters.persistence.profile.modelos_work_units import WorkUnitCatalogueRepository
from ...adapters.persistence.profile.transactions import TransactionCatalogueRepository
from ...domain.user_profile import UserProfilePortableExport
from ._orchestration import build_lifecycle_service
record = build_lifecycle_service(bucket_id=bucket_id).read(bucket_id)
work_unit_catalogue = WorkUnitCatalogueRepository(bucket_id=bucket_id).load()
work_units = tuple(work_unit_catalogue)
transaction_catalogue = TransactionCatalogueRepository(bucket_id=bucket_id).load()
ledger_transactions = tuple(transaction_catalogue)
revision_catalogue = CalculationRevisionCatalogueRepository(bucket_id=bucket_id).load()
calculation_revisions = tuple(revision_catalogue)
filing_catalogue = ModeloRecordCatalogueRepository(bucket_id=bucket_id).load()
filing_records = tuple(filing_catalogue)
carried_objects, coverage_manifest = _build_secure_object_custody_payload(
bucket_id=bucket_id,
custody_profile=_normalize_custody_profile(custody_profile),
)
return UserProfilePortableExport(
profile=record,
work_units=work_units,
ledger_transactions=ledger_transactions,
calculation_revisions=calculation_revisions,
filing_records=filing_records,
carried_objects=carried_objects,
coverage_manifest=coverage_manifest,
)
def _normalize_custody_profile(custody_profile: StorageCustodyProfile | str) -> StorageCustodyProfile:
if isinstance(custody_profile, StorageCustodyProfile):
return custody_profile
try:
return StorageCustodyProfile(custody_profile)
except ValueError as exc:
from ...domain.user_profile import ProfileExportError
raise ProfileExportError(
f"unsupported custody_profile {custody_profile!r}; expected one of "
f"{tuple(profile.value for profile in StorageCustodyProfile)}",
context={"custody_profile": custody_profile},
) from exc
#: Namespaces carried by the typed bundle fields; they count as covered for the
#: full-custody coverage assertion even though the generic carry skips them.
_TYPED_CATEGORY_NAMESPACES: frozenset[str] = frozenset(
{
"aeat.application.user_profile.value",
"aeat.domain.transactions.bucket",
"aeat.domain.modelos.work_units",
"aeat.domain.modelos.calculation_revisions",
"aeat.domain.modelos.filing_records",
},
)
def _build_secure_object_custody_payload(
*,
bucket_id: str,
custody_profile: StorageCustodyProfile,
) -> tuple[tuple[CarriedSecureObject, ...], CoverageManifest]:
from ...adapters.persistence.storage import secure_object_repository_for_bucket
from ...domain.user_profile import CoverageManifest
from ._custody_carry import carried_namespace_definitions, serialize_carried_objects
repository = secure_object_repository_for_bucket(bucket_id)
populated_namespaces = tuple(repository.list_namespaces())
row_counts_by_namespace = {namespace: len(repository.list_keys(namespace)) for namespace in populated_namespaces}
carried_namespace_set = frozenset(
definition.namespace for definition in carried_namespace_definitions(custody_profile)
)
carried_or_typed = carried_namespace_set | _TYPED_CATEGORY_NAMESPACES
# ``excluded_namespaces`` (for the manifest) is every populated namespace not
# carried by this profile — the deliberately-excluded host-local / derived /
# full-only stores plus the typed-category-covered ones are reported honestly.
excluded_namespaces = tuple(
namespace for namespace in populated_namespaces if namespace not in carried_namespace_set
)
if custody_profile is StorageCustodyProfile.FULL:
# Every registered namespace declares a custody disposition (carried,
# typed-category, or deliberately excluded such as PROCESS_LOCAL credentials
# or the DERIVED participation index), so it is accounted for. The gate fails
# closed only on a populated namespace that is NOT in the registry at all — an
# unclassified durable store that would otherwise be silently dropped.
registered_namespaces = frozenset(definition.namespace for definition in STORAGE_NAMESPACE_REGISTRY.namespaces)
_assert_full_custody_coverage(
populated_namespaces=populated_namespaces,
covered_namespaces=carried_or_typed | registered_namespaces,
)
carried_objects = serialize_carried_objects(bucket_id=bucket_id, profile=custody_profile)
carried_namespaces = tuple(
namespace
for namespace in (definition.namespace for definition in carried_namespace_definitions(custody_profile))
if row_counts_by_namespace.get(namespace, 0) > 0
)
coverage_manifest = CoverageManifest(
custody_profile=custody_profile.value,
carried_namespaces=carried_namespaces,
excluded_namespaces=excluded_namespaces,
row_counts_by_namespace=row_counts_by_namespace,
)
return carried_objects, coverage_manifest
def _assert_full_custody_coverage(
*,
populated_namespaces: tuple[str, ...],
covered_namespaces: frozenset[str],
) -> None:
missing = tuple(namespace for namespace in populated_namespaces if namespace not in covered_namespaces)
if not missing:
return
from ...domain.user_profile import ProfileExportError
raise ProfileExportError(
"full custody profile found a populated secure-object namespace with no registry "
"classification; register a custody disposition for it before it can be backed up",
context={"unclassified_namespaces": missing, "custody_profile": StorageCustodyProfile.FULL.value},
)
# ---------------------------------------------------------------------------
# Deserialiser — S106
# ---------------------------------------------------------------------------
[docs]
def deserialize_profile_bundle(bundle: UserProfilePortableExport, *, target_bucket_id: str) -> None:
"""Import financial-history objects from ``bundle`` into ``target_bucket_id``.
Validates ``bundle.bundle_schema_version`` against
``SUPPORTED_BUNDLE_SCHEMA_VERSIONS`` before any writes; only the
current v3 shape is accepted.
Saves work units, ledger transactions, calculation revisions, and
filing records into the target bucket via the standard repository
save paths. Each domain object is re-encrypted under the target
bucket's own data-encryption key. No ``dict[str, Any]`` intermediate
is used; pydantic models flow directly into typed catalogue saves.
The caller is responsible for:
- Provisioning the target bucket before calling this function.
- Ensuring a live bucket session is active for ``target_bucket_id``.
- Running the two-tier collision guard before provisioning.
Args:
bundle: The validated export bundle.
target_bucket_id: The bucket id under which to write the objects.
Raises:
UnsupportedBundleSchemaVersionError: When
``bundle.bundle_schema_version`` is not in
``SUPPORTED_BUNDLE_SCHEMA_VERSIONS``.
"""
if bundle.bundle_schema_version not in SUPPORTED_BUNDLE_SCHEMA_VERSIONS:
raise UnsupportedBundleSchemaVersionError(
f"bundle_schema_version {bundle.bundle_schema_version!r} is not supported; "
f"supported versions: {sorted(SUPPORTED_BUNDLE_SCHEMA_VERSIONS)}",
)
# The five typed financial-history categories restore through their typed
# catalogue save paths; every other durable secure-object store restores
# generically through the raw substrate, re-keyed and re-encrypted under the
# recipient bucket DEK.
_import_work_units(bundle, target_bucket_id=target_bucket_id)
_import_ledger_transactions(bundle, target_bucket_id=target_bucket_id)
_import_calculation_revisions(bundle, target_bucket_id=target_bucket_id)
_import_filing_records(bundle, target_bucket_id=target_bucket_id)
from ._custody_carry import restore_carried_objects
restore_carried_objects(bundle.carried_objects, target_bucket_id=target_bucket_id)
_rebuild_participation_index(target_bucket_id=target_bucket_id)
def _rebuild_participation_index(*, target_bucket_id: str) -> None:
"""Rebuild the derived transaction-revision participation index after import.
The index is a derived, rebuildable read-cache (excluded from the carry per
``ledger-participation-index-is-derived-rebuildable``); it is regenerated from
the restored revision, work-unit, and filing catalogues.
"""
from ..modelo import rebuild_participation_index
rebuild_participation_index(bucket_id=target_bucket_id)
def _import_work_units(bundle: UserProfilePortableExport, *, target_bucket_id: str) -> None:
from ...adapters.persistence.profile.modelos_work_units import WorkUnitCatalogueRepository
from ...domain.modelos import (
upsert_work_unit,
)
if not bundle.work_units:
return
repo = WorkUnitCatalogueRepository(bucket_id=target_bucket_id)
catalogue = repo.load()
for unit in bundle.work_units:
catalogue = upsert_work_unit(catalogue, unit)
repo.save(catalogue)
def _import_ledger_transactions(bundle: UserProfilePortableExport, *, target_bucket_id: str) -> None:
from ...adapters.persistence.profile.transactions import TransactionCatalogueRepository
from ...domain.transactions import Transaction, TransactionCatalogue
if not bundle.ledger_transactions:
return
repo = TransactionCatalogueRepository(bucket_id=target_bucket_id)
existing = repo.load()
merged: dict[str, Transaction] = dict(existing.transactions)
for txn in bundle.ledger_transactions:
merged[txn.transaction_id] = txn
repo.save(TransactionCatalogue(transactions=merged))
def _import_calculation_revisions(bundle: UserProfilePortableExport, *, target_bucket_id: str) -> None:
from ...adapters.persistence.profile.modelos_calculation import CalculationRevisionCatalogueRepository
from ...domain.modelos import upsert_calculation_revision
if not bundle.calculation_revisions:
return
repo = CalculationRevisionCatalogueRepository(bucket_id=target_bucket_id)
catalogue = repo.load()
for revision in bundle.calculation_revisions:
catalogue = upsert_calculation_revision(catalogue, revision)
repo.save(catalogue)
def _import_filing_records(bundle: UserProfilePortableExport, *, target_bucket_id: str) -> None:
from ...adapters.persistence.profile.modelos_filing import ModeloRecordCatalogueRepository
from ...domain.modelos import (
upsert_filing_record,
)
if not bundle.filing_records:
return
repo = ModeloRecordCatalogueRepository(bucket_id=target_bucket_id)
catalogue = repo.load()
for record in bundle.filing_records:
catalogue = upsert_filing_record(catalogue, record)
repo.save(catalogue)
[docs]
class UnsupportedBundleSchemaVersionError(AeatError):
"""Raised when a bundle carries an unsupported ``bundle_schema_version``."""