Store

This module provides a SQLAlchemy-backed implementation of Litestar’s Store protocol. Register a SQLAlchemyStore under the "sessions" store name to back Litestar’s server-side session middleware with your SQLAlchemy database.

Store Model Mixin

class advanced_alchemy.extensions.litestar.store.StoreModelMixin

Bases: UUIDv7Base

Mixin for session storage.

is_expired

SQL-Expression to check if the session has expired.

Returns:

SQL-Expression to check if the session has expired.

SQLAlchemy Store

class advanced_alchemy.extensions.litestar.store.SQLAlchemyStore

Bases: NamespacedStore, Generic[SQLAlchemyConfigT]

SQLAlchemy based, thread and process safe asynchronous key/value store.

Supports both synchronous and asynchronous SQLAlchemy configurations.

__init__(config, model=StoreModelMixin, namespace=Empty)

Initialize SQLAlchemyStore.

Parameters:

• config – An instance of SQLAlchemyAsyncConfig or SQLAlchemySyncConfig.

• model – The SQLAlchemy model to use for storing data. Defaults to StoreItem.

• namespace – A virtual namespace for keys. If not given, defaults to LITESTAR. Namespacing can be explicitly disabled by passing None. This will make delete_all() unavailable.

async set(key, value, expires_in=None)

Set a value. Handles both sync and async backends.

Return type:

None

Parameters:

• key (str)

• value (bytes | str)

• expires_in (int | timedelta | None)

async get(key, renew_for=None)

Get a value. Handles both sync and async backends.

Parameters:

• key (str) – The key to get the value for.

• renew_for (Union[int, timedelta, None]) – The amount of time to renew the value for.

• key (str)

• renew_for (int | timedelta | None)

Return type:

Optional[bytes]

Returns:

The value of the key, or None if the key does not exist or has expired.

async delete(key)

Delete a value. Handles both sync and async backends.

Parameters:

• key (str) – The key to delete.

• key (str)

Return type:

None

async delete_all()

Delete all values in the namespace. Handles both sync and async backends.

Parameters:

key – The key to delete.

Return type:

None

async exists(key)

Check if a key exists. Handles both sync and async backends.

Parameters:

• key (str) – The key to check if it exists.

• key (str)

Return type:

bool

Returns:

True if the key exists, False otherwise.

async expires_in(key)

Get expiration time. Handles both sync and async backends.

Parameters:

• key (str) – The key to get the expiration time for.

• key (str)

Return type:

Optional[int]

Returns:

The expiration time in seconds, or None if the key does not exist or has no expiration time.

with_namespace(namespace)

Return a new SQLAlchemyStore with a nested virtual key namespace.

Return type:

SQLAlchemyStore[TypeVar(SQLAlchemyConfigT, bound= Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig])]

Parameters:

namespace (str)