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[source]#

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[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

Delete a value. Handles both sync and async backends.

Parameters:

key¶ (str) – The key to delete.
key (str)

Return type:

None

async delete_all()[source]#

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

Parameters:: key¶ – The key to delete.
Return type:: None

async exists(key)[source]#

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)[source]#

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)[source]#

Return a new SQLAlchemyStore with a nested virtual key namespace.

Return type:: SQLAlchemyStore[TypeVar(SQLAlchemyConfigT, bound= Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig])]
Parameters:: namespace (str)