cadrumo.adapters.persistence.storage.envelope.contract module

Schema-version envelope for file-backed persistence.

The envelope is the single contract every file-backed persistence consumer adheres to. It pins:

  • the on-disk schema version, which must match the consumer’s current schema exactly;

  • the timestamp of the write (timezone-aware datetime);

  • the sensitivity classification (so the substrate can refuse to load a record if a consumer accidentally bypasses its repository);

  • the payload itself (typed strict pydantic v2 model);

  • optional encryption metadata (when the payload is at-rest ciphertext).

The save_envelope() and load_envelope() helpers atomically write and read non-sensitive envelope JSON via atomic_write_text() (standard tier).

The substrate refuses any payload whose schema_version differs from the consumer’s expected version, or which fails classification validation. Sensitive repositories use SecureBoundRepository, which stores the envelope payload shape in encrypted SQL secure objects rather than files.

class AeadAlgorithm(*values)[source]

Bases: StrEnum

Closed catalogue of AEAD identifiers recognised by the substrate.

Members:
AES_256_GCM_V1: AES-256 Galois Counter Mode, version 1 wire

format (12-byte nonce, 16-byte tag). The only algorithm shipping today.

AES_256_GCM_V1
class EncryptionMetadata(**data)[source]

Bases: BaseModel

Encryption envelope describing how the payload was encrypted.

Variables:

  • algorithm – Stable identifier for the AEAD primitive used. Today only aes-256-gcm-v1 is defined; future primitives register their own identifier.

  • nonce_b64 – Base64-encoded 12-byte nonce.

  • ciphertext_b64 – Base64-encoded ciphertext_with_tag.

  • associated_data_b64 – Base64-encoded AAD bytes. The field is required so persisted metadata distinguishes an explicitly empty AAD from malformed metadata where the AAD member is missing.

Parameters:

data (Any)

algorithm: AeadAlgorithm
nonce_b64: str
ciphertext_b64: str
associated_data_b64: str
classmethod from_blob(blob, *, associated_data=b'')[source]

Build metadata from an encrypted blob.

Return type:

EncryptionMetadata

Returns:

EncryptionMetadata derived from an EncryptedBlob.

Parameters:
to_blob()[source]

Reconstruct the EncryptedBlob from encoded fields.

Return type:

EncryptedBlob

associated_data()[source]

Decode the associated-data bytes.

Return type:

bytes

class Envelope(**data)[source]

Bases: BaseModel, Generic

Frozen pydantic v2 envelope wrapping a typed file-backed payload.

Variables:

  • schema_version – Integer version that consumers compare to their expected version. Older and newer versions are refused.

  • written_at – Timezone-aware datetime captured at write time.

  • classification – The SensitivityClass declared by the writer. Mismatches at load time raise ClassificationError.

  • payload – The typed payload. Plaintext is stored when encryption is None; ciphertext lives in encryption.ciphertext_b64 when present, and payload is then a placeholder consumer-typed value.

  • encryption – Optional encryption metadata. None for plaintext envelopes.

Parameters:

data (Any)

schema_version: int
written_at: datetime
classification: SensitivityClass
payload: PayloadT
encryption: EncryptionMetadata | None
classmethod for_payload_type(payload_cls)[source]

Return the Envelope parameterised for payload_cls.

This typed factory avoids a bare cast(Any, Envelope).__class_getitem__(...) at call sites. The returned class is the concrete generic alias Pydantic needs at the JSON validation boundary. The cast to type[Envelope[PayloadT]] is safe because __class_getitem__ on a PEP-695 generic model returns exactly the parameterised subtype; Pydantic registers it as a model class whose payload field is constrained to payload_cls.

Return type:

type[Envelope[PayloadT]]

Parameters:

payload_cls (type[PayloadT])

parameterized_envelope_type(payload_cls, *, envelope_cls=<class 'cadrumo.adapters.persistence.storage.envelope.contract.Envelope'>)[source]

Return the runtime Pydantic generic class for one payload type.

Keeping the dynamic __class_getitem__ operation behind a free generic function gives callers a type-inferable boundary; Ty cannot propagate the payload variable through the generic model’s classmethod receiver.

Core types: Envelope.

Return type:

type[Envelope[TypeVar]]

Parameters:
save_envelope(envelope, path)[source]

Atomically persist envelope as JSON to path.

Parameters:

  • envelope (Envelope[TypeVar]) – The Envelope to write.

  • path (Path) – Destination file. Parent directory is created if absent.

Raises:

StorageValidationError – When the temporary file or atomic replace operation fails.

Return type:

None

load_envelope(path, envelope_type, *, expected_class, max_supported_version)[source]

Load and validate an envelope from disk.

Parameters:

  • path (Path) – Source file (must exist).

  • envelope_type (type[Envelope[TypeVar]]) – The parameterised envelope class (e.g. Envelope[MyPayloadV1]). Pydantic uses this to validate the JSON against the typed payload.

  • expected_class (SensitivityClass) – The SensitivityClass the consumer expects. Mismatch raises ClassificationError.

  • max_supported_version (int) – The current schema_version the consumer expects. Any different version raises EnvelopeVersionError.

Return type:

Envelope[TypeVar]

Returns:

The validated Envelope at the consumer’s expected version.

Raises: