With the sqlalchemy extra installed, SqlAlchemyEncryption binds encrypted fields onto mapped classes: plaintext-facing attributes are descriptors that encrypt at assignment, and writes that cannot bind per-record coordinates are refused.

from sqlalchemy import Column, LargeBinary, String, create_engine
from sqlalchemy.orm import DeclarativeBase

from floorvault import AdaptiveKeyProvider, FloorVault, SqlAlchemyEncryption

crypto = FloorVault(AdaptiveKeyProvider(service_name="my-app").resolve_key(), app_instance_id="my-app")


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id = Column(String, primary_key=True)
    tenant = Column(String, nullable=False)
    ssn_ct = Column(LargeBinary, nullable=True)


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

vault = SqlAlchemyEncryption(crypto, schema_id="my-app.v1")
vault.protect(User, id_attr="id", tenant_attr="tenant", fields={"ssn": "ssn_ct"})
Session = vault.session_factory(bind=engine)

with Session() as session:
    user = User(id="u1", tenant="acme")  # identity attrs first — binding needs them
    user.ssn = "123-45-6789"  # encrypts here; only ciphertext is mapped
    session.add(user)
    session.commit()
    assert session.get(User, "u1").ssn == "123-45-6789"

Fail-closed invariants (full contract: SPEC.md §14.1):

• obj.ssn_ct = b"plaintext" is refused at assignment — only envelope-shaped values may occupy a ciphertext column.
• session.execute(insert(User).values(ssn_ct=...)), executemany sets, Query.update, and the legacy bulk APIs (bulk_insert_mappings/bulk_update_mappings/bulk_save_objects) on ciphertext columns or protected classes raise UnsupportedWriteError.
• tenant_attr binds the tenant into the record coordinate, so a ciphertext cannot be replayed across tenants; revision_attr binds a per-record revision (same-coordinate replay detection, not whole-database rollback protection).

Encrypt a plaintext column into an existing destination column without deleting the source. This continues the existing-tables example (crypto) on its own in-memory table:

import sqlite3

from floorvault import migrate_plaintext_column, verify_encrypted_column

connection = sqlite3.connect(":memory:")
connection.execute(
    "CREATE TABLE users (id TEXT PRIMARY KEY, private_value TEXT, private_value_cipher BLOB)"
)
connection.execute("INSERT INTO users (id, private_value) VALUES (?, ?)", ("u1", "plain-secret"))

migrate_plaintext_column(
    connection, crypto, table="users", id_column="id",
    source_column="private_value", destination_column="private_value_cipher",
)
connection.commit()

assert verify_encrypted_column(
    connection, crypto, table="users", id_column="id",
    source_column="private_value", destination_column="private_value_cipher",
) == 1

The migration validates identifiers, refuses a populated destination, binds each value to its real record ID, and rolls back on failure. Remove the plaintext column only after independent verification and a backup policy — drop_plaintext_column() arms PRAGMA secure_delete, checkpoints the WAL, and optionally VACUUMs, but filesystem residue can survive; destroying the file is the only complete guarantee.

Exact-match lookups over an encrypted column need an indexed value beside the ciphertext. Storing one leaks something, so this is opt-in and a trade made deliberately. The snippet is self-contained after crypto and master_key from the existing-tables example; tests/test_searchable_beacons.py executes the same workflow.

import sqlite3

from floorvault.beacons import BeaconIndexer, derive_beacon_key, suggest_beacon_bits

connection = sqlite3.connect(":memory:")
connection.execute(
    "CREATE TABLE contacts (id TEXT PRIMARY KEY, email_cipher BLOB, email_beacon BLOB)"
)
connection.execute("CREATE INDEX contacts_email_beacon ON contacts (email_beacon)")

record_id, email = "c1", "alice@example.com"

# Size the width to the table, don't pick a constant: the anonymity a beacon
# gives is roughly one bucket's occupancy.
indexer = BeaconIndexer(derive_beacon_key(master_key), bits=suggest_beacon_bits(200_000))

# On write: store the ciphertext and the beacon in an indexed column.
connection.execute(
    "INSERT INTO contacts (id, email_cipher, email_beacon) VALUES (?, ?, ?)",
    (record_id, crypto.encrypt(email, table="contacts", record_id=record_id, column="email_cipher"),
     indexer.beacon(email, scope="contacts.email")),
)
connection.commit()

# On read: the bucket narrows candidates; decryption confirms the match.
bucket = indexer.beacon(email, scope="contacts.email")
for candidate_id, ciphertext in connection.execute(
    "SELECT id, email_cipher FROM contacts WHERE email_beacon = ?", (bucket,)
):
    if crypto.decrypt(ciphertext, table="contacts", record_id=candidate_id, column="email_cipher") == email:
        break  # confirmed
else:
    raise AssertionError("beacon lookup found no matching record")

What this costs: equal values produce equal beacons, so unequal beacons prove unequal values; bucket occupancy still tracks the plaintext distribution, so a skewed column shows a skewed beacon histogram. Truncation makes the index non-injective — it does not make the data uniform — so size the width to the row count, and do not beacon a low-cardinality column or one you never look up by equality. Widths are byte-aligned (4 and 8 bits are the same index); BeaconIndexer rejects a width it cannot store rather than rounding it. A beacon hit is not proof of equality — always confirm by decrypting. Full statement: SECURITY.md §5.

Three ways to satisfy resolve_key():

import os

from floorvault import AdaptiveKeyProvider, FloorVault

# 1. Explicit key, 64 hex characters, from your own secret source.
crypto = FloorVault(bytes.fromhex(os.environ["APPSTATE_KEY"]), app_instance_id="my-app")

# 2. A 0600 local key file, created on first use and reused afterwards. This is
#    Tier 3: anyone who can copy the file can recover the key.
crypto = FloorVault(
    AdaptiveKeyProvider(service_name="my-app", allow_disk_fallback=True).resolve_key(),
    app_instance_id="my-app",
)

# 3. Install the OS-native tier for the platform (floorvault[macos] on macOS).
crypto = FloorVault(AdaptiveKeyProvider(service_name="my-app").resolve_key(), app_instance_id="my-app")

VaultTransitProvider is a fourth option: it keeps the master key wrapped by a HashiCorp Vault Transit key instead of a local file; the wrapped blob lives in a governed on-disk generation store (SPEC.md §10.6–10.7):

from floorvault.providers.vault_transit import VaultTransitProvider

provider = VaultTransitProvider(
    vault_addr="https://vault.internal:8200",
    token=os.environ["VAULT_TOKEN"],  # Transit datakey/decrypt/rewrap perms
    key_name="floorvault-master",
    store_dir="/var/lib/myapp/floorvault",  # 0700; holds store.id + generations
    app_instance_id="my-app",
)
crypto = FloorVault(provider.resolve_key(), app_instance_id="my-app")

# Rewrap under a new Transit KEK version (CAS-publishes a new generation):
provider.rewrap()

Every Vault failure raises CustodyDowngradeError; a configured provider never silently falls back to file custody. HTTPS only, verified TLS, bounded timeouts/retries, redirects refused unless a standby host is trusted. The resolved key is cached briefly (300 s default) — that bounds Transit latency, not revocation protection. Full setup: docs/VAULT-TRANSIT.md.

Rotate a store under a new key with resumable progress and post-rotation verification:

from floorvault import AdaptiveKeyProvider, FloorVault, KeyRing, rotate_vault_store

# VaultStore is the structured item store this rotation operates on; it is not
# re-exported at the top level.
from floorvault.vaultkit import VaultStore
from pathlib import Path

# VaultStore does NOT expand "~" -- a literal tilde would become a directory
# named "~" relative to the working directory. Expand it explicitly.
VAULT_DIR = Path("~/.floor/vault").expanduser()

old_crypto = FloorVault(AdaptiveKeyProvider(service_name="my-app").resolve_key(), app_instance_id="my-app")
store = VaultStore(VAULT_DIR, crypto=old_crypto)

# A new_master_key from your key provider, wrapped in its own engine.
new_crypto = FloorVault(new_master_key, app_instance_id="my-app")

rotate_vault_store(store, source_ring=KeyRing({0: old_crypto}), new_vault=new_crypto, new_key_id=1)

# REQUIRED: the `store` above was built on old_crypto and CANNOT read the
# re-sealed records -- rotation sealed every envelope under the NEW master key.
# Rebuild the store on the new key and repoint every holder of the old one.
store = VaultStore(VAULT_DIR, crypto=new_crypto)

When rotating to different key material, rebuild the original store. Its convenience reads cannot authenticate records sealed under the new master: resolve_secret, get_meta, and list_items raise DecryptionVerificationError, while has_items() does not decrypt and cannot surface the mismatch. Rotation is durable and verified — if the process exits before rebuilding and the provider still resolves the old key, the store stays unreadable across restarts. Rebuild the store rather than mutating it, and make the new key resolvable before you rotate. To read a store spanning generations, use KeyRing with read_sealed_item.

Create an authenticated recovery bundle using a separately protected recovery key — keep it separate from the vault and its backups. This recipe requires master_key and recovery_key as distinct 32-byte bytes or bytearray values from your secret source, not the HardenedMemoryKey handle returned by resolve_key():

from floorvault import recover_master_key, wrap_master_key

bundle = wrap_master_key(master_key, recovery_key)
recovered = recover_master_key(bundle, recovery_key)

The recovery bundle is not a substitute for protecting the recovery key.