fastapi-user-manager

User management for FastAPI applications backed by SQLModel / SQLAlchemy (async). Registration, login, e-mail verification, password reset and OAuth account linking, with verification and reset tokens issued as PASETO tokens via auth-paseto.

It gives you the building blocks — models, a database adapter and a UserManager with lifecycle hooks — and leaves routing, e-mail delivery and session handling to you.

Features

• Ready-made SQLModel tables for users and OAuth accounts, with UUID or integer primary keys.
• Async database adapter on top of sqlmodel.ext.asyncio.session.AsyncSession.
• Secure password handling with pwdlib: Argon2 by default, bcrypt hashes still accepted and transparently upgraded on login.
• E-mail verification and password reset tokens. Reset tokens are bound to the current password hash, so they stop working as soon as the password changes.
• OAuth account linking: create or update users from an OAuth provider callback.
• Lifecycle hooks (on_after_register, on_after_forgot_password, …) to send e-mails, write audit logs, etc.
• Typed, with a py.typed marker.

Installation

pip install fastapi-user-manager
# or
uv add fastapi-user-manager

Requires Python 3.12+. You also need an async driver for your database, e.g. asyncpg for PostgreSQL or aiosqlite for SQLite. Form-based login (OAuth2PasswordRequestForm) also needs python-multipart.

The package is imported as user_manager:

from user_manager import BaseUserManager

Quick start

1. Define your models

Subclass the base table and schemas. Add your own columns as needed.

import uuid

from user_manager.db.user import SQLModelBaseUserUUID
from user_manager.schemas import BaseUser, BaseUserCreate, BaseUserUpdate


class User(SQLModelBaseUserUUID, table=True):
    pass


class UserRead(BaseUser[uuid.UUID]):
    pass


class UserCreate(BaseUserCreate):
    pass


class UserUpdate(BaseUserUpdate):
    pass

The user table gets id, email (unique, matched case-insensitively), hashed_password, is_active, is_superuser and is_verified columns. Use SQLModelBaseUserInt for an auto-incrementing integer id instead.

2. Create your user manager

Pick the ID mixin matching your model (UUIDIDMixin or IntegerIDMixin) and override the hooks you need.

from fastapi import Request

from user_manager import BaseUserManager
from user_manager.core.base_manager import UUIDIDMixin


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    async def on_after_register(self, user: User, request: Request | None = None) -> None:
        print(f"User {user.id} has registered.")

    async def on_after_request_verify(
        self, user: User, token: str, request: Request | None = None
    ) -> None:
        # Send `token` to the user, e.g. as a link in an e-mail.
        print(f"Verification token for {user.email}: {token}")

    async def on_after_forgot_password(
        self, user: User, token: str, request: Request | None = None
    ) -> None:
        print(f"Reset token for {user.email}: {token}")

3. Wire it into FastAPI

The manager needs two auth-paseto instances: one for e-mail verification tokens and one for password reset tokens. Give each its own secret key, distinct from the one you use for login tokens, so one kind of token can never be accepted as another. Their lifetime is the instance's access_token_expires.

import os
import uuid
from collections.abc import AsyncIterator
from datetime import timedelta
from typing import Annotated

from auth_paseto import AuthConfig, PASETOAuth
from fastapi import Body, Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession

from user_manager.db.user import SQLModelUserDatabase
from user_manager.exceptions import UserAlreadyExists

engine = create_async_engine(os.environ["DATABASE_URL"])

login_auth = PASETOAuth(AuthConfig(secret_key=os.environ["LOGIN_SECRET_KEY"]))
verify_auth = PASETOAuth(
    AuthConfig(
        secret_key=os.environ["VERIFY_SECRET_KEY"],
        access_token_expires=timedelta(days=1),
    )
)
reset_auth = PASETOAuth(
    AuthConfig(
        secret_key=os.environ["RESET_SECRET_KEY"],
        access_token_expires=timedelta(hours=1),
    )
)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session


async def get_user_manager(
    session: Annotated[AsyncSession, Depends(get_session)],
) -> UserManager:
    return UserManager(SQLModelUserDatabase(session, User), verify_auth, reset_auth)


Manager = Annotated[UserManager, Depends(get_user_manager)]

app = FastAPI()


@app.post("/auth/register", response_model=UserRead, status_code=201)
async def register(user_create: UserCreate, manager: Manager):
    try:
        # safe=True ignores is_superuser / is_verified / is_active from the request body
        return await manager.create(user_create, safe=True)
    except UserAlreadyExists:
        raise HTTPException(status_code=400, detail="REGISTER_USER_ALREADY_EXISTS")


@app.post("/auth/login")
async def login(
    credentials: Annotated[OAuth2PasswordRequestForm, Depends()], manager: Manager
):
    user = await manager.authenticate(credentials)
    if user is None or not user.is_active:
        raise HTTPException(status_code=400, detail="LOGIN_BAD_CREDENTIALS")
    return {
        "access_token": login_auth.create_access_token(subject=str(user.id)),
        "token_type": "bearer",
    }


@app.post("/auth/request-verify-token", status_code=202)
async def request_verify_token(email: Annotated[str, Body(embed=True)], manager: Manager):
    user = await manager.get_by_email(email)
    await manager.request_verify(user)  # calls on_after_request_verify with the token
    return None


@app.post("/auth/verify", response_model=UserRead)
async def verify(token: Annotated[str, Body(embed=True)], manager: Manager):
    payload = await verify_auth.get_fresh_access_token_payload(token)
    return await manager.verify(payload)


@app.post("/auth/forgot-password", status_code=202)
async def forgot_password(email: Annotated[str, Body(embed=True)], manager: Manager):
    user = await manager.get_by_email(email)
    await manager.forgot_password(user)  # calls on_after_forgot_password with the token
    return None


@app.post("/auth/reset-password")
async def reset_password(
    token: Annotated[str, Body()], password: Annotated[str, Body()], manager: Manager
):
    payload = await reset_auth.get_fresh_access_token_payload(token)
    await manager.reset_password(payload, password)
    return None

The manager raises the exceptions listed under Exceptions rather than HTTP errors, and auth-paseto raises TokenVerificationError for bad tokens. Map them to responses with try/except as above or with @app.exception_handler. For example, a production forgot-password route should catch UserNotExists and still return 202, so it does not reveal which e-mails are registered.

Create the tables at startup or, preferably, with a migration tool such as Alembic:

from sqlmodel import SQLModel

async with engine.begin() as conn:
    await conn.run_sync(SQLModel.metadata.create_all)

The user manager

Hooks

Override any of these async methods on your manager. They all receive the optional FastAPI request that triggered the operation.

Example password policy:

from user_manager.exceptions import InvalidPasswordException


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    async def validate_password(self, password: str, user) -> None:
        if len(password) < 8:
            raise InvalidPasswordException(reason="Password should be at least 8 characters")
        if user.email in password:
            raise InvalidPasswordException(reason="Password should not contain e-mail")

Exceptions

All live in user_manager.exceptions and inherit from FastAPIUsersException.

OAuth accounts

To store OAuth accounts, add an OAuth table with relationships in both directions and use SQLModelOauthUserDatabase. Loading oauth_accounts eagerly (lazy="selectin") is required with async sessions.

from sqlmodel import Relationship

from user_manager.db.user import (
    SQLModelBaseOAuthAccountUUID,
    SQLModelBaseUserUUID,
    SQLModelOauthUserDatabase,
)


class OAuthAccount(SQLModelBaseOAuthAccountUUID, table=True):
    user: "User" = Relationship(back_populates="oauth_accounts")


class User(SQLModelBaseUserUUID, table=True):
    oauth_accounts: list[OAuthAccount] = Relationship(
        back_populates="user", sa_relationship_kwargs={"lazy": "selectin"}
    )


user_db = SQLModelOauthUserDatabase(session, User, OAuthAccount)

After your OAuth client (e.g. httpx-oauth) has exchanged the authorization code, hand the result to the manager:

user = await manager.oauth_callback(
    oauth_name="google",
    access_token=token["access_token"],
    account_id=account_id,
    account_email=account_email,
    expires_at=token.get("expires_at"),
    refresh_token=token.get("refresh_token"),
    associate_by_email=True,
    is_verified_by_default=True,
)

• An existing account for this provider and account_id is updated with the new tokens.
• Otherwise, if a user with account_email exists, the account is linked to them when associate_by_email=True, and UserAlreadyExists is raised when it is False.
• Otherwise a new user is created with a random password.

Only enable associate_by_email and is_verified_by_default for providers that verify e-mail addresses; otherwise anyone controlling an account with a victim's e-mail could take over their account.

Custom databases

To use another storage backend, subclass user_manager.BaseUserDatabase and implement get, get_by_email, create, update and delete (plus list and the OAuth methods if you need them). Your user objects must have the attributes described by user_manager.models.UserProtocol.

Development

git clone https://gitlab.com/Soralsei/user_manager.git
cd user_manager
uv sync
uv run pytest

Acknowledgements

This project is derived from FastAPI Users and its SQLModel adapter by François Voron, released under the MIT license.

License

MIT