### Install pydantic-encryption with all extras

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Install with all optional dependencies included.

```bash
pip install "pydantic-encryption[all]"
```

--------------------------------

### Install pydantic-encryption

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Install the core library using pip.

```bash
pip install pydantic-encryption
```

--------------------------------

### Install Development Dependencies and Run Tests

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Install the project with development dependencies using `pip install -e '.[dev]'` and run tests using `pytest -v`.

```bash
pip install -e ".[dev]"
pytest -v
```

--------------------------------

### Install pydantic-encryption with SQLAlchemy

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Install with SQLAlchemy integration for database support.

```bash
pip install "pydantic-encryption[sqlalchemy]"
```

--------------------------------

### Install pydantic-encryption

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Install the core package or with optional extras for SQLAlchemy or AWS KMS support.

```bash
pip install pydantic-encryption                     # core only
pip install "pydantic-encryption[sqlalchemy]"       # + SQLAlchemy TypeDecorators
pip install "pydantic-encryption[aws]"              # + AWS KMS adapter
pip install "pydantic-encryption[all]"              # everything
```

--------------------------------

### Quick Start: Deferred Decryption with SQLAlchemy

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Demonstrates using DeferredDecryptMixin for batch-decryption of encrypted columns in SQLAlchemy models. Encrypted attributes are only decrypted when first accessed, improving performance.

```python
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic_encryption import DeferredDecryptMixin, SQLAlchemyEncryptedValue


class Base(DeclarativeBase):
    pass


class User(Base, DeferredDecryptMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())


engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async with Session() as session:
    session.add(User(email="john@example.com"))
    await session.commit()

    result = await session.execute(select(User))
    user = result.scalar_one()
    print(user.email)  # "john@example.com" — decrypted on first read
```

--------------------------------

### Install pydantic-encryption with AWS KMS

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Install with AWS KMS encryption support.

```bash
pip install "pydantic-encryption[aws]"
```

--------------------------------

### Searchable Deterministic Hashing with BlindIndex Annotation

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Employ the BlindIndex annotation for deterministic keyed hashing, enabling equality lookups on encrypted data. The constructor accepts a BlindIndexMethod and normalization flags. Requires BLIND_INDEX_SECRET_KEY environment variable. The example demonstrates case-insensitive, whitespace-normalized email lookup and phone number stripping.

```python
from typing import Annotated
from pydantic_encryption import BaseModel, BlindIndex, BlindIndexMethod

class Contact(BaseModel):
    # Case-insensitive, whitespace-normalized email lookup index
    email_index: Annotated[bytes, BlindIndex(
        BlindIndexMethod.HMAC_SHA256,
        normalize_to_lowercase=True,
        strip_whitespace=True,
    )]
    # Phone number stripped of non-digits
    phone_index: Annotated[bytes, BlindIndex(
        BlindIndexMethod.ARGON2,  # memory-hard, brute-force resistant
        strip_non_digits=True,
    )]

c1 = Contact(email_index="Alice@Example.COM ", phone_index="+1 (555) 123-4567")
c2 = Contact(email_index="alice@example.com",  phone_index="15551234567")

# Same normalized value → identical blind index
print(c1.email_index == c2.email_index)  # True
print(c1.phone_index == c2.phone_index)  # True
```

--------------------------------

### DeferredDecryptMixin

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

`DeferredDecryptMixin` hooks into SQLAlchemy's ORM events to install per-column `DecryptOnAccessDescriptor` proxies. This allows for batch decryption of columns on first attribute access, improving performance by only decrypting accessed data.

```APIDOC
## `DeferredDecryptMixin` — on-access batch decryption for async SQLAlchemy

`DeferredDecryptMixin` hooks into SQLAlchemy's ORM events (`mapper_configured`, `load`, `refresh`) to install per-column `DecryptOnAccessDescriptor` proxies and register freshly loaded rows in a session-scoped pending-decrypt bucket. On first attribute access the descriptor issues a single `asyncio.gather` across every sibling row in the session. Columns never accessed stay encrypted at zero cost.

```python
import asyncio
import os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic_encryption import DeferredDecryptMixin, SQLAlchemyEncryptedValue

class Base(DeclarativeBase): pass

class User(Base, DeferredDecryptMixin):
    __tablename__ = "users"
    id:    Mapped[int]   = mapped_column(primary_key=True)
    email: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())
    name:  Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def main():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with Session() as session:
        session.add_all([
            User(email="a@example.com", name="Alice"),
            User(email="b@example.com", name="Bob"),
        ])
        await session.commit()

    async with Session() as session:
        users = (await session.execute(select(User))).scalars().all()
        # First access to `email` on any user triggers batch decrypt of ALL users' email column
        for u in users:
            print(u.email)  # decrypted — "a@example.com", "b@example.com"

        # Manual instance decrypt
        await users[0].decrypt()
        # Batch class-level decrypt
        await User.decrypt_many(users)

asyncio.run(main())
```
```

--------------------------------

### SQLAlchemy PG Encrypted Array

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Example of using SQLAlchemyPGEncryptedArray for encrypting list elements in PostgreSQL. Each element in the list is encrypted individually.

```python
from pydantic_encryption import SQLAlchemyPGEncryptedArray

tags: Mapped[list[str] | None] = mapped_column(SQLAlchemyPGEncryptedArray(), nullable=True)
```

--------------------------------

### Configure Environment Settings

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Configure encryption, KMS, and blind indexing settings using environment variables or .env files. Validation errors are raised at import time.

```bash
# Fernet
ENCRYPTION_METHOD=fernet
ENCRYPTION_KEY=<base64-url-safe-fernet-key>    # generate: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

# AWS KMS — single key
ENCRYPTION_METHOD=aws
AWS_KMS_KEY_ARN=arn:aws:kms:us-east-1:123456789:key/your-key-id
AWS_KMS_REGION=us-east-1
AWS_KMS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE
AWS_KMS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

# AWS KMS — split encrypt/decrypt keys for rotation
AWS_KMS_ENCRYPT_KEY_ARN=arn:aws:kms:us-east-1:123456789:key/new-key
AWS_KMS_DECRYPT_KEY_ARN=arn:aws:kms:us-east-1:123456789:key/old-key

# Blind indexing
BLIND_INDEX_SECRET_KEY=my-secret-blind-index-key
```

```python
from pydantic_encryption import settings

print(settings.ENCRYPTION_METHOD)   # EncryptionMethod.FERNET
print(settings.ENCRYPTION_KEY)      # "gAAAAAB..."
# Validation errors are raised at import time, not at first use:
# - Cannot combine AWS_KMS_KEY_ARN with split ARNs
# - AWS_KMS_ENCRYPT_KEY_ARN requires AWS_KMS_DECRYPT_KEY_ARN
# - ENCRYPTION_METHOD=fernet requires ENCRYPTION_KEY
```

--------------------------------

### Async Model Initialization and Decryption

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Use `async_init()` for constructing models with asynchronous encryption, hashing, and blind indexing. `async_decrypt_data()` is used for asynchronous decryption. All encryption phases run concurrently.

```python
user = await User.async_init(name="John", address="123 Main St", password="secret")
await user.async_decrypt_data()
```

--------------------------------

### Registering Custom Encryption and Blind Index Backends

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Allows registration of custom encryption and blind-indexing backends using factory functions or direct class registration. This enables plugging in alternative implementations.

```python
from pydantic_encryption import EncryptionAdapter, BlindIndexAdapter
from pydantic_encryption.adapters.registry import (
    register_encryption_backend,
    register_encryption_backend_lazy,
    register_blind_index_backend,
)
from pydantic_encryption.types import EncryptionMethod, BlindIndexMethod, EncryptedValue, BlindIndexValue
import os
```

--------------------------------

### Argon2Adapter - Password Hashing

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Explains password hashing using the Argon2 algorithm with `Argon2Adapter`. Includes synchronous, asynchronous hashing, and verification.

```APIDOC
## Argon2Adapter - Password hashing

`Argon2Adapter` implements `HashingAdapter` with a shared `PasswordHasher` instance. Passing an already-hashed `HashedValue` is a no-op.

```python
from pydantic_encryption import Argon2Adapter
from argon2 import PasswordHasher

hashed = Argon2Adapter.hash("my_password")
print(type(hashed))           # <class 'pydantic_encryption.types.HashedValue'>
print(hashed[:15])            # b'$argon2id$v=19$'

# Idempotent
hashed2 = Argon2Adapter.hash(hashed)
print(hashed == hashed2)      # True

# Verify using argon2-cffi directly
ph = PasswordHasher()
print(ph.verify(hashed, "my_password"))  # True

# Async
import asyncio
async def async_hash():
    h = await Argon2Adapter.async_hash("async_password")
    return h
hashed_async = asyncio.run(async_hash())
```
```

--------------------------------

### Configure Fernet Environment Variables

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Set the `ENCRYPTION_METHOD` to `fernet` and provide the generated `ENCRYPTION_KEY` as environment variables for Fernet encryption.

```bash
ENCRYPTION_METHOD=fernet
ENCRYPTION_KEY=your_generated_key
```

--------------------------------

### Implement Custom SHA512 Blind Index Adapter

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Create a custom blind-index adapter by inheriting from BlindIndexAdapter and implementing compute_blind_index. Register it using register_blind_index_backend.

```python
class SHA512BlindIndex(BlindIndexAdapter):
    @classmethod
    def compute_blind_index(cls, value, key):
        import hashlib, hmac
        if isinstance(value, str):
            value = value.encode()
        return BlindIndexValue(hmac.new(key, value, hashlib.sha512).digest())

register_blind_index_backend(BlindIndexMethod.HMAC_SHA256, SHA512BlindIndex)
```

--------------------------------

### Configure AWS KMS Environment Variables

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Set the `ENCRYPTION_METHOD` to `aws` and provide the necessary AWS KMS configuration details as environment variables.

```bash
ENCRYPTION_METHOD=aws
AWS_KMS_KEY_ARN=arn:aws:kms:us-east-1:123456789:key/your-key-id
AWS_KMS_REGION=us-east-1
AWS_KMS_ACCESS_KEY_ID=your_access_key
AWS_KMS_SECRET_ACCESS_KEY=your_secret_key
```

--------------------------------

### Implement Custom XOR Encryption Adapter

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Define a custom encryption adapter by inheriting from EncryptionAdapter and implementing encrypt/decrypt methods. Register it using register_encryption_backend.

```python
class XORAdapter(EncryptionAdapter):
    KEY = b"\xAA" * 32

    @classmethod
    def encrypt(cls, plaintext, *, key=None):
        if isinstance(plaintext, EncryptedValue):
            return plaintext
        raw = plaintext.encode() if isinstance(plaintext, str) else plaintext
        return EncryptedValue(bytes(b ^ cls.KEY[i % 32] for i, b in enumerate(raw)))

    @classmethod
    def decrypt(cls, ciphertext, *, key=None):
        raw = bytes(ciphertext)
        return bytes(b ^ cls.KEY[i % 32] for i, b in enumerate(raw)).decode()

# Register against an existing enum value (overrides the built-in)
register_encryption_backend(EncryptionMethod.FERNET, XORAdapter)
```

```python
# Lazy registration (ImportError deferred until first use)
register_encryption_backend_lazy(
    EncryptionMethod.AWS,
    lambda: __import__("my_custom_kms").Adapter
)
```

--------------------------------

### FernetAdapter - Symmetric Encryption

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Demonstrates symmetric encryption and decryption using the Fernet algorithm. It shows both synchronous and asynchronous operations.

```APIDOC
## FernetAdapter - Symmetric Encryption

`FernetAdapter` implements `EncryptionAdapter` using `cryptography.fernet.Fernet`. Clients are cached per key. The key defaults to `settings.ENCRYPTION_KEY` but can be overridden per call.

```python
from pydantic_encryption import FernetAdapter
from cryptography.fernet import Fernet

key = Fernet.generate_key().decode()

ciphertext = FernetAdapter.encrypt("hello world", key=key)
print(repr(ciphertext))   # <EncryptedValue: 60 bytes>

plaintext = FernetAdapter.decrypt(ciphertext, key=key)
print(plaintext)          # "hello world"

# Async variants delegate to asyncio.to_thread
import asyncio
async def example():
    ct = await FernetAdapter.async_encrypt("async data", key=key)
    pt = await FernetAdapter.async_decrypt(ct, key=key)
    print(pt)              # "async data"
asyncio.run(example())
```
```

--------------------------------

### Custom Backend Registration

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Describes how to register custom encryption and blind index backends using `register_encryption_backend` and `register_blind_index_backend`.

```APIDOC
## register_encryption_backend / register_blind_index_backend — custom backend registration

The adapter registry maps `EncryptionMethod` and `BlindIndexMethod` enums to adapter classes. Register custom implementations to plug in alternative encryption or blind-indexing backends. Lazy (factory-based) registration is also supported to defer `ImportError` until first use.

```python
from pydantic_encryption import EncryptionAdapter, BlindIndexAdapter
from pydantic_encryption.adapters.registry import (
    register_encryption_backend,
    register_encryption_backend_lazy,
    register_blind_index_backend,
)
from pydantic_encryption.types import EncryptionMethod, BlindIndexMethod, EncryptedValue, BlindIndexValue
import os
```
```

--------------------------------

### Blind Index Normalization Options

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Configure normalization options for blind indexes to ensure consistent hashing. Options include `normalize_to_lowercase`, `strip_whitespace`, `strip_non_characters`, `strip_non_digits`, and `normalize_to_uppercase`.

```python
email_index: Annotated[bytes, BlindIndex(
    BlindIndexMethod.HMAC_SHA256,
    normalize_to_lowercase=True,
    strip_whitespace=True,
)]
```

--------------------------------

### Set Fernet Encryption Method

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Configure Fernet symmetric encryption by setting the `ENCRYPTION_METHOD` environment variable to `fernet`. This requires `ENCRYPTION_KEY` to be set as well.

```bash
ENCRYPTION_METHOD=fernet   # Fernet symmetric encryption (requires ENCRYPTION_KEY)
```

--------------------------------

### Per-Class Encryption Configuration with SecureModel

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use SecureModel to override environment variables for encryption_method and encryption_key on a per-class basis. Each model uses its own key independent of the ENCRYPTION_KEY env var.

```python
from typing import Annotated
from pydantic_encryption import BaseModel, Encrypted, EncryptionMethod

class InternalRecord(BaseModel, encryption_method=EncryptionMethod.FERNET, encryption_key="specific-fernet-key="):
    secret_data: Annotated[bytes, Encrypted]

class PublicRecord(BaseModel, encryption_method="fernet", encryption_key="other-fernet-key="):
    public_data: Annotated[bytes, Encrypted]

# Each model uses its own key independent of the ENCRYPTION_KEY env var
record = InternalRecord(secret_data="classified")
record.decrypt_data()
print(record.secret_data)  # "classified"
```

--------------------------------

### Async Decryption with DeferredDecryptMixin

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Use DeferredDecryptMixin to defer decryption of encrypted columns until the first read. This batches decryption across sibling instances in the same session for efficiency. Mix it into your SQLAlchemy models.

```python
from pydantic_encryption import DeferredDecryptMixin, SQLAlchemyEncryptedValue


class User(Base, DeferredDecryptMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())


Session = async_sessionmaker(engine, expire_on_commit=False)

async with Session() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

    # First read of `email` batch-decrypts it across every user in the session.
    for user in users:
        print(user.email)
```

--------------------------------

### SQLAlchemy Blind Index for Email

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Demonstrates using SQLAlchemyBlindIndexValue to store a normalized, keyed hash of an email alongside the encrypted email. Equality comparisons on the blind index automatically normalize the query value.

```python
import os
os.environ["BLIND_INDEX_SECRET_KEY"] = "my-secret-key"
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from pydantic_encryption import SQLAlchemyEncryptedValue, SQLAlchemyBlindIndexValue, BlindIndexMethod

class Base(DeclarativeBase): pass

class Customer(Base):
    __tablename__ = "customers"
    id:          Mapped[int]   = mapped_column(primary_key=True)
    email:       Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())
    email_index: Mapped[bytes] = mapped_column(
        SQLAlchemyBlindIndexValue(
            BlindIndexMethod.HMAC_SHA256,
            normalize_to_lowercase=True,
            strip_whitespace=True,
        )
    )

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

with Session(engine) as session:
    session.add(Customer(email="Alice@Example.COM", email_index="Alice@Example.COM"))
    session.commit()

with Session(engine) as session:
    # Query by normalized email — the blind index is recomputed on the filter value automatically
    found = session.execute(
        select(Customer).where(Customer.email_index == "alice@example.com")
    ).scalar_one_or_none()
    print(found.email)  # "Alice@Example.COM"  — decrypted from the encrypted column
```

--------------------------------

### SQLAlchemy Integration: Encrypted, Hashed, and Blind Indexed Fields

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Defines a SQLAlchemy model with fields for encryption, hashing, and blind indexing. The blind index is automatically hashed upon insertion and querying.

```python
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

from pydantic_encryption import (
    SQLAlchemyEncryptedValue,
    SQLAlchemyHashedValue,
    SQLAlchemyBlindIndexValue,
    BlindIndexMethod,
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]
    email: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())
    password: Mapped[bytes] = mapped_column(SQLAlchemyHashedValue())
    blind_index_email: Mapped[bytes] = mapped_column(
        SQLAlchemyBlindIndexValue(BlindIndexMethod.HMAC_SHA256)
    )


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

with Session(engine) as session:
    user = User(
        username="john",
        email="john@example.com",
        password="secret123",
        blind_index_email="john@example.com",
    )
    session.add(user)
    session.commit()

    # Query by blind index — automatically hashed
    found = session.query(User).filter(
        User.blind_index_email == "john@example.com"
    ).first()
    print(found.email)  # decrypted
```

--------------------------------

### Deferred Decryption with SQLAlchemy using DeferredDecryptMixin

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use DeferredDecryptMixin to automatically decrypt encrypted columns on first access. Columns not accessed remain encrypted. Batch decryption is triggered on first attribute access across all sibling rows in the session.

```python
import asyncio
import os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic_encryption import DeferredDecryptMixin, SQLAlchemyEncryptedValue

class Base(DeclarativeBase): pass

class User(Base, DeferredDecryptMixin):
    __tablename__ = "users"
    id:    Mapped[int]   = mapped_column(primary_key=True)
    email: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())
    name:  Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def main():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with Session() as session:
        session.add_all([
            User(email="a@example.com", name="Alice"),
            User(email="b@example.com", name="Bob"),
        ])
        await session.commit()

    async with Session() as session:
        users = (await session.execute(select(User))).scalars().all()
        # First access to `email` on any user triggers batch decrypt of ALL users' email column
        for u in users:
            print(u.email)  # decrypted — "a@example.com", "b@example.com"

        # Manual instance decrypt
        await users[0].decrypt()
        # Batch class-level decrypt
        await User.decrypt_many(users)

asyncio.run(main())
```

--------------------------------

### Set AWS KMS Encryption Method

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Configure AWS KMS encryption by setting the `ENCRYPTION_METHOD` environment variable to `aws`. This requires AWS credentials and the `AWS_KMS_KEY_ARN`.

```bash
ENCRYPTION_METHOD=aws      # AWS KMS (requires AWS_KMS_KEY_ARN, AWS_KMS_REGION, etc.)
```

--------------------------------

### Manual Decryption Helpers for SQLAlchemy Rows and Values

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Provides manual helpers for decrypting rows loaded outside a session or flat ciphertext lists. Supports single instance decryption, batch decryption for a class, decryption by row and column, and decryption of value lists.

```python
from pydantic_encryption import decrypt_rows, decrypt_values


async with AsyncSession(engine) as session:
    users = (await session.execute(select(User))).scalars().all()
    ciphertexts = [u.email for u in users]

    await users[0].decrypt()                              # one mixin instance
    await User.decrypt_many(users)                        # batch of one class
    await decrypt_rows(users, User.email)                 # InstrumentedAttribute or column names
    await decrypt_values(ciphertexts)                     # flat ciphertexts; preserves None positions
```

--------------------------------

### Argon2 Password Hashing with Argon2Adapter

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Hashes passwords using the Argon2 algorithm. The hashing process is idempotent, meaning hashing an already hashed value returns the same hash. Verification can be done directly using argon2-cffi.

```python
from pydantic_encryption import Argon2Adapter
from argon2 import PasswordHasher

hashed = Argon2Adapter.hash("my_password")
print(type(hashed))           # <class 'pydantic_encryption.types.HashedValue'>
print(hashed[:15])            # b'$argon2id$v=19$'

# Idempotent
hashed2 = Argon2Adapter.hash(hashed)
print(hashed == hashed2)      # True

# Verify using argon2-cffi directly
ph = PasswordHasher()
print(ph.verify(hashed, "my_password"))  # True
```

```python
# Async
import asyncio
async def async_hash():
    h = await Argon2Adapter.async_hash("async_password")
    return h
hashed_async = asyncio.run(async_hash())
```

--------------------------------

### Custom Encryption Logic

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Subclass `BaseModel` and override `encrypt_data` to implement custom encryption logic. The changes should be applied in-place to `self`.

```python
from pydantic_encryption import BaseModel

class MyModel(BaseModel):
    def encrypt_data(self) -> None:
        # your encryption logic (mutate self in-place)
        ...
```

--------------------------------

### Generate Fernet Key

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Generate a Fernet encryption key using Python. This key is required when `ENCRYPTION_METHOD` is set to `fernet`.

```bash
# Generate a key
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
```

--------------------------------

### AWSAdapter - AWS KMS Envelope Encryption

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Details AWS KMS envelope encryption using `AWSAdapter`. It covers synchronous and asynchronous encryption/decryption, and configuration for key rotation.

```APIDOC
## AWSAdapter - AWS KMS envelope encryption

`AWSAdapter` implements `EncryptionAdapter` using AWS KMS `GenerateDataKey` + AES-256-GCM. Each `encrypt` call generates a new data key; the wrapped data key and nonce are embedded in the ciphertext envelope. Async operations use `aioboto3` with a lazily-opened per-event-loop client. Requires `pip install "pydantic-encryption[aws]"`.

```python
import os
os.environ["ENCRYPTION_METHOD"] = "aws"
os.environ["AWS_KMS_KEY_ARN"] = "arn:aws:kms:us-east-1:123456789:key/your-key-id"
os.environ["AWS_KMS_REGION"] = "us-east-1"
os.environ["AWS_KMS_ACCESS_KEY_ID"] = "AKIAIOSFODNN7EXAMPLE"
os.environ["AWS_KMS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

from pydantic_encryption import AWSAdapter

# Sync — uses boto3; blocks the thread
ct = AWSAdapter.encrypt("sensitive payload")
pt = AWSAdapter.decrypt(ct)
print(pt)  # "sensitive payload"

# Async — uses aioboto3; yields the event loop during KMS call
import asyncio
async def kms_example():
    ct = await AWSAdapter.async_encrypt("async sensitive payload")
    pt = await AWSAdapter.async_decrypt(ct)
    print(pt)                          # "async sensitive payload"
    await AWSAdapter.aclose_async_kms()  # clean up when the event loop is shutting down

asyncio.run(kms_example())

# Split encrypt/decrypt keys for key rotation
os.environ.pop("AWS_KMS_KEY_ARN", None)
os.environ["AWS_KMS_ENCRYPT_KEY_ARN"] = "arn:aws:kms:...:new-key"
os.environ["AWS_KMS_DECRYPT_KEY_ARN"] = "arn:aws:kms:...:old-key"
# All old ciphertexts can still be decrypted; new ciphertexts use the new key
```
```

--------------------------------

### Safe Operations with EncryptedValue

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Demonstrates safe ways to interact with EncryptedValue objects, including checking encryption status and preventing dangerous coercions. Accidental leakage is prevented by raising errors on direct string conversion.

```python
print(repr(ct))           # <EncryptedValue: 4 bytes>
print(bytes(ct))          # b'\xde\xad\xbe\xef'  — raw ciphertext
print(is_encrypted(ct))   # True
print(is_encrypted("plain_text"))  # False

try:
    str(ct)
except EncryptedValueAccessError as e:
    print(e)  # "Encrypted value coerced to str before decryption..."

try:
    f"{ct}"
except EncryptedValueAccessError:
    print("f-string coercion also blocked")
```

```python
from pydantic_encryption import FernetAdapter
import os
os.environ["ENCRYPTION_KEY"] = "gAAAAAB..."  # real key required
result = FernetAdapter.encrypt(ct)  # returns ct unchanged, no double-encryption
```

--------------------------------

### Model-Level Encryption Configuration

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Override global encryption settings per model by passing `encryption_method` and `encryption_key` as keyword arguments during `BaseModel` definition. Falls back to environment variables if not set.

```python
from pydantic_encryption import BaseModel, Encrypted, EncryptionMethod
from typing import Annotated

class SpecialUser(BaseModel, encryption_method=EncryptionMethod.FERNET, encryption_key="my-key"):
    email: Annotated[bytes, Encrypted]
```

--------------------------------

### Configure Split AWS KMS Keys

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Optionally, configure separate encrypt and decrypt keys for AWS KMS by setting `AWS_KMS_ENCRYPT_KEY_ARN` and `AWS_KMS_DECRYPT_KEY_ARN`. This is useful for key rotation or read-only scenarios.

```bash
AWS_KMS_ENCRYPT_KEY_ARN=arn:aws:kms:...encrypt-key
AWS_KMS_DECRYPT_KEY_ARN=arn:aws:kms:...decrypt-key
```

--------------------------------

### Pydantic Model with Blind Index

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Define a Pydantic model with a field configured for blind indexing using `BlindIndex`. This enables equality searches on encrypted data by storing a deterministic keyed hash.

```python
from typing import Annotated
from pydantic_encryption import BaseModel, BlindIndex, BlindIndexMethod

class User(BaseModel):
    email_index: Annotated[bytes, BlindIndex(BlindIndexMethod.HMAC_SHA256)]
```

--------------------------------

### Fernet Symmetric Encryption with FernetAdapter

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Encrypts and decrypts data using Fernet symmetric encryption. Requires a Fernet key, which can be generated or provided. Async variants are available for non-blocking operations.

```python
from pydantic_encryption import FernetAdapter
from cryptography.fernet import Fernet

key = Fernet.generate_key().decode()

ciphertext = FernetAdapter.encrypt("hello world", key=key)
print(repr(ciphertext))   # <EncryptedValue: 60 bytes>

plaintext = FernetAdapter.decrypt(ciphertext, key=key)
print(plaintext)          # "hello world"
```

```python
# Async variants delegate to asyncio.to_thread
import asyncio
async def example():
    ct = await FernetAdapter.async_encrypt("async data", key=key)
    pt = await FernetAdapter.async_decrypt(ct, key=key)
    print(pt)              # "async data"
asyncio.run(example())
```

--------------------------------

### Manual Encryption and Hashing with Pydantic

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Annotate model fields with `Encrypted` or `Hashed` to automatically encrypt or hash data during model initialization. Use `Annotated` for type hinting.

```python
from typing import Annotated
from pydantic_encryption import BaseModel, Encrypted, Hashed

class User(BaseModel):
    name: str
    address: Annotated[bytes, Encrypted]
    password: Annotated[str, Hashed]

user = User(name="John Doe", address="123 Main St", password="secret123")

print(user.name)      # "John Doe"
print(user.address)   # encrypted bytes
print(user.password)  # argon2 hash bytes
```

--------------------------------

### Pre-warm deferred columns with decrypt_pending_fields

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use decrypt_pending_fields to decrypt all pending encrypted columns in a session concurrently before leaving the session context. This is useful for serializing data or when full decryption is required.

```python
import asyncio, os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from pydantic_encryption import decrypt_pending_fields, DeferredDecryptMixin, SQLAlchemyEncryptedValue
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass
class Item(Base, DeferredDecryptMixin):
    __tablename__ = "items"
    id:    Mapped[int]   = mapped_column(primary_key=True)
    value: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def get_items():
    async with Session() as session:
        items = (await session.execute(select(Item))).scalars().all()
        # Decrypt every encrypted column on every row before we leave this context
        await decrypt_pending_fields(session)
        # Safe to use outside a greenlet / async context
        return [{"id": i.id, "value": i.value} for i in items]

asyncio.run(get_items())
```

--------------------------------

### Normalize Input for Blind Indexes

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use `normalize_value` to apply stripping and case-normalization transforms to a string before blind-index computation. This function is used internally by `BlindIndex` and `SQLAlchemyBlindIndexValue` but can be called directly.

```python
from pydantic_encryption.normalization import normalize_value

# Strip leading/trailing and collapse internal whitespace
print(normalize_value("  hello   world  ", strip_whitespace=True))
# "hello world"

# Remove all non-letter characters
print(normalize_value("John O'Brien Jr.", strip_non_characters=True))
# "JohnOBrienJr"

# Remove all non-digit characters (phone normalization)
print(normalize_value("+1 (555) 123-4567", strip_non_digits=True))
# "15551234567"

# Lowercase + whitespace strip combined
print(normalize_value("  Alice@EXAMPLE.COM  ", normalize_to_lowercase=True, strip_whitespace=True))
# "alice@example.com"
```

--------------------------------

### Pre-decrypting Pending Fields in SQLAlchemy Session

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Use `decrypt_pending_fields(session)` to pre-decrypt all encrypted columns on loaded rows before exiting the session context. This is useful for serialization outside a greenlet spawn.

```python
from pydantic_encryption import decrypt_pending_fields

async with Session() as session:
    users = (await session.execute(select(User))).scalars().all()

    # Decrypt every encrypted column on every row loaded so far.
    await decrypt_pending_fields(session)

    payload = [{"id": u.id, "email": u.email} for u in users]
```

--------------------------------

### Handle Invalid Combination Raises ValueError

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Demonstrates how an invalid combination of arguments to normalize_value raises a ValueError. Ensure that strip_non_characters and strip_non_digits are not both set to True.

```python
try:
    normalize_value("test", strip_non_characters=True, strip_non_digits=True)
except ValueError as e:
    print(e)  # "strip_non_characters and strip_non_digits cannot both be True."
```

--------------------------------

### Finalizing SQLAlchemy Session with Decryption and Commit

Source: https://github.com/julien777z/pydantic-encryption/blob/main/README.md

Use `finalize_sqlalchemy_session(session)` to combine decryption of pending fields with a commit, releasing the pooled connection. This is convenient for read endpoints that might hold a DB connection during descriptor-driven KMS decryption.

```python
from pydantic_encryption import finalize_sqlalchemy_session

async with Session() as session:
    users = (await session.execute(select(User))).scalars().all()
    await finalize_sqlalchemy_session(session)  # decrypt pending + commit — connection released
    return [{"id": u.id, "email": u.email} for u in users]
```

--------------------------------

### Pydantic BaseModel with Automatic Field Processing

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Define Pydantic models with encrypted, hashed, or blind-indexed fields. Encryption, hashing, and blind-indexing occur automatically during initialization. Use `decrypt_data()` for sync decryption and `async_init`/`async_decrypt_data` for async operations.

```python
import asyncio
import os
from typing import Annotated
from pydantic_encryption import BaseModel, Encrypted, Hashed, BlindIndex, BlindIndexMethod, EncryptionMethod

os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="  # replace with real key
os.environ["BLIND_INDEX_SECRET_KEY"] = "my-blind-index-secret"

class User(BaseModel):
    name: str
    email: Annotated[bytes, Encrypted]
    password: Annotated[str, Hashed]
    email_index: Annotated[bytes, BlindIndex(
        BlindIndexMethod.HMAC_SHA256,
        normalize_to_lowercase=True,
        strip_whitespace=True,
    )]

# Sync construction — encryption, hashing, blind-indexing happen in model_post_init
user = User(name="Alice", email="alice@example.com", password="secret", email_index="alice@example.com")
print(user.name)         # "Alice"
print(type(user.email))  # <class 'pydantic_encryption.types.EncryptedValue'>
print(type(user.password))     # <class 'pydantic_encryption.types.HashedValue'>
print(type(user.email_index))  # <class 'pydantic_encryption.types.BlindIndexValue'>

# Sync decryption — in-place, returns self for chaining
user.decrypt_data()
print(user.email)        # "alice@example.com"

# Async construction (concurrent encrypt + hash + blind-index via asyncio.TaskGroup)
async def main():
    user = await User.async_init(
        name="Bob", email="bob@example.com", password="hunter2", email_index="bob@example.com"
    )
    await user.async_decrypt_data()
    print(user.email)    # "bob@example.com"

asyncio.run(main())
```

--------------------------------

### SQLAlchemy Transparent Encryption with SQLAlchemyEncryptedValue

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use SQLAlchemyEncryptedValue as a TypeDecorator for transparent encryption and decryption of various data types in SQLAlchemy models. Ensure encryption environment variables are set.

```python
import os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from pydantic_encryption import SQLAlchemyEncryptedValue
from datetime import date
from decimal import Decimal

class Base(DeclarativeBase):
    pass

class Record(Base):
    __tablename__ = "records"
    id:        Mapped[int]     = mapped_column(primary_key=True)
    name:      Mapped[bytes]   = mapped_column(SQLAlchemyEncryptedValue())
    score:     Mapped[bytes]   = mapped_column(SQLAlchemyEncryptedValue())
    birth_date:Mapped[bytes]   = mapped_column(SQLAlchemyEncryptedValue())
    amount:    Mapped[bytes]   = mapped_column(SQLAlchemyEncryptedValue())

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

with Session(engine) as session:
    session.add(Record(name="Alice", score=99, birth_date=date(1990, 5, 20), amount=Decimal("123.45")))
    session.commit()

with Session(engine) as session:
    record = session.execute(select(Record)).scalar_one()
    print(record.name)       # "Alice"        — str
    print(record.score)      # 99             — int
    print(record.birth_date) # date(1990, 5, 20)
    print(record.amount)     # Decimal('123.45')
```

--------------------------------

### One-Way Hashing with Hashed Annotation

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use the Hashed annotation for one-way Argon2 hashing. The hash is computed once on model creation, and the field stores a HashedValue. Already-hashed values are idempotent. Use argon2.PasswordHasher to verify passwords against the stored hash.

```python
from typing import Annotated
from pydantic_encryption import BaseModel, Hashed

class Account(BaseModel):
    username: str
    password: Annotated[str, Hashed]

account = Account(username="alice", password="super_secret_123")
print(account.password)          # b"$argon2id$v=19$m=65536,t=3,p=4$ early hash output
print(isinstance(account.password, bytes))  # True

# Verify a password against the stored hash
from argon2 import PasswordHasher
ph = PasswordHasher()
is_valid = ph.verify(account.password, "super_secret_123")
print(is_valid)  # True
```

--------------------------------

### AWS KMS Envelope Encryption with AWSAdapter

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Encrypts and decrypts data using AWS KMS envelope encryption. Requires AWS credentials and KMS key ARN configuration. Supports both synchronous and asynchronous operations.

```python
import os
os.environ["ENCRYPTION_METHOD"] = "aws"
os.environ["AWS_KMS_KEY_ARN"] = "arn:aws:kms:us-east-1:123456789:key/your-key-id"
os.environ["AWS_KMS_REGION"] = "us-east-1"
os.environ["AWS_KMS_ACCESS_KEY_ID"] = "AKIAIOSFODNN7EXAMPLE"
os.environ["AWS_KMS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

from pydantic_encryption import AWSAdapter

# Sync — uses boto3; blocks the thread
ct = AWSAdapter.encrypt("sensitive payload")
pt = AWSAdapter.decrypt(ct)
print(pt)  # "sensitive payload"
```

```python
# Async — uses aioboto3; yields the event loop during KMS call
import asyncio
async def kms_example():
    ct = await AWSAdapter.async_encrypt("async sensitive payload")
    pt = await AWSAdapter.async_decrypt(ct)
    print(pt)                          # "async sensitive payload"
    await AWSAdapter.aclose_async_kms()  # clean up when the event loop is shutting down

asyncio.run(kms_example())
```

```python
# Split encrypt/decrypt keys for key rotation
os.environ.pop("AWS_KMS_KEY_ARN", None)
os.environ["AWS_KMS_ENCRYPT_KEY_ARN"] = "arn:aws:kms:...:new-key"
os.environ["AWS_KMS_DECRYPT_KEY_ARN"] = "arn:aws:kms:...:old-key"
# All old ciphertexts can still be decrypted; new ciphertexts use the new key
```

--------------------------------

### finalize_sqlalchemy_session

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

`finalize_sqlalchemy_session` is a combined helper that commits the transaction (releasing the DB connection back to the pool) and then decrypts all pending rows. Ideal for read endpoints where KMS decryption should not hold a DB connection open.

```APIDOC
## `finalize_sqlalchemy_session(session)` — commit + batch decrypt

`finalize_sqlalchemy_session` is a combined helper that commits the transaction (releasing the DB connection back to the pool) and then decrypts all pending rows. Ideal for read endpoints where KMS decryption should not hold a DB connection open.

```python
import asyncio, os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from pydantic_encryption import finalize_sqlalchemy_session, DeferredDecryptMixin, SQLAlchemyEncryptedValue
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass
class Order(Base, DeferredDecryptMixin):
    __tablename__ = "orders"
    id:      Mapped[int]   = mapped_column(primary_key=True)
    payload: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def list_orders():
    async with Session() as session:
        orders = (await session.execute(select(Order))).scalars().all()
        # commit() releases the pool connection; decrypt happens afterward
        await finalize_sqlalchemy_session(session)
        return [{"id": o.id, "payload": o.payload} for o in orders]

asyncio.run(list_orders())
```
```

--------------------------------

### Commit and Batch Decrypt SQLAlchemy Session

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

Use `finalize_sqlalchemy_session` to commit a transaction and then decrypt all pending rows. This is ideal for read endpoints where KMS decryption should not hold a database connection open.

```python
import asyncio, os
os.environ["ENCRYPTION_METHOD"] = "fernet"
os.environ["ENCRYPTION_KEY"] = "gAAAAABkExampleKeyPadding1234567890123456="

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from pydantic_encryption import finalize_sqlalchemy_session, DeferredDecryptMixin, SQLAlchemyEncryptedValue
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass
class Order(Base, DeferredDecryptMixin):
    __tablename__ = "orders"
    id:      Mapped[int]   = mapped_column(primary_key=True)
    payload: Mapped[bytes] = mapped_column(SQLAlchemyEncryptedValue())

engine = create_async_engine("sqlite+aiosqlite:///:memory:")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def list_orders():
    async with Session() as session:
        orders = (await session.execute(select(Order))).scalars().all()
        # commit() releases the pool connection; decrypt happens afterward
        await finalize_sqlalchemy_session(session)
        return [{"id": o.id, "payload": o.payload} for o in orders]

asyncio.run(list_orders())
```

--------------------------------

### normalize_value

Source: https://context7.com/julien777z/pydantic-encryption/llms.txt

`normalize_value` applies stripping and case-normalization transforms to a string before blind-index computation. Used internally by `BlindIndex` annotations and `SQLAlchemyBlindIndexValue`, but can also be called directly.

```APIDOC
## `normalize_value` — input normalization for blind indexes

`normalize_value` applies stripping and case-normalization transforms to a string before blind-index computation. Used internally by `BlindIndex` annotations and `SQLAlchemyBlindIndexValue`, but can also be called directly.

```python
from pydantic_encryption.normalization import normalize_value

# Strip leading/trailing and collapse internal whitespace
print(normalize_value("  hello   world  ", strip_whitespace=True))
# "hello world"

# Remove all non-letter characters
print(normalize_value("John O'Brien Jr.", strip_non_characters=True))
# "JohnOBrienJr"

# Remove all non-digit characters (phone normalization)
print(normalize_value("+1 (555) 123-4567", strip_non_digits=True))
# "15551234567"

# Lowercase + whitespace strip combined
print(normalize_value("  Alice@EXAMPLE.COM  ", normalize_to_lowercase=True, strip_whitespace=True))
# "alice@example.com"

```
```