Adapters API Reference

Complete API reference for all available adapters.

PydanticAdapter

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.

RaiseOnMissingAdapter

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