Authentication Library

FastAPI Users

Overview

FastAPI Users is a comprehensive authentication and user management library designed exclusively for FastAPI. With the rapid growth of FastAPI in 2025, it has significantly expanded as the standard for asynchronous application development. It provides integrated support for JWT, Cookie, and OAuth authentication with complete async compatibility and high performance through a pluggable design, establishing itself as the authentication standard in modern API development. Integration with major ORMs like SQLAlchemy and Beanie enables rapid construction of enterprise-level user management systems.

Details

FastAPI Users is a complete user management ecosystem for FastAPI applications. It achieves high performance through complete async support and enables flexible customization through a pluggable architecture.

Technical Features

• Complete Async Support: High-performance processing leveraging FastAPI's Async/Await mechanisms
• Multiple Authentication Backends: Combined support for JWT, Cookie, Database, and Redis strategies
• Pluggable Design: Separation of Transport (token transmission method) and Strategy (token generation method)
• OAuth2 Integration: Easy integration with major providers like Google and GitHub
• Database Agnostic: Support for multiple ORMs including SQLAlchemy and Beanie (MongoDB)
• OpenAPI Schema: Automatic documentation generation and Swagger UI integration

Position in 2025

With the accelerated adoption of FastAPI, FastAPI Users has become the de facto standard for async web development. It features a design optimized for modern microservice architectures and applications requiring real-time communication.

Pros and Cons

Pros

• Development Efficiency: Accelerated development with ready-to-use routes and components
• High Performance: Excellent throughput through complete async processing
• Flexibility: High customizability through modular design
• Security: Built-in industry-standard security practices
• Scalability: Extensibility through multiple authentication strategies
• TypeScript Integration: Type safety leveraging FastAPI's type hints
• Ecosystem: Rich sample code and active community

Cons

• Learning Curve: Requires mastering FastAPI-specific design patterns
• Dependencies: Strong dependence on FastAPI ecosystem
• Configuration Complexity: Complex configuration when utilizing advanced features
• JWT Limitations: Cannot invalidate JWT tokens server-side
• Feature Overload: May be too complex for small-scale projects

Reference Pages

• FastAPI Users Official Site
• GitHub - fastapi-users/fastapi-users
• PyPI - fastapi-users
• FastAPI Users Documentation

Code Examples

Basic Installation and Setup

# Basic installation (with SQLAlchemy backend)
pip install fastapi-users[sqlalchemy]

# Install with OAuth functionality
pip install fastapi-users[sqlalchemy,oauth]

# Install with Beanie (MongoDB) backend
pip install fastapi-users[beanie]

# Additional dependencies
pip install uvicorn[standard] aiosqlite

Basic User Model and Database Setup

# db.py - User model definition with SQLAlchemy
import uuid
from typing import AsyncGenerator

from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(SQLAlchemyBaseUserTableUUID, Base):
    """User table definition"""
    # Basic fields inherited from SQLAlchemyBaseUserTableUUID
    # id: UUID (Primary Key)
    # email: str (Unique)
    # hashed_password: str
    # is_active: bool
    # is_superuser: bool
    # is_verified: bool
    
    # Add custom fields
    first_name: Mapped[str] = mapped_column(String(50), nullable=True)
    last_name: Mapped[str] = mapped_column(String(50), nullable=True)
    phone_number: Mapped[str] = mapped_column(String(20), nullable=True)


# Database configuration
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


async def get_user_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyUserDatabase(session, User)

Pydantic Schema Definition

# schemas.py - Request/Response schema definitions
import uuid
from typing import Optional

from fastapi_users import schemas


class UserRead(schemas.BaseUser[uuid.UUID]):
    """User information read schema"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None


class UserCreate(schemas.BaseUserCreate):
    """User creation schema"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None


class UserUpdate(schemas.BaseUserUpdate):
    """User update schema"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None

User Manager and Authentication Backend Setup

# users.py - User manager and authentication configuration
import uuid
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, UUIDIDMixin
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    CookieTransport,
    JWTStrategy,
)

from .db import User, get_user_db


SECRET = "your-secret-key-here"  # Use environment variables in production


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")

    async def validate_password(
        self,
        password: str,
        user: User | UserCreate,
    ) -> None:
        if len(password) < 8:
            raise ValueError("Password should be at least 8 characters")
        if not any(char.isdigit() for char in password):
            raise ValueError("Password should contain at least one digit")


async def get_user_manager(user_db=Depends(get_user_db)):
    yield UserManager(user_db)


# Authentication backend configuration

# JWT Bearer authentication
bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")

def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)

jwt_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)

# Cookie authentication
cookie_transport = CookieTransport(cookie_max_age=3600)

cookie_backend = AuthenticationBackend(
    name="cookie",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

# Use multiple backends
auth_backends = [jwt_backend, cookie_backend]

FastAPI Application Setup

# app.py - Main application
import uuid
from fastapi import FastAPI, Depends
from fastapi_users import FastAPIUsers

from .db import User, create_db_and_tables
from .schemas import UserCreate, UserRead, UserUpdate
from .users import auth_backends, get_user_manager

# FastAPIUsers instance
fastapi_users = FastAPIUsers[User, uuid.UUID](
    get_user_manager,
    auth_backends,
)

# Current user dependency functions
current_active_user = fastapi_users.current_user(active=True)
current_superuser = fastapi_users.current_user(active=True, superuser=True)

app = FastAPI(title="FastAPI Users Example")

# Authentication routers
app.include_router(
    fastapi_users.get_auth_router(jwt_backend),
    prefix="/auth/jwt",
    tags=["auth"],
)

app.include_router(
    fastapi_users.get_auth_router(cookie_backend),
    prefix="/auth/cookie",
    tags=["auth"],
)

# User registration router
app.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix="/auth",
    tags=["auth"],
)

# Password reset router
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)

# Email verification router
app.include_router(
    fastapi_users.get_verify_router(UserRead),
    prefix="/auth",
    tags=["auth"],
)

# User management router
app.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="/users",
    tags=["users"],
)


@app.on_event("startup")
async def on_startup():
    # Create database and tables
    await create_db_and_tables()


# Protected route examples
@app.get("/protected")
async def protected_route(user: User = Depends(current_active_user)):
    return f"Hello, {user.email}!"


@app.get("/admin")
async def admin_route(user: User = Depends(current_superuser)):
    return f"Logged in as admin: {user.email}"

OAuth Provider Integration

# oauth.py - OAuth configuration
from httpx_oauth.clients.google import GoogleOAuth2

google_oauth_client = GoogleOAuth2(
    client_id="YOUR_GOOGLE_CLIENT_ID",
    client_secret="YOUR_GOOGLE_CLIENT_SECRET",
)

# Add OAuth router (in app.py)
app.include_router(
    fastapi_users.get_oauth_router(
        google_oauth_client,
        jwt_backend,
        SECRET,
        is_verified_by_default=True,  # OAuth users are automatically verified
        associate_by_email=True,      # Associate accounts with same email
    ),
    prefix="/auth/google",
    tags=["auth"],
)

# Router for existing users to associate OAuth accounts
app.include_router(
    fastapi_users.get_oauth_associate_router(google_oauth_client, UserRead, SECRET),
    prefix="/auth/associate/google",
    tags=["auth"],
)

Advanced User Management and Customization

# advanced_users.py - Advanced user management
import uuid
from typing import Optional
from fastapi import Request, HTTPException, status
from fastapi_users import BaseUserManager
from fastapi_users.password import PasswordHelper

from .db import User
from .schemas import UserCreate


class AdvancedUserManager(BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    def __init__(self, user_db):
        super().__init__(user_db)
        # Custom password helper
        self.password_helper = PasswordHelper()

    async def create(
        self,
        user_create: UserCreate,
        safe: bool = False,
        request: Optional[Request] = None,
    ) -> User:
        """Custom user creation logic"""
        # Email domain restriction
        allowed_domains = ["company.com", "trusted-partner.com"]
        email_domain = user_create.email.split("@")[1]
        
        if email_domain not in allowed_domains:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email domain not allowed"
            )

        # Pre-creation processing
        user = await super().create(user_create, safe, request)
        
        # Post-creation processing (e.g., assign default roles)
        # await self.assign_default_role(user)
        
        return user

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        """Post-registration processing"""
        print(f"New user registered: {user.email}")
        # Send emails, logging, external system integration, etc.
        # await self.send_welcome_email(user)
        # await self.log_user_registration(user)

    async def on_after_login(
        self,
        user: User,
        request: Optional[Request] = None,
        response: Optional[Response] = None,
    ):
        """Post-login processing"""
        print(f"User logged in: {user.email}")
        # Login history recording, session management, etc.
        # await self.update_last_login(user)

    async def on_after_update(
        self,
        user: User,
        update_dict: dict,
        request: Optional[Request] = None,
    ):
        """Post-user update processing"""
        print(f"User information updated: {user.email}")
        # Change notifications, audit logs, etc.

    async def validate_password(
        self,
        password: str,
        user: User | UserCreate,
    ) -> None:
        """Custom password validation"""
        if len(password) < 12:
            raise ValueError("Password should be at least 12 characters")
        
        if not any(char.isupper() for char in password):
            raise ValueError("Password should contain uppercase letters")
        
        if not any(char.islower() for char in password):
            raise ValueError("Password should contain lowercase letters")
        
        if not any(char.isdigit() for char in password):
            raise ValueError("Password should contain digits")
        
        if not any(char in "!@#$%^&*()_+-=[]{}|;:,.<>?" for char in password):
            raise ValueError("Password should contain special characters")

    async def authenticate(
        self, credentials: OAuth2PasswordRequestForm
    ) -> Optional[User]:
        """Custom authentication logic"""
        try:
            user = await self.get_by_email(credentials.username)
        except UserNotExists:
            # Dummy verification for timing attack protection
            self.password_helper.hash("dummy")
            return None

        # Account lockout feature example
        if hasattr(user, 'failed_login_attempts') and user.failed_login_attempts >= 5:
            raise HTTPException(
                status_code=status.HTTP_423_LOCKED,
                detail="Account is locked"
            )

        verified, updated_password_hash = self.password_helper.verify_and_update(
            credentials.password, user.hashed_password
        )
        
        if not verified:
            # Increment failed login attempts
            # await self.increment_failed_login_attempts(user)
            return None
        
        # Reset counter on successful login
        # await self.reset_failed_login_attempts(user)
        
        if updated_password_hash is not None:
            await self.user_db.update(user, {"hashed_password": updated_password_hash})

        return user

Dynamic Authentication Backends and Security Configuration

# security.py - Security and access control
from fastapi import Request, Depends, HTTPException, status
from fastapi_users.authentication import AuthenticationBackend

async def get_enabled_backends(request: Request):
    """Dynamically determine enabled authentication backends based on request"""
    
    # Admin endpoints use JWT only
    if request.url.path.startswith("/admin"):
        return [jwt_backend]
    
    # API endpoints use JWT only
    if request.url.path.startswith("/api"):
        return [jwt_backend]
    
    # Others allow both Cookie and JWT
    return [cookie_backend, jwt_backend]


# Dynamic authentication usage example
current_user_dynamic = fastapi_users.current_user(
    active=True, 
    get_enabled_backends=get_enabled_backends
)

@app.get("/admin/dashboard")
async def admin_dashboard(user: User = Depends(current_user_dynamic)):
    if not user.is_superuser:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin privileges required"
        )
    return {"message": f"Admin dashboard - {user.email}"}


# Role-based access control implementation
def require_roles(*required_roles: str):
    """Decorator that requires specified roles"""
    def role_checker(user: User = Depends(current_active_user)):
        # Get user roles (custom implementation required)
        user_roles = getattr(user, 'roles', [])
        
        if not any(role in user_roles for role in required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Required roles: {', '.join(required_roles)}"
            )
        return user
    
    return role_checker


@app.get("/editor/content")
async def editor_content(user: User = Depends(require_roles("editor", "admin"))):
    return {"message": "Editor-only content"}