Skip to content

Database adapters

litestar_auth.db

Database abstractions and implementations.

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 
55
56
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, *, oauth_name, account_id, account_email, access_token, expires_at, refresh_token) async

Create or update the linked OAuth account for user.

Source code in litestar_auth/db/base.py 
58
59
60
61
62
63
64
65
66
67
68
69
async def upsert_oauth_account(  # noqa: PLR0913
    self,
    user: UP,
    *,
    oauth_name: str,
    account_id: str,
    account_email: str,
    access_token: str,
    expires_at: int | None,
    refresh_token: str | None,
) -> None:
    """Create or update the linked OAuth account for ``user``."""

BaseUserStore

Bases: ABC

Abstract CRUD interface for user persistence backends.

create(user_dict) abstractmethod async

Persist and return a newly created user.

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

delete(user_id) abstractmethod async

Delete the user identified by user_id from storage.

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

get(user_id) abstractmethod async

Return the user with the given identifier, if present.

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

get_by_email(email) abstractmethod async

Return the user matching the provided email, if present.

Source code in litestar_auth/db/base.py 
21
22
23
@abstractmethod
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) abstractmethod async

Return the user where field_name equals value, if present.

Implementations may perform a direct column/attribute lookup. Invalid field_name values are a programming error and may surface as backend-specific errors at runtime.

Source code in litestar_auth/db/base.py 
25
26
27
28
29
30
31
32
@abstractmethod
async def get_by_field(self, field_name: str, value: str) -> UP | None:
    """Return the user where ``field_name`` equals ``value``, if present.

    Implementations may perform a direct column/attribute lookup. Invalid
    ``field_name`` values are a programming error and may surface as
    backend-specific errors at runtime.
    """

list_users(*, offset, limit) abstractmethod async

Return paginated users and the total available count.

Source code in litestar_auth/db/base.py 
38
39
40
@abstractmethod
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) abstractmethod async

Persist and return updates for an existing user.

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