Metadata-Version: 2.4
Name: fastapi-user-manager
Version: 0.1.0
Summary: User management for FastAPI with SQLModel: registration, verification, password reset and OAuth accounts, using PASETO tokens.
Keywords: fastapi,sqlmodel,sqlalchemy,users,authentication,paseto,oauth
Author: Kohio Deflesselle
Author-email: Kohio Deflesselle <kohiodef@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Topic :: Security
Classifier: Topic :: Internet :: WWW/HTTP :: Session
Classifier: Typing :: Typed
Requires-Dist: auth-paseto>=0.1.0
Requires-Dist: fastapi>=0.143.0
Requires-Dist: pwdlib[argon2,bcrypt]
Requires-Dist: pydantic[email]>=2.14.1
Requires-Dist: sqlalchemy[asyncio]>=2.1.4
Requires-Dist: sqlmodel>=0.0.48
Requires-Python: >=3.12
Project-URL: Homepage, https://gitlab.com/Soralsei/user_manager
Project-URL: Source, https://gitlab.com/Soralsei/user_manager
Project-URL: Issues, https://gitlab.com/Soralsei/user_manager/-/issues
Description-Content-Type: text/markdown

# fastapi-user-manager

[![PyPI](https://img.shields.io/pypi/v/fastapi-user-manager)](https://pypi.org/project/fastapi-user-manager/)
[![Python](https://img.shields.io/pypi/pyversions/fastapi-user-manager)](https://pypi.org/project/fastapi-user-manager/)
[![License](https://img.shields.io/pypi/l/fastapi-user-manager)](https://gitlab.com/Soralsei/user_manager/-/blob/main/LICENSE)

User management for [FastAPI](https://fastapi.tiangolo.com/) applications backed by
[SQLModel](https://sqlmodel.tiangolo.com/) / SQLAlchemy (async). Registration, login, e-mail
verification, password reset and OAuth account linking, with verification and reset tokens issued
as [PASETO](https://paseto.io/) tokens via [auth-paseto](https://pypi.org/project/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](https://github.com/frankie567/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

```bash
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`:

```python
from user_manager import BaseUserManager
```

## Quick start

### 1. Define your models

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

```python
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.

```python
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](https://pypi.org/project/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`.

```python
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](#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:

```python
from sqlmodel import SQLModel

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

## The user manager

| Method | Description |
| :--- | :--- |
| `get(id)` / `get_by_email(email)` | Fetch a user, or raise `UserNotExists`. |
| `list(**filters)` | List users, optionally filtered, e.g. `list(is_active=True)`. |
| `create(user_create, safe=False)` | Validate the password, hash it and create the user. With `safe=True`, `is_superuser`, `is_active` and `is_verified` are ignored. |
| `update(user_update, user, safe=False)` | Update a user. Changing the e-mail resets `is_verified`; a new password is validated and hashed. |
| `delete(user)` | Delete a user. |
| `authenticate(credentials)` | Check an `OAuth2PasswordRequestForm`; return the user or `None`. Upgrades outdated password hashes. |
| `request_verify(user)` / `verify(payload)` | Issue a verification token; mark the user verified from a decoded token payload. |
| `forgot_password(user)` / `reset_password(payload, password)` | Issue a reset token; set a new password from a decoded token payload. |
| `oauth_callback(...)` / `oauth_associate_callback(user, ...)` | Create, update or link a user from an OAuth provider. See [OAuth accounts](#oauth-accounts). |

### Hooks

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

| Hook | Called |
| :--- | :--- |
| `validate_password(password, user)` | Before a password is set. Raise `InvalidPasswordException(reason)` to reject it. |
| `on_after_register(user)` | After a user is created. |
| `on_after_update(user, update_dict)` | After a user is updated. |
| `on_after_request_verify(user, token)` | After a verification token is issued. Deliver it here. |
| `on_after_verify(user)` | After a user is verified. |
| `on_after_forgot_password(user, token)` | After a reset token is issued. Deliver it here. |
| `on_after_reset_password(user)` | After a password is reset. |
| `on_after_login(user, response)` | Not called by the manager; call it from your login route if you need it. |
| `on_before_delete(user)` / `on_after_delete(user)` | Around a user deletion. |

Example password policy:

```python
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`.

| Exception | Raised when |
| :--- | :--- |
| `UserNotExists` | No user matches the given id, e-mail or OAuth account. |
| `UserAlreadyExists` | The e-mail is already taken (create, update, OAuth callback without `associate_by_email`). |
| `UserInactive` | The user is inactive (verification request, password reset). |
| `UserAlreadyVerified` | The user is already verified. |
| `InvalidVerifyToken` | The verification payload is malformed or its e-mail no longer matches the user's. |
| `InvalidResetPasswordToken` | The reset payload is malformed or the password changed since it was issued. |
| `InvalidPasswordException` | `validate_password` rejected the password; see `.reason`. |
| `InvalidID` | The ID could not be parsed. |

## 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.

```python
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](https://github.com/frankie567/httpx-oauth)) has
exchanged the authorization code, hand the result to the manager:

```python
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

```bash
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](https://github.com/fastapi-users/fastapi-users) and
its [SQLModel adapter](https://github.com/fastapi-users/fastapi-users-db-sqlmodel) by François
Voron, released under the MIT license.

## License

[MIT](LICENSE)
