Skip to content

Database adapters

The litestar_auth.db package exposes only the abstract persistence contracts and their lightweight data payloads: BaseUserStore, BaseOAuthAccountStore, BaseApiKeyStore, OAuthAccountData, and ApiKeyData. These protocols describe how the user manager talks to your storage layer without tying the library to a particular ORM.

The concrete SQLAlchemy implementations live in a dedicated submodule: import SQLAlchemyUserDatabase and SQLAlchemyApiKeyStore from litestar_auth.db.sqlalchemy. They are not re-exported from litestar_auth.db on purpose—eagerly importing the adapter would register SQLAlchemy mappers and break the lazy-import boundary described in the project guide. Use the submodule when you are ready to wire real tables.

For end-to-end ORM setup (session maker, models, plugin config), see User and manager and Backends; the Configuration index lists every split reference page. For customizing the user table while keeping OAuth accounts on the bundled model, see Custom user + OAuth.

from litestar_auth.db.sqlalchemy import SQLAlchemyUserDatabase
from litestar_auth.db.sqlalchemy import SQLAlchemyApiKeyStore

litestar_auth.db

Database abstractions and implementations.

ApiKeyData(key_id, user_id, hashed_secret, encrypted_secret, name, scopes, prefix_env, signing_required, expires_at, created_via, client_metadata=None) dataclass

Persistence fields required to create an API-key row.

BaseApiKeyStore

Bases: Protocol

Structural CRUD contract for API-key persistence backends.

create(data) async

Persist and return a newly created API key.

Source code in litestar_auth/db/base.py 
97
98
async def create(self, data: ApiKeyData[ID]) -> AK:
    """Persist and return a newly created API key."""

create_for_user_with_limit(data, *, max_keys_per_user) async

Persist an API key only when the user is still below the active-key limit.

Source code in litestar_auth/db/base.py 
100
101
async def create_for_user_with_limit(self, data: ApiKeyData[ID], *, max_keys_per_user: int) -> AK | None:
    """Persist an API key only when the user is still below the active-key limit."""

delete_for_user(user_id) async

Permanently delete all API-key rows for user_id.

Returns:

Type Description
int

Number of rows deleted when the backend can report it, otherwise 0.
Source code in litestar_auth/db/base.py 
109
110
111
112
113
114
async def delete_for_user(self, user_id: ID) -> int:
    """Permanently delete all API-key rows for ``user_id``.

    Returns:
        Number of rows deleted when the backend can report it, otherwise ``0``.
    """

get_by_key_id(key_id, *, include_inactive=False) async

Return an API key by public key id when present and active.

Source code in litestar_auth/db/base.py 
103
104
async def get_by_key_id(self, key_id: str, *, include_inactive: bool = False) -> AK | None:
    """Return an API key by public key id when present and active."""

list_for_user(user_id, *, include_inactive=False) async

Return API keys for a user, excluding revoked or expired rows by default.

Source code in litestar_auth/db/base.py 
106
107
async def list_for_user(self, user_id: ID, *, include_inactive: bool = False) -> list[AK]:
    """Return API keys for a user, excluding revoked or expired rows by default."""

list_signing_keys_requiring_reencrypt(requires_reencrypt, *, include_inactive=False) async

Return signing API-key rows whose encrypted secret needs keyring rotation.

Source code in litestar_auth/db/base.py 
125
126
127
128
129
130
131
async def list_signing_keys_requiring_reencrypt(
    self,
    requires_reencrypt: Callable[[AK], bool],
    *,
    include_inactive: bool = False,
) -> list[AK]:
    """Return signing API-key rows whose encrypted secret needs keyring rotation."""

replace_signing_key_encrypted_secret(key_id, *, encrypted_secret) async

Replace one signing API-key row's encrypted secret without changing other fields.

Source code in litestar_auth/db/base.py 
133
134
async def replace_signing_key_encrypted_secret(self, key_id: str, *, encrypted_secret: bytes) -> AK | None:
    """Replace one signing API-key row's encrypted secret without changing other fields."""

revoke(key_id, *, revoked_at) async

Soft-revoke an API key and return the updated row when present.

Source code in litestar_auth/db/base.py 
116
117
async def revoke(self, key_id: str, *, revoked_at: datetime) -> AK | None:
    """Soft-revoke an API key and return the updated row when present."""

update(key_id, *, name=None, scopes=None) async

Update mutable API-key metadata and return the updated active row.

Source code in litestar_auth/db/base.py 
119
120
async def update(self, key_id: str, *, name: str | None = None, scopes: list[str] | None = None) -> AK | None:
    """Update mutable API-key metadata and return the updated active row."""

update_last_used_at(key_id, *, last_used_at) async

Update the last-used timestamp for an active API key.

Source code in litestar_auth/db/base.py 
122
123
async def update_last_used_at(self, key_id: str, *, last_used_at: datetime) -> AK | None:
    """Update the last-used timestamp for an active API key."""

BaseOAuthAccountStore

Bases: Protocol

Structural contract for linked OAuth-account persistence backends.

get_by_oauth_account(oauth_name, account_id) async

Return a user linked to the given provider account, if present.

Source code in litestar_auth/db/base.py 
81
82
async def get_by_oauth_account(self, oauth_name: str, account_id: str) -> UP | None:
    """Return a user linked to the given provider account, if present."""

upsert_oauth_account(user, *, account) async

Create or update the linked OAuth account for user.

Source code in litestar_auth/db/base.py 
84
85
86
87
88
89
90
async def upsert_oauth_account(
    self,
    user: UP,
    *,
    account: OAuthAccountData,
) -> None:
    """Create or update the linked OAuth account for ``user``."""

BaseUserStore

Bases: Protocol

Structural CRUD interface for user persistence backends.

create(user_dict) async

Persist and return a newly created user.

Source code in litestar_auth/db/base.py 
64
65
async def create(self, user_dict: Mapping[str, Any]) -> UP:
    """Persist and return a newly created user."""

delete(user_id) async

Delete the user identified by user_id from storage.

Source code in litestar_auth/db/base.py 
73
74
async def delete(self, user_id: ID) -> None:
    """Delete the user identified by ``user_id`` from storage."""

get(user_id) async

Return the user with the given identifier, if present.

Source code in litestar_auth/db/base.py 
48
49
async def get(self, user_id: ID) -> UP | None:
    """Return the user with the given identifier, if present."""

get_by_email(email) async

Return the user matching the provided email, if present.

Source code in litestar_auth/db/base.py 
51
52
async def get_by_email(self, email: str) -> UP | None:
    """Return the user matching the provided email, if present."""

get_by_field(field_name, value) async

Return the user where field_name equals value, if present.

field_name must be "email" or "username" (see :data:~litestar_auth.types.LoginIdentifier). Implementations may perform a direct column/attribute lookup. Values outside that set are a programming error and may surface as backend-specific errors at runtime when callers bypass static typing.

Source code in litestar_auth/db/base.py 
54
55
56
57
58
59
60
61
62
async def get_by_field(self, field_name: LoginIdentifier, value: str) -> UP | None:
    """Return the user where ``field_name`` equals ``value``, if present.

    ``field_name`` must be ``"email"`` or ``"username"`` (see
    :data:`~litestar_auth.types.LoginIdentifier`). Implementations may perform a
    direct column/attribute lookup. Values outside that set are a
    programming error and may surface as backend-specific errors at
    runtime when callers bypass static typing.
    """

list_users(*, offset, limit) async

Return paginated users and the total available count.

Source code in litestar_auth/db/base.py 
67
68
async def list_users(self, *, offset: int, limit: int) -> tuple[list[UP], int]:
    """Return paginated users and the total available count."""

update(user, update_dict) async

Persist and return updates for an existing user.

Source code in litestar_auth/db/base.py 
70
71
async def update(self, user: UP, update_dict: Mapping[str, Any]) -> UP:
    """Persist and return updates for an existing user."""

OAuthAccountData(oauth_name, account_id, account_email, access_token, expires_at, refresh_token) dataclass

Provider account identity and token fields for OAuth-account persistence.