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
DataclassAdapter	Type-safe storage/retrieval of dataclass models with transparent serialization
PydanticAdapter	Type-safe storage/retrieval of Pydantic models with transparent serialization
RaiseOnMissingAdapter	Optional raise-on-missing behavior for get operations

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"

PydanticAdapter¶

The PydanticAdapter provides type-safe storage and retrieval of Pydantic models. It automatically handles serialization and validation, ensuring data integrity.

PydanticAdapter ¶

Bases: BasePydanticAdapter[T]

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

_default_collection `instance-attribute` ¶

_default_collection = default_collection

_is_list_model `instance-attribute` ¶

_is_list_model = origin is list

_key_value `instance-attribute` ¶

_key_value = key_value

_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`	`type[T]`	The Pydantic model to use. Can be a single Pydantic model or list[Pydantic model].	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 data storage
Automatic validation on retrieval
Working with complex data models
Ensuring data integrity

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"

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¶

Adapters vs Wrappers¶

Adapter Details¶

DataclassAdapter¶

Use Cases¶

Basic Example¶

PydanticAdapter¶

PydanticAdapter ¶

_default_collection instance-attribute ¶

_is_list_model instance-attribute ¶

_key_value 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¶

Validation Error Handling¶

Default Collection¶

Batch Operations¶

TTL Support¶

Complex Models¶

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` ¶

_is_list_model `instance-attribute` ¶

_key_value `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` ¶