Custom adapter¶

Implement the protocols in fastauth.core.protocols to plug in your own ORM. You only implement what you use.

Minimum viable adapter: `UserAdapter`¶

UserAdapter is the only one required by every app. Implement these seven methods:

from fastauth.core.protocols import UserAdapter
from fastauth.types import UserData

class MyUserAdapter(UserAdapter):
    async def create_user(self, email, hashed_password=None, **kwargs) -> UserData: ...
    async def get_user_by_id(self, user_id) -> UserData | None: ...
    async def get_user_by_email(self, email) -> UserData | None: ...
    async def update_user(self, user_id, **kwargs) -> UserData: ...
    async def delete_user(self, user_id, soft=True) -> None: ...
    async def get_hashed_password(self, user_id) -> str | None: ...
    async def set_hashed_password(self, user_id, hashed_password) -> None: ...

UserData is a TypedDict:

class UserData(TypedDict):
    id: str
    email: str
    name: str | None
    email_verified: bool
    is_active: bool
    image: str | None

Email normalization¶

FastAuth normalizes every email — lowercase, strip whitespace — via fastauth.core.identity.normalize_email before passing it to your adapter. Your create_user and get_user_by_email are responsible for storing and querying that normalized form consistently. If you store mixed-case emails, lookups will silently fail.

from fastauth.core.identity import normalize_email

async def get_user_by_email(self, email):
    return await self.db.users.find_one({"email": normalize_email(email)})

When to implement which¶

Protocol	Implement when
`UserAdapter`	Always
`TokenAdapter`	Always (refresh tokens default to requiring it; magic links, verification, reset, email change, lockout all need it)
`SessionAdapter`	You want `/auth/sessions` endpoints
`RoleAdapter`	You want RBAC
`OAuthAccountAdapter`	You want OAuth
`PasskeyAdapter`	You want passkeys

`TokenAdapter` contract¶

class TokenAdapter(Protocol):
    async def create_token(self, token: TokenData) -> TokenData: ...
    async def get_token(self, token: str, token_type: str) -> TokenData | None: ...
    async def delete_token(self, token: str) -> None: ...
    async def delete_user_tokens(self, user_id: str, token_type: str | None = None) -> None: ...
    async def consume_token(self, token: str, token_type: str) -> TokenData | None: ...

consume_token is the load-bearing one. It must be atomic: pop the row (or row-equivalent) and return it only if it existed and was unexpired. If anything else (two concurrent refresh calls, a replay) races it, only one caller gets the row. The SQL adapter uses an in-transaction delete-then-return; the memory adapter uses a dict.pop. Implement this with a row lock, conditional delete, or a single atomic DELETE … RETURNING.

TokenData:

class TokenData(TypedDict):
    token: str
    user_id: str
    token_type: str
    expires_at: datetime
    raw_data: NotRequired[dict[str, Any] | None]

For password_reset, email_change, magic_link_login_request, and refresh_jti, the token field stores the SHA-256 hex of the raw token, never the raw token. Read the existing SQL adapter (adapters/sqlalchemy/token.py) for a model.

Wiring a custom adapter¶

config = FastAuthConfig(
    ...,
    adapter=MyUserAdapter(),
    token_adapter=MyTokenAdapter(),
    # oauth_adapter=MyOAuthAdapter(),      # only when used
)
auth = FastAuth(config)
auth.role_adapter = MyRoleAdapter()           # only when used
auth.session_adapter = MySessionAdapter()     # only when used

Reference¶

Full signatures: see Protocols reference (auto-generated) or read packages/fastauth/src/fastauth/core/protocols.py.