Source code for aeat.adapters.persistence.storage._schema_lineage

"""Schema-lineage policy for persisted secure-object payloads.

Persisted rows carry the ``schema_version`` they were written under, while
the consumer contract names the current version. This module owns the
policy that keeps every version from the durability floor to the current
version readable: the version gate is a *ceiling* — a version above the
consumer's current version is refused as written-by-a-newer-application,
and a version below it is readable exactly when the per-hop upgrade chain
up to the current version is complete. Registered upgraders transform
decrypted plaintext payload bytes one version step at a time; ciphertext,
AEAD associated data, and revision-lineage metadata are never rewritten by
a read.

The upgrader registry is EMPTY while every registered namespace sits at
schema version 1. A future schema bump MUST land the one-hop upgrader for
its namespace in the same change, or the lineage gate
(``tests/test_schema_lineage.py``) fails — that gate is what makes "a
version bump strands years-old taxpayer data" structurally impossible.
:data:`SECURE_OBJECT_DURABILITY_FLOOR` moves forward only through a
superseding accepted ADR.

See Also:
    :func:`~adapters.persistence.storage.sql._secure_object_row_codec.secure_object_record_from_row`
        Row decode path that applies this policy before returning a record.
    Governing vault record
        ``2026-07-08-released-data-durability-adr`` (version gates become
        ceilings with an upgrade dispatch).
"""

from __future__ import annotations

from collections.abc import Callable, Mapping
from typing import Final

from .errors import EnvelopeVersionError, StorageValidationError

#: Upgrades one decrypted plaintext payload from ``from_version`` to
#: ``from_version + 1`` for its namespace. Pure bytes-to-bytes; never touches
#: ciphertext or row metadata.
SecureObjectSchemaUpgrader = Callable[[bytes], bytes]

#: Oldest secure-object schema version every read path keeps readable.
#: Starts at the from-birth version; moves forward only through a
#: superseding accepted ADR.
SECURE_OBJECT_DURABILITY_FLOOR: Final[int] = 1

_SCHEMA_UPGRADERS: dict[tuple[str, int], SecureObjectSchemaUpgrader] = {}


[docs] def register_secure_object_schema_upgrader( namespace: str, from_version: int, upgrader: SecureObjectSchemaUpgrader, ) -> None: """Register the one-hop payload upgrader for ``(namespace, from_version)``. A schema bump for a namespace lands its upgrader through this function in the same change that raises the namespace's declared ``schema_version``; the lineage gate fails until it does. Raises: StorageValidationError: When an upgrader for the hop is already registered — two competing transformations for one hop is a wiring error, never a merge. """ key = (namespace, from_version) if key in _SCHEMA_UPGRADERS: raise StorageValidationError( context={"namespace": namespace, "from_version": from_version}, translated_message="errors.storage.namespace.schema_upgrader_already_registered", ) _SCHEMA_UPGRADERS[key] = upgrader
[docs] def deregister_secure_object_schema_upgrader(namespace: str, from_version: int) -> None: """Remove a registered upgrader hop. Production wiring never removes a hop — dropping one would re-open the stranding gap the lineage gate closes. This exists so tests that register a real upgrader against a scratch namespace can restore the registry in their teardown. """ _SCHEMA_UPGRADERS.pop((namespace, from_version), None)
[docs] def missing_upgrade_hops( *, namespace: str, from_version: int, to_version: int, upgraders: Mapping[tuple[str, int], SecureObjectSchemaUpgrader] | None = None, ) -> tuple[int, ...]: """Return the ``from_version`` of every unregistered hop in the chain. An empty tuple means every step from ``from_version`` to ``to_version`` has a registered upgrader (vacuously so when the versions are equal). """ source = _SCHEMA_UPGRADERS if upgraders is None else upgraders return tuple(version for version in range(from_version, to_version) if (namespace, version) not in source)
[docs] def ensure_schema_version_readable( *, namespace: str, schema_version: int, current_version: int, upgraders: Mapping[tuple[str, int], SecureObjectSchemaUpgrader] | None = None, ) -> None: """Refuse a stored version the current application cannot read. The gate is a ceiling, not an equality: a version above ``current_version`` was written by a newer application and is refused outright; a version below it is accepted exactly when the registered upgrade chain up to ``current_version`` is complete, and refused loudly — naming the first missing hop — when it is not. Raises: EnvelopeVersionError: When ``schema_version`` exceeds ``current_version`` (future shape) or the upgrade chain has a missing hop (readable only after the missing upgrader ships). """ if schema_version > current_version: raise EnvelopeVersionError( context={ "namespace": namespace, "schema_version": schema_version, "expected": current_version, }, translated_message="errors.storage.namespace.schema_version_from_future", ) missing = missing_upgrade_hops( namespace=namespace, from_version=schema_version, to_version=current_version, upgraders=upgraders, ) if missing: raise EnvelopeVersionError( context={ "namespace": namespace, "schema_version": schema_version, "expected": current_version, "missing_from_version": missing[0], }, translated_message="errors.storage.namespace.schema_upgrade_path_missing", )
[docs] def upgrade_secure_object_payload( payload: bytes, *, namespace: str, from_version: int, to_version: int, upgraders: Mapping[tuple[str, int], SecureObjectSchemaUpgrader] | None = None, ) -> bytes: """Chain-upgrade a decrypted payload from ``from_version`` to ``to_version``. Validates the chain first (so a missing hop refuses before any transformation runs), then applies each registered one-hop upgrader in order. Equal versions return the payload unchanged. Raises: EnvelopeVersionError: When ``from_version`` exceeds ``to_version`` or the chain has a missing hop. """ ensure_schema_version_readable( namespace=namespace, schema_version=from_version, current_version=to_version, upgraders=upgraders, ) source = _SCHEMA_UPGRADERS if upgraders is None else upgraders for version in range(from_version, to_version): payload = source[(namespace, version)](payload) return payload
__all__ = [ "SECURE_OBJECT_DURABILITY_FLOOR", "SecureObjectSchemaUpgrader", "deregister_secure_object_schema_upgrader", "ensure_schema_version_readable", "missing_upgrade_hops", "register_secure_object_schema_upgrader", "upgrade_secure_object_payload", ]