Adapters¶

Adapters provide specialized interfaces for working with key-value stores. Unlike wrappers, adapters don't implement the AsyncKeyValue protocol - instead, they provide alternative APIs tailored for specific use cases.

Available Adapters¶

Adapter	Description	Safety Level
DataclassAdapter	Type-safe storage/retrieval of dataclass models	✅ Safe (dataclass-only)
BaseModelAdapter	Type-safe storage/retrieval of Pydantic BaseModel instances	✅ Safe (BaseModel-only)
PydanticAdapter	Storage/retrieval of any pydantic-serializable type	⚠️ Less safe (any type)
RaiseOnMissingAdapter	Optional raise-on-missing behavior for get operations	N/A

Choosing the Right Adapter¶

The three primary adapters offer different levels of type safety:

DataclassAdapter: Use when working with Python dataclasses. Provides compile-time and runtime type safety for dataclass types only.
BaseModelAdapter: Use when working with Pydantic BaseModel subclasses. Provides compile-time and runtime type safety for BaseModel types only.
PydanticAdapter: Use when you need maximum flexibility to store any pydantic-serializable type (primitives, datetime, UUID, lists of primitives, etc.). Less type-safe but more flexible.

Adapters vs Wrappers¶

Wrappers:

Implement the AsyncKeyValue protocol
Can be stacked and used anywhere a store is expected
Add transparent functionality (compression, encryption, etc.)
Don't change the API

Adapters:

Provide a different API
Cannot be used in place of a store
Add type safety and specialized behavior
Transform how you interact with the store

Adapter Details¶

DataclassAdapter¶

The DataclassAdapter provides type-safe storage and retrieval of Python dataclass models. It automatically handles serialization and validation using Pydantic for validation.

Use Cases¶

Type-safe data storage with dataclasses
Automatic validation on retrieval
Working with Python's native dataclass decorator
Ensuring data integrity

Basic Example¶

from dataclasses import dataclass
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.dataclass import DataclassAdapter

@dataclass
class User:
    name: str
    email: str
    age: int

# Create adapter
adapter = DataclassAdapter(
    key_value=MemoryStore(),
    dataclass_type=User
)

# Store a user (type-safe)
user = User(name="Alice", email="alice@example.com", age=30)
await adapter.put(key="user:123", value=user, collection="users")

# Retrieve and get a validated model
retrieved_user = await adapter.get(key="user:123", collection="users")
if retrieved_user:
    print(retrieved_user.name)  # Type-safe: "Alice"
    print(retrieved_user.email)  # Type-safe: "alice@example.com"

BaseModelAdapter¶

The BaseModelAdapter provides type-safe storage and retrieval of Pydantic BaseModel instances. It's the recommended adapter for Pydantic models because it enforces type safety at both compile-time and runtime.

BaseModelAdapter ¶

Bases: BasePydanticAdapter[T]

Adapter around a KVStore-compliant Store that allows type-safe persistence of Pydantic BaseModels.

This adapter is constrained to BaseModel subclasses and sequences of BaseModels, providing a safe, type-checked interface for Pydantic model persistence.

_default_collection `instance-attribute` ¶

_default_collection = default_collection

_key_value `instance-attribute` ¶

_key_value = key_value

_needs_wrapping `instance-attribute` ¶

_needs_wrapping = origin is list

_raise_on_validation_error `instance-attribute` ¶

_raise_on_validation_error = raise_on_validation_error

_type_adapter `instance-attribute` ¶

_type_adapter = TypeAdapter[T](pydantic_model)

init ¶

__init__(
    key_value,
    pydantic_model,
    default_collection=None,
    raise_on_validation_error=False,
)

Create a new BaseModelAdapter.

Parameters:

Name	Type	Description	Default
`key_value`	`AsyncKeyValue`	The KVStore to use.	required
`pydantic_model`	`type[T]`	The Pydantic model to use. Can be a single BaseModel subclass or list[BaseModel].	required
`default_collection`	`str \| None`	The default collection to use.	`None`
`raise_on_validation_error`	`bool`	Whether to raise a DeserializationError if validation fails during reads. Otherwise, calls will return None if validation fails.	`False`

Raises:

Type	Description
`TypeError`	If pydantic_model is a sequence type other than list (e.g., tuple is not supported).

_get_model_type_name ¶

_get_model_type_name()

Return the model type name for error messages.

Use Cases¶

Type-safe storage of Pydantic BaseModel instances
When you want compile-time type checking for model types
Projects using Pydantic for data validation
Ensuring only BaseModel subclasses are stored

Basic Example¶

from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.base_model import BaseModelAdapter

class User(BaseModel):
    name: str
    email: str
    age: int

# Create adapter
adapter = BaseModelAdapter(
    key_value=MemoryStore(),
    pydantic_model=User
)

# Store a user (type-safe)
user = User(name="Alice", email="alice@example.com", age=30)
await adapter.put(key="user:123", value=user, collection="users")

# Retrieve and get a validated model
retrieved_user = await adapter.get(key="user:123", collection="users")
if retrieved_user:
    print(retrieved_user.name)  # Type-safe: "Alice"
    print(retrieved_user.email)  # Type-safe: "alice@example.com"

Storing Lists of Models¶

The BaseModelAdapter supports storing lists of BaseModel instances:

from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.base_model import BaseModelAdapter

class User(BaseModel):
    name: str
    email: str

# Create adapter for list of users
adapter = BaseModelAdapter(
    key_value=MemoryStore(),
    pydantic_model=list[User]
)

# Store a list of users
users = [
    User(name="Alice", email="alice@example.com"),
    User(name="Bob", email="bob@example.com"),
]
await adapter.put(key="all-users", value=users, collection="users")

# Retrieve the list
retrieved_users = await adapter.get(key="all-users", collection="users")
if retrieved_users:
    for user in retrieved_users:
        print(user.name)  # Type-safe access

Type Safety Benefits¶

The BaseModelAdapter enforces that only BaseModel subclasses can be used:

from pydantic import BaseModel
from key_value.aio.adapters.base_model import BaseModelAdapter

class User(BaseModel):
    name: str

# ✅ This works - User is a BaseModel
adapter = BaseModelAdapter(pydantic_model=User, key_value=store)

# ✅ This works - list[User] where User is a BaseModel
adapter = BaseModelAdapter(pydantic_model=list[User], key_value=store)

# ❌ This fails at runtime - int is not a BaseModel
adapter = BaseModelAdapter(pydantic_model=int, key_value=store)  # TypeError!

# ❌ This fails at runtime - list[int] inner type is not a BaseModel
adapter = BaseModelAdapter(pydantic_model=list[int], key_value=store)  # TypeError!

PydanticAdapter¶

The PydanticAdapter provides storage and retrieval of any pydantic-serializable type. Unlike BaseModelAdapter, it accepts primitives, collections, datetime objects, and more—not just BaseModel subclasses.

PydanticAdapter ¶

Bases: BasePydanticAdapter[T]

Adapter for persisting any pydantic-serializable type.

This is the "less safe" adapter that accepts any Python type that Pydantic can serialize. Unlike BaseModelAdapter (which is constrained to BaseModel types), this adapter can handle: - Pydantic BaseModel instances - Dataclasses (standard and Pydantic) - TypedDict - Primitive types (int, str, float, bool, etc.) - Collection types (list, dict, set, tuple, etc.) - Datetime and other common types

Types that serialize to dicts (BaseModel, dataclass, TypedDict, dict) are stored directly. Other types are wrapped in {"items": value} to ensure consistent dict-based storage.

_default_collection `instance-attribute` ¶

_default_collection = default_collection

_key_value `instance-attribute` ¶

_key_value = key_value

_needs_wrapping `instance-attribute` ¶

_needs_wrapping = _check_needs_wrapping()

_raise_on_validation_error `instance-attribute` ¶

_raise_on_validation_error = raise_on_validation_error

_type_adapter `instance-attribute` ¶

_type_adapter = TypeAdapter[T](pydantic_model)

init ¶

__init__(
    key_value,
    pydantic_model,
    default_collection=None,
    raise_on_validation_error=False,
)

Create a new PydanticAdapter.

Parameters:

Name	Type	Description	Default
`key_value`	`AsyncKeyValue`	The KVStore to use.	required
`pydantic_model`	`TypeForm[T]`	The type to serialize/deserialize. Can be any pydantic-serializable type.	required
`default_collection`	`str \| None`	The default collection to use.	`None`
`raise_on_validation_error`	`bool`	Whether to raise a DeserializationError if validation fails during reads. Otherwise, calls will return None if validation fails.	`False`

_check_needs_wrapping ¶

_check_needs_wrapping()

Check if a type needs to be wrapped in {"items": ...} for storage.

Types that serialize to dicts don't need wrapping. Other types do.

Returns:

Type	Description
`bool`	True if the type needs wrapping, False otherwise.

_get_model_type_name ¶

_get_model_type_name()

Return the model type name for error messages.

_serializes_to_dict ¶

_serializes_to_dict()

Check if a type serializes to a dict by inspecting the TypeAdapter's JSON schema.

This uses Pydantic's TypeAdapter.json_schema() to reliably determine the output structure. Types that produce a JSON object (schema type "object") are dict-serializable.

Uses a custom schema generator to skip fields that can't be represented in JSON schema (e.g., Callable fields), avoiding PydanticInvalidForJsonSchema errors.

Returns:

Type	Description
`bool`	True if the type serializes to a dict (JSON object), False otherwise.

Use Cases¶

Storing primitive types (int, str, float, bool)
Storing datetime, UUID, Decimal, and other common types
Storing lists of primitives (e.g., list[int], list[str])
Maximum flexibility when type constraints aren't needed
Working with any pydantic-serializable type

Basic Example¶

from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter

class User(BaseModel):
    name: str
    email: str
    age: int

# Create adapter
adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User
)

# Store a user (type-safe)
user = User(name="Alice", email="alice@example.com", age=30)
await adapter.put(key="user:123", value=user, collection="users")

# Retrieve and get a validated model
retrieved_user = await adapter.get(key="user:123", collection="users")
if retrieved_user:
    print(retrieved_user.name)  # Type-safe: "Alice"
    print(retrieved_user.email)  # Type-safe: "alice@example.com"

Storing Lists of Models¶

The PydanticAdapter supports storing lists of Pydantic models:

from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter

class User(BaseModel):
    name: str
    email: str

# Create adapter for list of users
adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=list[User]
)

# Store a list of users
users = [
    User(name="Alice", email="alice@example.com"),
    User(name="Bob", email="bob@example.com"),
]
await adapter.put(key="all-users", value=users, collection="users")

# Retrieve the list
retrieved_users = await adapter.get(key="all-users", collection="users")
if retrieved_users:
    for user in retrieved_users:
        print(user.name)  # Type-safe access

Validation Error Handling¶

By default, the adapter returns None if validation fails. You can configure it to raise an error instead:

from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.shared.errors import DeserializationError

adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User,
    raise_on_validation_error=True
)

# Manually corrupt data in the underlying store
await adapter._key_value.put(
    key="user:123",
    value={"name": "Alice"},  # Missing required 'email' field
    collection="users"
)

try:
    user = await adapter.get(key="user:123", collection="users")
except DeserializationError as e:
    print(f"Validation failed: {e}")

Default Collection¶

Set a default collection to avoid repeating it in every call:

adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User,
    default_collection="users"
)

# No need to specify collection
await adapter.put(key="user:123", value=user)
user = await adapter.get(key="user:123")

Batch Operations¶

The PydanticAdapter supports batch operations for better performance:

# Store multiple users
users = [
    User(name="Alice", email="alice@example.com", age=30),
    User(name="Bob", email="bob@example.com", age=25),
    User(name="Charlie", email="charlie@example.com", age=35),
]

await adapter.put_many(
    keys=["user:1", "user:2", "user:3"],
    values=users,
    collection="users"
)

# Retrieve multiple users
retrieved = await adapter.get_many(
    keys=["user:1", "user:2", "user:3"],
    collection="users"
)

for user in retrieved:
    if user:
        print(user.name)

TTL Support¶

The PydanticAdapter supports TTL for automatic expiration:

# Store with TTL
await adapter.put(
    key="session:abc",
    value=session_data,
    collection="sessions",
    ttl=3600  # Expires in 1 hour
)

# Get with TTL information
session, ttl = await adapter.ttl(key="session:abc", collection="sessions")
if session:
    print(f"Session expires in {ttl} seconds")

Complex Models¶

The PydanticAdapter works with complex nested models:

from pydantic import BaseModel
from datetime import datetime

class Address(BaseModel):
    street: str
    city: str
    country: str

class User(BaseModel):
    name: str
    email: str
    address: Address
    created_at: datetime

adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User
)

user = User(
    name="Alice",
    email="alice@example.com",
    address=Address(
        street="123 Main St",
        city="New York",
        country="USA"
    ),
    created_at=datetime.now()
)

await adapter.put(key="user:123", value=user, collection="users")
retrieved = await adapter.get(key="user:123", collection="users")

if retrieved:
    print(retrieved.address.city)  # Type-safe: "New York"

Storing Non-Model Types¶

The PydanticAdapter can store types that BaseModelAdapter cannot:

from datetime import datetime
from uuid import UUID
from key_value.aio.adapters.pydantic import PydanticAdapter

# ✅ Store primitives
int_adapter = PydanticAdapter(key_value=store, pydantic_model=int)
await int_adapter.put(key="count", value=42, collection="stats")

# ✅ Store datetime
datetime_adapter = PydanticAdapter(key_value=store, pydantic_model=datetime)
await datetime_adapter.put(key="timestamp", value=datetime.now(), collection="events")

# ✅ Store UUID
uuid_adapter = PydanticAdapter(key_value=store, pydantic_model=UUID)
await uuid_adapter.put(key="id", value=UUID("12345678-1234-5678-1234-567812345678"))

# ✅ Store list of primitives
list_adapter = PydanticAdapter(key_value=store, pydantic_model=list[int])
await list_adapter.put(key="scores", value=[10, 20, 30], collection="game")

Note: BaseModelAdapter would reject all of these types because they're not BaseModel subclasses.

BaseModelAdapter vs PydanticAdapter: When to Use Each¶

Use BaseModelAdapter When¶

✅ You're working exclusively with Pydantic BaseModel subclasses ✅ You want compile-time type safety ✅ You want runtime validation that only BaseModel types are stored ✅ You prefer strict type constraints

Example Use Case: A user management system where all entities are Pydantic models:

from pydantic import BaseModel, EmailStr
from key_value.aio.adapters.base_model import BaseModelAdapter

class User(BaseModel):
    id: int
    name: str
    email: EmailStr

class Organization(BaseModel):
    id: int
    name: str
    users: list[User]

# Type-safe: only User instances allowed
user_adapter = BaseModelAdapter[User](
    key_value=store,
    pydantic_model=User
)

# Type-safe: only Organization instances allowed
org_adapter = BaseModelAdapter[Organization](
    key_value=store,
    pydantic_model=Organization
)

# ✅ This works - User is a BaseModel
await user_adapter.put(key="user:1", value=User(id=1, name="Alice", email="alice@example.com"))

# ❌ This fails at type-check time - wrong type
# await user_adapter.put(key="user:1", value="not a user")

Use PydanticAdapter When¶

✅ You need to store primitive types (int, str, datetime, UUID, etc.) ✅ You need to store lists of primitives (list[int], list[str]) ✅ You're storing a mix of different types ✅ You need maximum flexibility ✅ Type constraints would be too restrictive

Example Use Case: A caching system that stores various data types:

from datetime import datetime
from uuid import UUID
from pydantic import BaseModel
from key_value.aio.adapters.pydantic import PydanticAdapter

# Different adapters for different types
session_adapter = PydanticAdapter[UUID](key_value=store, pydantic_model=UUID)
timestamp_adapter = PydanticAdapter[datetime](key_value=store, pydantic_model=datetime)
counter_adapter = PydanticAdapter[int](key_value=store, pydantic_model=int)
tags_adapter = PydanticAdapter[list[str]](key_value=store, pydantic_model=list[str])

# ✅ All of these work with PydanticAdapter
await session_adapter.put(key="session:abc", value=UUID("..."))
await timestamp_adapter.put(key="last-login:user:1", value=datetime.now())
await counter_adapter.put(key="views:page:1", value=1000)
await tags_adapter.put(key="tags:post:1", value=["python", "tutorial"])

# ❌ BaseModelAdapter would reject all of these (not BaseModel types)

Side-by-Side Comparison¶

from pydantic import BaseModel
from key_value.aio.adapters.base_model import BaseModelAdapter
from key_value.aio.adapters.pydantic import PydanticAdapter

class Product(BaseModel):
    name: str
    price: float

# BaseModelAdapter: Strict type safety
base_adapter = BaseModelAdapter[Product](
    key_value=store,
    pydantic_model=Product
)

# PydanticAdapter: Flexible
pydantic_adapter = PydanticAdapter[Product](
    key_value=store,
    pydantic_model=Product
)

# Both can store Product instances
product = Product(name="Widget", price=29.99)
await base_adapter.put(key="product:1", value=product)
await pydantic_adapter.put(key="product:1", value=product)

# But only PydanticAdapter can be used with non-BaseModel types:

# ✅ PydanticAdapter can do this
price_adapter = PydanticAdapter[float](key_value=store, pydantic_model=float)
await price_adapter.put(key="price:product:1", value=29.99)

# ❌ BaseModelAdapter cannot
# base_adapter = BaseModelAdapter[float](...) # TypeError at runtime!

Key Differences¶

Feature	BaseModelAdapter	PydanticAdapter
Accepted Types	Only `BaseModel` or `list[BaseModel]`	Any pydantic-serializable type
Type Safety	Compile-time + runtime validation	Runtime serialization only
Primitives	❌ No (int, str, float, etc.)	✅ Yes
DateTime/UUID	❌ No	✅ Yes
BaseModel	✅ Yes	✅ Yes
List of Primitives	❌ No	✅ Yes
Type Constraints	Strict	Flexible
Use Case	BaseModel-only projects	Mixed-type storage

RaiseOnMissingAdapter¶

The RaiseOnMissingAdapter changes the behavior of get operations to raise an error instead of returning None when a key is not found.

RaiseOnMissingAdapter ¶

Adapter around a KVStore that raises on missing values for get/get_many/ttl/ttl_many.

When raise_on_missing=True, methods raise MissingKeyError instead of returning None.

key_value `instance-attribute` ¶

key_value = key_value

init ¶

__init__(key_value)

delete `async` ¶

delete(key, *, collection=None)

Delete a key-value pair from the specified collection.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to delete the value from.	required
`collection`	`str \| None`	The collection to delete the value from. If no collection is provided, it will use the default collection.	`None`

delete_many `async` ¶

delete_many(keys, *, collection=None)

Delete multiple key-value pairs from the specified collection.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[str]`	The keys to delete the values from.	required
`collection`	`str \| None`	The collection to delete keys from. If no collection is provided, it will use the default collection.	`None`

Returns:

Type	Description
`int`	The number of keys deleted.

get `async` ¶

get(
    key: str,
    *,
    collection: str | None = None,
    raise_on_missing: Literal[False] = False,
) -> dict[str, Any] | None

get(
    key: str, *, collection: str | None = None, raise_on_missing: Literal[True]
) -> dict[str, Any]

get(key, *, collection=None, raise_on_missing=False)

Retrieve a value by key from the specified collection.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to retrieve the value from.	required
`collection`	`str \| None`	The collection to retrieve the value from. If no collection is provided, it will use the default collection.	`None`
`raise_on_missing`	`bool`	Whether to raise a MissingKeyError if the key is not found.	`False`

Returns:

Type	Description
`dict[str, Any] \| None`	The value associated with the key. If the key is not found, None will be returned.

get_many `async` ¶

get_many(
    keys: Sequence[str],
    *,
    collection: str | None = None,
    raise_on_missing: Literal[False] = False,
) -> list[dict[str, Any] | None]

get_many(
    keys: Sequence[str],
    *,
    collection: str | None = None,
    raise_on_missing: Literal[True],
) -> list[dict[str, Any]]

get_many(keys, *, collection=None, raise_on_missing=False)

Retrieve multiple values by key from the specified collection.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[str]`	The keys to retrieve the values from.	required
`collection`	`str \| None`	The collection to retrieve keys from. If no collection is provided, it will use the default collection.	`None`

Returns:

Type	Description
`list[dict[str, Any]] \| list[dict[str, Any] \| None]`	The values for the keys, or [] if the key is not found.

put `async` ¶

put(key, value, *, collection=None, ttl=None)

Store a key-value pair in the specified collection with optional TTL.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to store the value in.	required
`value`	`Mapping[str, Any]`	The value to store.	required
`collection`	`str \| None`	The collection to store the value in. If no collection is provided, it will use the default collection.	`None`
`ttl`	`SupportsFloat \| None`	The optional time-to-live (expiry duration) for the key-value pair. Defaults to no TTL. Note: The backend store will convert the provided format to its own internal format.	`None`

put_many `async` ¶

put_many(keys, values, *, collection=None, ttl=None)

Store multiple key-value pairs in the specified collection.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[str]`	The keys to store the values in.	required
`values`	`Sequence[Mapping[str, Any]]`	The values to store.	required
`collection`	`str \| None`	The collection to store keys in. If no collection is provided, it will use the default collection.	`None`
`ttl`	`SupportsFloat \| None`	The optional time-to-live (expiry duration) for all key-value pairs. The same TTL will be applied to all items in the batch. Defaults to no TTL. Note: The backend store will convert the provided format to its own internal format.	`None`

ttl `async` ¶

ttl(
    key: str,
    *,
    collection: str | None = None,
    raise_on_missing: Literal[False] = False,
) -> tuple[dict[str, Any] | None, float | None]

ttl(
    key: str, *, collection: str | None = None, raise_on_missing: Literal[True]
) -> tuple[dict[str, Any], float | None]

ttl(key, *, collection=None, raise_on_missing=False)

Retrieve the value and TTL information for a key-value pair from the specified collection.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key to retrieve the TTL information from.	required
`collection`	`str \| None`	The collection to retrieve the TTL information from. If no collection is provided, it will use the default collection.	`None`

Returns:

Type	Description
`tuple[dict[str, Any] \| None, float \| None]`	The value and TTL information for the key. If the key is not found, (None, None) will be returned.

ttl_many `async` ¶

ttl_many(
    keys: Sequence[str],
    *,
    collection: str | None = None,
    raise_on_missing: Literal[False] = False,
) -> list[tuple[dict[str, Any] | None, float | None]]

ttl_many(
    keys: Sequence[str],
    *,
    collection: str | None = None,
    raise_on_missing: Literal[True],
) -> list[tuple[dict[str, Any], float | None]]

ttl_many(keys, *, collection=None, raise_on_missing=False)

Retrieve multiple values and TTL information by key from the specified collection.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[str]`	The keys to retrieve the values and TTL information from.	required
`collection`	`str \| None`	The collection to retrieve keys from. If no collection is provided, it will use the default collection.	`None`

Use Cases¶

Enforcing required data
Fail-fast behavior
APIs where missing data is an error

Example¶

from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.raise_on_missing import RaiseOnMissingAdapter
from key_value.shared.errors import KeyNotFoundError

adapter = RaiseOnMissingAdapter(
    key_value=MemoryStore()
)

# Store a value
await adapter.put(key="user:123", value={"name": "Alice"}, collection="users")

# Get existing key - works normally
user = await adapter.get(key="user:123", collection="users")
print(user)  # {"name": "Alice"}

# Get missing key - raises error
try:
    user = await adapter.get(key="user:999", collection="users")
except KeyNotFoundError as e:
    print(f"Key not found: {e}")

Batch Operations¶

The RaiseOnMissingAdapter also affects batch operations:

# If any key is missing, raises KeyNotFoundError
try:
    users = await adapter.get_many(
        keys=["user:1", "user:999", "user:3"],
        collection="users"
    )
except KeyNotFoundError as e:
    print(f"One or more keys not found: {e}")

Combining Adapters and Wrappers¶

You can combine adapters with wrappers by wrapping the store before passing it to the adapter:

from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.encryption.fernet import FernetEncryptionWrapper
from key_value.aio.wrappers.compression import CompressionWrapper
from key_value.aio.adapters.pydantic import PydanticAdapter
from cryptography.fernet import Fernet

class User(BaseModel):
    name: str
    email: str

# Create encrypted + compressed store
wrapped_store = CompressionWrapper(
    key_value=FernetEncryptionWrapper(
        key_value=MemoryStore(),
        fernet=Fernet(Fernet.generate_key())
    )
)

# Wrap with PydanticAdapter for type safety
adapter = PydanticAdapter(
    key_value=wrapped_store,
    pydantic_model=User
)

# Now you have type-safe, encrypted, and compressed storage!
await adapter.put(key="user:123", value=User(name="Alice", email="alice@example.com"))

Creating Custom Adapters¶

To create a custom adapter, wrap an AsyncKeyValue instance and provide your own API:

from key_value.aio.protocols.key_value import AsyncKeyValue

class CustomAdapter:
    def __init__(self, key_value: AsyncKeyValue):
        self._key_value = key_value

    async def custom_method(self, key: str) -> dict:
        # Implement custom logic
        value = await self._key_value.get(key=key, collection="custom")
        if value is None:
            return {}
        return value

See the API Reference for complete adapter documentation.

Adapters¶

Available Adapters¶

Choosing the Right Adapter¶

Adapters vs Wrappers¶

Adapter Details¶

DataclassAdapter¶

Use Cases¶

Basic Example¶

BaseModelAdapter¶

BaseModelAdapter ¶

_default_collection instance-attribute ¶

_key_value instance-attribute ¶

_needs_wrapping instance-attribute ¶

_raise_on_validation_error instance-attribute ¶

_type_adapter instance-attribute ¶

__init__ ¶

_get_model_type_name ¶

Use Cases¶

Basic Example¶

Storing Lists of Models¶

Type Safety Benefits¶

PydanticAdapter¶

PydanticAdapter ¶

_default_collection instance-attribute ¶

_key_value instance-attribute ¶

_needs_wrapping instance-attribute ¶

_raise_on_validation_error instance-attribute ¶

_type_adapter instance-attribute ¶

__init__ ¶

_check_needs_wrapping ¶

_get_model_type_name ¶

_serializes_to_dict ¶

Use Cases¶

Basic Example¶

Storing Lists of Models¶

Validation Error Handling¶

Default Collection¶

Batch Operations¶

TTL Support¶

Complex Models¶

Storing Non-Model Types¶

BaseModelAdapter vs PydanticAdapter: When to Use Each¶

Use BaseModelAdapter When¶

Use PydanticAdapter When¶

Side-by-Side Comparison¶

Key Differences¶

RaiseOnMissingAdapter¶

RaiseOnMissingAdapter ¶

key_value instance-attribute ¶

__init__ ¶

delete async ¶

delete_many async ¶

get async ¶

get_many async ¶

put async ¶

put_many async ¶

ttl async ¶

ttl_many async ¶

Use Cases¶

Example¶

Batch Operations¶

Combining Adapters and Wrappers¶

Creating Custom Adapters¶

_default_collection `instance-attribute` ¶

_key_value `instance-attribute` ¶

_needs_wrapping `instance-attribute` ¶

_raise_on_validation_error `instance-attribute` ¶

_type_adapter `instance-attribute` ¶

init ¶

_default_collection `instance-attribute` ¶

_key_value `instance-attribute` ¶

_needs_wrapping `instance-attribute` ¶

_raise_on_validation_error `instance-attribute` ¶

_type_adapter `instance-attribute` ¶

init ¶

key_value `instance-attribute` ¶

init ¶

delete `async` ¶

delete_many `async` ¶

get `async` ¶

get_many `async` ¶

put `async` ¶

put_many `async` ¶

ttl `async` ¶

ttl_many `async` ¶