Skip to content

Database adapters

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

The concrete SQLAlchemy implementations live in a dedicated submodule: import SQLAlchemyUserDatabase, SQLAlchemyApiKeyStore, and SQLAlchemyOrganizationStore 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, Backends, and Organizations; 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
from litestar_auth.db.sqlalchemy import SQLAlchemyOrganizationStore

Organization persistence

Use BaseOrganizationStore[ORG, MEMBERSHIP, INVITATION, ID] for custom organization backends. The protocol supports organization create/get/get-by-slug operations, exact membership add/get/list/remove operations, atomic membership remove/role-update operations that preserve at least one privileged member, listing organizations for a user, and email-scoped invitation create/get/list/revoke/consume operations that store only a hashed token reference. Membership, user-organization, and pending-invitation list methods are paginated store calls: they accept keyword-only offset and limit arguments and return (items, total), where total is the count of the full filtered result set.

from litestar_auth.db import BaseOrganizationStore, MembershipData, OrganizationData, OrganizationInvitationData

Use SQLAlchemyOrganizationStore when the bundled SQLAlchemy adapter matches your model family:

from sqlalchemy.ext.asyncio import AsyncSession

from litestar_auth.db.sqlalchemy import SQLAlchemyOrganizationStore
from litestar_auth.models import Organization, OrganizationInvitation, OrganizationMembership


def create_store(session: AsyncSession) -> SQLAlchemyOrganizationStore:
    return SQLAlchemyOrganizationStore(
        session,
        organization_model=Organization,
        membership_model=OrganizationMembership,
        invitation_model=OrganizationInvitation,
    )

The adapter requires explicit organization_model and membership_model arguments. Invitation methods additionally require invitation_model, and fail with TypeError when it is omitted. Organization models remain lazy exports from litestar_auth.models; they are not exposed by litestar_auth.db.

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 
125
126
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 
128
129
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 
137
138
139
140
141
142
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 
131
132
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 
134
135
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 
153
154
155
156
157
158
159
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 
161
162
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 
144
145
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 
147
148
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 
150
151
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 
109
110
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 
112
113
114
115
116
117
118
async def upsert_oauth_account(
    self,
    user: UP,
    *,
    account: OAuthAccountData,
) -> None:
    """Create or update the linked OAuth account for ``user``."""

BaseOrganizationStore

Bases: Protocol

Structural CRUD contract for organization persistence backends.

add_membership(data) async

Persist and return a user's membership in an organization.

Source code in litestar_auth/db/base.py 
184
185
async def add_membership(self, data: MembershipData[ID]) -> MEMBERSHIP:
    """Persist and return a user's membership in an organization."""

consume_invitation(invitation_id, *, consumed_at) async

Atomically mark one pending invitation as consumed and return it when successful.

Source code in litestar_auth/db/base.py 
251
252
async def consume_invitation(self, invitation_id: ID, *, consumed_at: datetime) -> INVITATION | None:
    """Atomically mark one pending invitation as consumed and return it when successful."""

create_invitation(data) async

Persist and return a newly created organization invitation.

Source code in litestar_auth/db/base.py 
229
230
async def create_invitation(self, data: OrganizationInvitationData[ID]) -> INVITATION:
    """Persist and return a newly created organization invitation."""

create_organization(data) async

Persist and return a newly created organization.

Source code in litestar_auth/db/base.py 
169
170
async def create_organization(self, data: OrganizationData) -> ORG:
    """Persist and return a newly created organization."""

delete_organization(organization_id) async

Delete one organization and report whether a row was removed.

Source code in litestar_auth/db/base.py 
181
182
async def delete_organization(self, organization_id: ID) -> bool:
    """Delete one organization and report whether a row was removed."""

get_invitation(invitation_id) async

Return an invitation by primary identifier when present.

Source code in litestar_auth/db/base.py 
235
236
async def get_invitation(self, invitation_id: ID) -> INVITATION | None:
    """Return an invitation by primary identifier when present."""

get_invitation_by_token_hash(token_hash) async

Return an invitation by token digest when present.

Source code in litestar_auth/db/base.py 
232
233
async def get_invitation_by_token_hash(self, token_hash: bytes) -> INVITATION | None:
    """Return an invitation by token digest when present."""

get_membership(*, organization_id, user_id) async

Return the exact organization membership for user_id when present.

Source code in litestar_auth/db/base.py 
187
188
async def get_membership(self, *, organization_id: ID, user_id: ID) -> MEMBERSHIP | None:
    """Return the exact organization membership for ``user_id`` when present."""

get_organization(organization_id) async

Return an organization by primary identifier when present.

Source code in litestar_auth/db/base.py 
172
173
async def get_organization(self, organization_id: ID) -> ORG | None:
    """Return an organization by primary identifier when present."""

get_organization_by_slug(slug) async

Return an organization by normalized slug when present.

Source code in litestar_auth/db/base.py 
175
176
async def get_organization_by_slug(self, slug: str) -> ORG | None:
    """Return an organization by normalized slug when present."""

list_memberships(organization_id, *, offset, limit) async

Return paginated memberships for one organization and the total available count.

Source code in litestar_auth/db/base.py 
190
191
async def list_memberships(self, organization_id: ID, *, offset: int, limit: int) -> tuple[list[MEMBERSHIP], int]:
    """Return paginated memberships for one organization and the total available count."""

list_organizations_for_user(user_id, *, offset, limit) async

Return paginated organizations for user_id and the total available count.

Source code in litestar_auth/db/base.py 
226
227
async def list_organizations_for_user(self, user_id: ID, *, offset: int, limit: int) -> tuple[list[ORG], int]:
    """Return paginated organizations for ``user_id`` and the total available count."""

list_pending_invitations(organization_id, *, now, offset, limit) async

Return paginated unexpired pending invitations and the total available count.

Source code in litestar_auth/db/base.py 
238
239
240
241
242
243
244
245
246
async def list_pending_invitations(
    self,
    organization_id: ID,
    *,
    now: datetime,
    offset: int,
    limit: int,
) -> tuple[list[INVITATION], int]:
    """Return paginated unexpired pending invitations and the total available count."""

remove_membership(*, organization_id, user_id) async

Remove the exact organization membership and report whether a row was removed.

Source code in litestar_auth/db/base.py 
193
194
async def remove_membership(self, *, organization_id: ID, user_id: ID) -> bool:
    """Remove the exact organization membership and report whether a row was removed."""

remove_membership_preserving_privileged_member(*, organization_id, user_id, privileged_roles) async

Atomically remove a membership without removing the final privileged member.

Raises:

Type Description
ValueError

If removing the row would leave the organization without a privileged member.
Source code in litestar_auth/db/base.py 
196
197
198
199
200
201
202
203
204
205
206
207
async def remove_membership_preserving_privileged_member(
    self,
    *,
    organization_id: ID,
    user_id: ID,
    privileged_roles: frozenset[str],
) -> bool:
    """Atomically remove a membership without removing the final privileged member.

    Raises:
        ValueError: If removing the row would leave the organization without a privileged member.
    """

revoke_invitation(invitation_id) async

Mark a pending invitation as revoked and return the updated row.

Source code in litestar_auth/db/base.py 
248
249
async def revoke_invitation(self, invitation_id: ID) -> INVITATION | None:
    """Mark a pending invitation as revoked and return the updated row."""

set_membership_roles(*, organization_id, user_id, roles) async

Replace roles on an existing organization membership and return the updated row.

Source code in litestar_auth/db/base.py 
209
210
async def set_membership_roles(self, *, organization_id: ID, user_id: ID, roles: list[str]) -> MEMBERSHIP | None:
    """Replace roles on an existing organization membership and return the updated row."""

set_membership_roles_preserving_privileged_member(*, organization_id, user_id, roles, privileged_roles) async

Atomically replace roles without demoting the final privileged member.

Raises:

Type Description
ValueError

If replacing roles would leave the organization without a privileged member.
Source code in litestar_auth/db/base.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
async def set_membership_roles_preserving_privileged_member(
    self,
    *,
    organization_id: ID,
    user_id: ID,
    roles: list[str],
    privileged_roles: frozenset[str],
) -> MEMBERSHIP | None:
    """Atomically replace roles without demoting the final privileged member.

    Raises:
        ValueError: If replacing roles would leave the organization without a privileged member.
    """

update_organization(organization_id, data) async

Persist mutable organization fields and return the updated row when present.

Source code in litestar_auth/db/base.py 
178
179
async def update_organization(self, organization_id: ID, data: OrganizationData) -> ORG | None:
    """Persist mutable organization fields and return the updated row when present."""

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 
92
93
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 
101
102
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 
76
77
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 
79
80
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 
82
83
84
85
86
87
88
89
90
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 
95
96
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 
98
99
async def update(self, user: UP, update_dict: Mapping[str, Any]) -> UP:
    """Persist and return updates for an existing user."""

MembershipData(organization_id, user_id, roles) dataclass

Persistence fields required to create an organization membership row.

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

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

OrganizationData(slug, name) dataclass

Persistence fields required to create an organization row.

OrganizationInvitationData(organization_id, invited_email, roles, token_hash, expires_at) dataclass

Persistence fields required to create an organization invitation row.