Skip to content
Learni
View all tutorials
Développement Backend

How to Implement an Advanced FastAPI API in 2026

22 minADVANCED
FastAPIPython AsyncAPI SécuritéSQLAlchemyJWT Auth
Lire en français

Introduction

In 2026, FastAPI remains the leading Python framework for high-performance APIs thanks to its native async support, automatic validation via Pydantic, and interactive OpenAPI generation. This advanced tutorial guides you through building a complete API to manage users: secure CRUD, JWT authentication, async database with SQLAlchemy, custom middlewares for rate limiting and CORS, as well as exhaustive tests with pytest.

Why is this crucial? Modern APIs must scale horizontally, withstand attacks (DDoS, injections), and integrate seamlessly with React/Vue frontends or microservices. Unlike basic tutorials, we dive into production-ready patterns: hierarchical dependencies, Alembic for migrations, and structured logging for monitoring. At the end, you'll have a copy-pasteable project, deployable on Docker, ready for AWS or Vercel. Expect 30 minutes for setup and tests—a huge time saver for senior devs.

Prerequisites

  • Python 3.12+
  • Advanced knowledge of Python async/await and typing
  • Familiarity with SQLAlchemy, Pydantic, and JWT
  • Tools: Poetry (dependency manager), Alembic (migrations), pytest (tests)
  • IDE with LSP support (VS Code + Ruff)

Project Setup

terminal
mkdir fastapi-advanced-api && cd fastapi-advanced-api
poetry init --no-interaction --python "^3.12"
poetry add fastapi uvicorn[standard] sqlalchemy[asyncio] aiosqlite pydantic[email] python-jose[cryptography] passlib[bcrypt] python-multipart httpx
poetry add --group dev pytest pytest-asyncio pytest-httpx alembic ruff black
poetry install
poetry shell
alembic init alembic

This script initializes a Poetry project for perfect dependency isolation. We install FastAPI core, a lightweight async DB (aiosqlite), auth (JWT + bcrypt), and dev tools for tests/migrations. Alembic sets up DB migrations—avoid pip to prevent version conflicts in prod.

Async Database Configuration

app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator

DATABASE_URL = "sqlite+aiosqlite:///./app.db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

This module creates an async SQLAlchemy engine with aiosqlite for non-blocking queries. The get_db() dependency injects a session per request, with auto-rollback on errors—essential pattern to avoid connection leaks under high load. echo=True aids debugging; disable in prod.

SQLAlchemy Models

app/models.py
from sqlalchemy import String, Integer, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String, nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, email={self.email!r})"

User model with modern typing (Mapped). Indexes on email/id boost queries. No Pydantic schemas here—they're separated to decouple DB and API, avoiding leaks of sensitive data like hashed_password in responses.

Pydantic Schemas for Validation

app/schemas.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime


class UserBase(BaseModel):
    email: EmailStr


class UserCreate(UserBase):
    password: str = Field(min_length=8)


class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    email: Optional[str] = None

Hierarchical schemas for inputs/outputs. EmailStr validates emails, Field enforces rules. from_attributes=True maps ORM to Pydantic without boilerplate. Always separate Create/Read schemas to hide hashed_password and prevent OWASP Top 10 breaches.

CRUD Logic and JWT Auth

app/crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from passlib.context import CryptContext
from jose import JWTError, jwt

from app.models import User
from app.schemas import UserCreate, User

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key-change-in-prod"
ALGORITHM = "HS256"


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()


async def create_user(db: AsyncSession, user: UserCreate) -> User:
    hashed_password = pwd_context.hash(user.password)
    db_user = User(email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    await db.flush()
    return db_user


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def create_access_token(data: dict):
    to_encode = data.copy()
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def authenticate_user(db: AsyncSession, email: str, password: str):
    user = await get_user_by_email(db, email)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

Minimal but robust async CRUD: scalar_one_or_none() handles uniqueness without unnecessary exceptions. Bcrypt for secure hashing, JWT HS256 for tokens (change SECRET_KEY!). authenticate_user centralizes verification—reusable in OAuth2 guards.

Users Router with Dependencies

app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

from app.database import get_db
from app.crud import authenticate_user, create_user, create_access_token
from app.schemas import User, UserCreate, Token

router = APIRouter(prefix="/users", tags=["users"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="users/token")


@router.post("/", response_model=User)
async def create_new_user(user: UserCreate, db=Depends(get_db)):
    db_user = await get_user_by_email(db, user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return await create_user(db, user)


@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
    user = await authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect credentials")
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

Modular router for scalability (one per entity). OAuth2PasswordBearer for Bearer tokens, Depends(get_db) injects the session. Implicit guards via HTTP exceptions—auto-generates Swagger UI. Add @router.get("/me") with current_user: User = Depends(get_current_user) for protected routes.

Middlewares and Main App

app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security.utils import get_authorization_scheme_param

from app.database import engine, Base, get_db
from app.routers import users


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in prod
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    # Simple in-memory limit: 10 req/min (use Redis in prod)
    if not hasattr(app.state, 'requests'):
        app.state.requests = {}
    now = request.url.path
    app.state.requests[client_ip] = app.state.requests.get(client_ip, []) + [now]
    recent = [r for r in app.state.requests[client_ip] if now - r < 60]
    if len(recent) > 10:
        raise HTTPException(429, "Rate limit exceeded")
    app.state.requests[client_ip] = recent
    response = await call_next(request)
    return response

app.include_router(users.router)

@app.get("/health")
async def health():
    return {"status": "healthy"}

App lifecycle with lifespan for auto DB init. CORS and custom rate-limit middleware (in-memory; migrate to Redis). include_router for modularity. Run with uvicorn app.main:app --reload—docs auto at /docs.

Unit Tests with pytest

tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app


@pytest.fixture(scope="module")
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={"email": "test@example.com", "password": "testpass123"})
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "test@example.com"


@pytest.mark.asyncio
async def test_login(client: AsyncClient):
    response = await client.post("/users/token", data={"username": "test@example.com", "password": "testpass123"})
    assert response.status_code == 200
    assert "access_token" in response.json()

Async tests with pytest-asyncio and httpx.AsyncClient to simulate real requests. Reusable client fixture. Covers happy paths—add DB mocks for CI/CD. Achieve >90% coverage with pytest --cov.

Best Practices

  • Alembic Migrations: Configure alembic.ini and env.py for DB versioning. Run alembic revision --autogenerate -m "add users".
  • Environment Vars: Use python-dotenv for SECRET_KEY, DB_URL.
  • Structured Logging: Integrate structlog for JSON traces in prod.
  • Custom Validation: Pydantic validators for business rules (e.g., password strength).
  • Caching: Redis + fastapi-cache for read-heavy endpoints.

Common Mistakes to Avoid

  • Forgetting await on DB ops: causes RuntimeWarning and data loss.
  • Exposing hashed_password in response schemas: always filter via response_model.
  • In-memory rate limiting: non-scalable; switch to Redis or slowapi.
  • No token expiration: Add exp JWT claim (15min) + refresh tokens.

Next Steps

Deepen your knowledge with our Learni advanced FastAPI courses. Resources: FastAPI Docs, SQLAlchemy Async, Typer for CLI. Deploy on Render/K8s with the provided Dockerfile.