FastAPI Users database adapter for SQLAlchemy

Warnings

• breaking Python 3.8 support was dropped in `fastapi-users-db-sqlalchemy` v7.0.0. Ensure your project uses Python 3.9 or higher.
  Fix: Upgrade your Python environment to 3.9+ or pin `fastapi-users-db-sqlalchemy<7.0.0` if you must remain on Python 3.8.
• breaking Version 5.0.0 migrated to SQLAlchemy 2.0. This introduces significant API changes (e.g., `Mapped` for columns, `select()` statements instead of `query()`). If you need to use SQLAlchemy 1.4, you must pin the dependency.
  Fix: Migrate your SQLAlchemy models and query patterns to SQLAlchemy 2.0 style or pin `fastapi-users-db-sqlalchemy<5.0.0` if you need to stay on SQLAlchemy 1.4.
• gotcha When using asynchronous SQLAlchemy, relationships are not implicitly lazy-loaded. Attempting to access unloaded relationships in generated FastAPI Users routes (e.g., in a Pydantic `UserRead` model) will result in `MissingGreenlet` errors or similar issues due to implicit I/O. Eager loading must be explicitly configured.
  Fix: Use SQLAlchemy's eager loading options (e.g., `selectinload`, `joinedload`) for relationships that need to be accessed when fetching users via `fastapi-users`'s underlying database operations. Ensure your `UserRead` schemas align with what's eagerly loaded.
• gotcha FastAPI path operation functions should receive Pydantic models as input parameters for data validation and serialization, not raw SQLAlchemy ORM models.
  Fix: Define separate Pydantic schemas (e.g., `UserCreate`, `UserUpdate`) for input data to your FastAPI endpoints. Use these Pydantic schemas for request body validation and then convert them to SQLAlchemy models for database operations.
• gotcha Ensure you install an appropriate asynchronous database driver (e.g., `aiosqlite` for SQLite, `asyncpg` for PostgreSQL) corresponding to your `DATABASE_URL` dialect. Failure to do so will result in connection errors.
  Fix: Install the correct driver: `pip install aiosqlite` or `pip install asyncpg`. The `DATABASE_URL` should then use `sqlite+aiosqlite://` or `postgresql+asyncpg://`.

Install

• pip install fastapi-users-db-sqlalchemy sqlalchemy aiosqlite With SQLite support
• pip install fastapi-users-db-sqlalchemy sqlalchemy asyncpg With PostgreSQL support

Imports

• SQLAlchemyUserDatabase

  from fastapi_users.db import SQLAlchemyUserDatabase

• SQLAlchemyBaseUserTable

  from fastapi_users.db import SQLAlchemyBaseUserTable

• SQLAlchemyBaseUserTableUUID

  from fastapi_users.db import SQLAlchemyBaseUserTableUUID

• AsyncSession

  from sqlalchemy.ext.asyncio import AsyncSession

• DeclarativeBase

  from sqlalchemy.orm import DeclarativeBase

  Use DeclarativeBase for SQLAlchemy 2.0 style models; declarative_base is for SQLAlchemy 1.x or legacy modes.

Quickstart

This quickstart demonstrates how to set up the `fastapi-users-db-sqlalchemy` adapter with a SQLAlchemy 2.0-style User model and asynchronous database session management for FastAPI. It includes defining the User model, database engine, session maker, and the necessary dependencies (`get_async_session`, `get_user_db`). Remember to install an async database driver like `aiosqlite` or `asyncpg`.

import os
from typing import AsyncGenerator

from fastapi import Depends, FastAPI
from fastapi_users import FastAPIUsers
from fastapi_users.authentication import CookieAuthentication, AuthenticationBackend
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String

# Define your database URL (using SQLite for this example)
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///./test.db")
SECRET = os.environ.get("SECRET", "YOUR_SECRET_KEY") # IMPORTANT: Change in production!

# 1. Define your SQLAlchemy Base
class Base(DeclarativeBase):
    pass

# 2. Define your User model
class User(SQLAlchemyBaseUserTableUUID, Base):
    # You can add custom fields here
    first_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
    last_name: Mapped[str | None] = mapped_column(String(255), nullable=True)

# 3. Create SQLAlchemy engine and session maker
engine = create_async_engine(DATABASE_URL)
async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# 4. Utility function to create database tables
async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 5. Database session dependency
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session

# 6. User database adapter dependency
async def get_user_db(
    session: AsyncSession = Depends(get_async_session),
) -> AsyncGenerator[SQLAlchemyUserDatabase, None]:
    yield SQLAlchemyUserDatabase(session, User)

# 7. Authentication Backend (example with Cookie Authentication)
cookie_authentication = CookieAuthentication(secret=SECRET, lifetime_seconds=3600)

auth_backend = AuthenticationBackend(
    name="cookie",
    transport=cookie_authentication,
    get_user_manager=lambda user_db: None # UserManager is typically defined outside this module
)

# Example FastAPI app (UserManager and other FastAPIUsers components needed for full functionality)
app = FastAPI()

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

# This is a minimal quickstart for the db adapter itself.
# For a full working FastAPIUsers example, you'd integrate this with FastAPIUsers, UserManager, and routers.
# Example of how you'd use get_user_db in a FastAPI route (simplified):
# @app.get("/users/me")
# async def get_current_user(
#     user_db: SQLAlchemyUserDatabase = Depends(get_user_db),
# ):
#     # In a real app, you'd get the current user from auth backend
#     # and then use user_db to fetch more user data if needed.
#     # This simply demonstrates the user_db dependency is available.
#     return {"message": "User database dependency available!"}

# To run this: 
# 1. Save as main.py
# 2. pip install fastapi uvicorn 'fastapi-users[sqlalchemy]' aiosqlite
# 3. uvicorn main:app --reload