FastAPI : Créer des APIs REST modernes en Python

Estimated reading time: 17 minutes

Introduction

FastAPI est un framework web moderne, rapide et basé sur les standards pour construire des APIs avec Python 3.10+. Il combine performance, validation automatique des données, documentation interactive et support natif de la programmation asynchrone.

Dans cet article, vous apprendrez à créer une API REST complète avec authentification, base de données, tests, et déploiement.

Configuration de l’environnement

Structure du projet

fastapi-project/
├── app/
│   ├── init.py
│   ├── main.py
│   ├── config.py
│   ├── database.py
│   ├── models/
│   │   ├── init.py
│   │   └── user.py
│   ├── schemas/
│   │   ├── init.py
│   │   └── user.py
│   ├── api/
│   │   ├── init.py
│   │   ├── deps.py
│   │   └── v1/
│   │       ├── init.py
│   │       ├── users.py
│   │       └── auth.py
│   └── core/
│       ├── init.py
│       ├── security.py
│       └── middleware.py
├── tests/
│   ├── init.py
│   └── testapi.py
├── requirements.txt
├── .env.example
└── README.md

requirements.txt

# FastAPI et serveur ASGI
fastapi==0.109.0
uvicorn[standard]==0.27.0
python-multipart==0.0.6

# Base de données
sqlalchemy==2.0.25
alembic==1.13.1
asyncpg==0.29.0
psycopg2-binary==2.9.9

# Authentification et sécurité
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-dotenv==1.0.0

# Validation
pydantic==2.5.3
pydantic-settings==2.1.0
email-validator==2.1.0

# Tests
pytest==7.4.3
pytest-asyncio==0.23.3
pytest-cov==4.1.0
httpx==0.26.0

# Outils de développement
black==23.12.1
mypy==1.8.0
ruff==0.1.11

Installation:

# Créer l'environnement virtuel
python3 -m venv venv
source venv/bin/activate  # Linux/Mac
# venvScriptsactivate  # Windows

# Installer les dépendances
pip install -r requirements.txt

Configuration de l’application

app/config.py

"""
app/config.py - Configuration de l'application avec Pydantic Settings
"""

from typing import Optional
from pydantic import PostgresDsn, fieldvalidator
from pydanticsettings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    """Configuration de l'application."""

    # Configuration de l'API
    APIV1PREFIX: str = "/api/v1"
    PROJECTNAME: str = "FastAPI Application"
    VERSION: str = "1.0.0"
    DEBUG: bool = False

    # CORS
    BACKENDCORSORIGINS: list[str] = ["http://localhost:3000"]

    # Base de données
    POSTGRESSERVER: str = "localhost"
    POSTGRESUSER: str = "postgres"
    POSTGRESPASSWORD: str = "postgres"
    POSTGRESDB: str = "fastapidb"
    POSTGRESPORT: str = "5432"

    DATABASEURL: Optional[PostgresDsn] = None

    @fieldvalidator("DATABASEURL", mode="before")
    @classmethod
    def assembledbconnection(cls, v: Optional[str], info) -> str:
        """Construit l'URL de la base de données."""
        if isinstance(v, str):
            return v

        values = info.data
        return str(
            PostgresDsn.build(
                scheme="postgresql+asyncpg",
                username=values.get("POSTGRESUSER"),
                password=values.get("POSTGRESPASSWORD"),
                host=values.get("POSTGRESSERVER"),
                port=int(values.get("POSTGRESPORT", 5432)),
                path=f"{values.get('POSTGRESDB') or ''}",
            )
        )

    # JWT
    SECRETKEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESSTOKENEXPIREMINUTES: int = 30
    REFRESHTOKENEXPIREDAYS: int = 7

    # Configuration Pydantic
    modelconfig = SettingsConfigDict(
        envfile=".env",
        envfileencoding="utf-8",
        casesensitive=True
    )

# Instance globale des settings
settings = Settings()

.env.example

# .env.example - Fichier d'exemple de configuration

# Database
POSTGRESSERVER=localhost
POSTGRESUSER=postgres
POSTGRESPASSWORD=yourpassword
POSTGRESDB=fastapidb
POSTGRESPORT=5432

# Security
SECRETKEY=your-super-secret-key-change-in-production
ALGORITHM=HS256
ACCESSTOKENEXPIREMINUTES=30

# API
DEBUG=False
PROJECTNAME=FastAPI Application

Modèles de base de données

app/database.py

"""
app/database.py - Configuration de la base de données
"""

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    createasyncengine,
    asyncsessionmaker
)
from sqlalchemy.orm import DeclarativeBase
from app.config import settings

# Créer le moteur async
engine = createasyncengine(
    str(settings.DATABASEURL),
    echo=settings.DEBUG,
    future=True,
    poolpreping=True,
    poolsize=10,
    maxoverflow=20
)

# Session factory
AsyncSessionLocal = asyncsessionmaker(
    engine,
    class=AsyncSession,
    expireoncommit=False,
    autoflush=False,
    autocommit=False
)

class Base(DeclarativeBase):
    """Classe de base pour tous les modèles."""
    pass

async def getdb() -> AsyncGenerator[AsyncSession, None]:
    """
    Dépendance pour obtenir une session de base de données.

    Yields:
        Session de base de données
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

async def initdb() -> None:
    """Initialise la base de données."""
    async with engine.begin() as conn:
        await conn.runsync(Base.metadata.createall)

async def closedb() -> None:
    """Ferme les connexions à la base de données."""
    await engine.dispose()

app/models/user.py

"""
app/models/user.py - Modèle SQLAlchemy pour les utilisateurs
"""

from datetime import datetime
from typing import Optional
from sqlalchemy import String, Boolean, DateTime, Text
from sqlalchemy.orm import Mapped, mappedcolumn
from app.database import Base

class User(Base):
    """Modèle utilisateur."""

    tablename = "users"

    id: Mapped[int] = mappedcolumn(primarykey=True, index=True)
    email: Mapped[str] = mappedcolumn(
        String(255),
        unique=True,
        index=True,
        nullable=False
    )
    username: Mapped[str] = mappedcolumn(
        String(50),
        unique=True,
        index=True,
        nullable=False
    )
    hashedpassword: Mapped[str] = mappedcolumn(String(255), nullable=False)
    fullname: Mapped[Optional[str]] = mappedcolumn(String(100))
    isactive: Mapped[bool] = mappedcolumn(Boolean, default=True)
    issuperuser: Mapped[bool] = mappedcolumn(Boolean, default=False)
    bio: Mapped[Optional[str]] = mappedcolumn(Text)
    avatarurl: Mapped[Optional[str]] = mappedcolumn(String(500))
    createdat: Mapped[datetime] = mappedcolumn(
        DateTime,
        default=datetime.utcnow,
        nullable=False
    )
    updatedat: Mapped[datetime] = mappedcolumn(
        DateTime,
        default=datetime.utcnow,
        onupdate=datetime.utcnow,
        nullable=False
    )

    def repr(self) -> str:
        """Représentation string."""
        return f""

Schémas Pydantic

app/schemas/user.py

"""
app/schemas/user.py - Schémas Pydantic pour validation
"""

from datetime import datetime
from typing import Optional
from pydantic import BaseModel, EmailStr, Field, ConfigDict

# Schémas de base
class UserBase(BaseModel):
    """Propriétés communes aux utilisateurs."""
    email: EmailStr
    username: str = Field(..., minlength=3, maxlength=50)
    fullname: Optional[str] = Field(None, maxlength=100)
    bio: Optional[str] = None
    avatarurl: Optional[str] = None

class UserCreate(UserBase):
    """Propriétés pour créer un utilisateur."""
    password: str = Field(..., minlength=8, maxlength=100)

class UserUpdate(BaseModel):
    """Propriétés pour mettre à jour un utilisateur."""
    email: Optional[EmailStr] = None
    username: Optional[str] = Field(None, minlength=3, maxlength=50)
    fullname: Optional[str] = Field(None, maxlength=100)
    bio: Optional[str] = None
    avatarurl: Optional[str] = None
    password: Optional[str] = Field(None, minlength=8, maxlength=100)

class UserInDB(UserBase):
    """Propriétés stockées en base de données."""
    id: int
    isactive: bool
    issuperuser: bool
    createdat: datetime
    updatedat: datetime

    modelconfig = ConfigDict(fromattributes=True)

class User(UserInDB):
    """Propriétés retournées au client."""
    pass

class UserPublic(BaseModel):
    """Profil public d'un utilisateur."""
    id: int
    username: str
    fullname: Optional[str]
    bio: Optional[str]
    avatarurl: Optional[str]
    createdat: datetime

    modelconfig = ConfigDict(fromattributes=True)

# Schémas d'authentification
class Token(BaseModel):
    """Token JWT."""
    accesstoken: str
    refreshtoken: str
    tokentype: str = "bearer"

class TokenPayload(BaseModel):
    """Payload du token JWT."""
    sub: Optional[int] = None
    exp: Optional[datetime] = None

class LoginRequest(BaseModel):
    """Requête de connexion."""
    username: str
    password: str

# Schémas de réponse
class UserResponse(BaseModel):
    """Réponse contenant un utilisateur."""
    user: User

class UsersResponse(BaseModel):
    """Réponse contenant une liste d'utilisateurs."""
    users: list[UserPublic]
    total: int
    page: int
    pagesize: int

class MessageResponse(BaseModel):
    """Réponse avec un message."""
    message: str

Sécurité et authentification

app/core/security.py

"""
app/core/security.py - Fonctions de sécurité et JWT
"""

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.config import settings

# Configuration du hachage de mot de passe
pwdcontext = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verifypassword(plainpassword: str, hashedpassword: str) -> bool:
    """
    Vérifie un mot de passe.

    Args:
        plainpassword: Mot de passe en clair
        hashedpassword: Mot de passe haché

    Returns:
        True si le mot de passe correspond
    """
    return pwdcontext.verify(plainpassword, hashedpassword)

def getpasswordhash(password: str) -> str:
    """
    Hache un mot de passe.

    Args:
        password: Mot de passe en clair

    Returns:
        Mot de passe haché
    """
    return pwdcontext.hash(password)

def createaccesstoken(
    subject: int,
    expiresdelta: Optional[timedelta] = None
) -> str:
    """
    Crée un token d'accès JWT.

    Args:
        subject: ID de l'utilisateur
        expiresdelta: Durée de validité

    Returns:
        Token JWT
    """
    if expiresdelta:
        expire = datetime.utcnow() + expiresdelta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=settings.ACCESSTOKENEXPIREMINUTES
        )

    toencode = {"exp": expire, "sub": str(subject)}
    encodedjwt = jwt.encode(
        toencode,
        settings.SECRETKEY,
        algorithm=settings.ALGORITHM
    )
    return encodedjwt

def createrefreshtoken(subject: int) -> str:
    """
    Crée un token de rafraîchissement JWT.

    Args:
        subject: ID de l'utilisateur

    Returns:
        Token JWT
    """
    expire = datetime.utcnow() + timedelta(days=settings.REFRESHTOKENEXPIREDAYS)
    toencode = {"exp": expire, "sub": str(subject)}
    encodedjwt = jwt.encode(
        toencode,
        settings.SECRETKEY,
        algorithm=settings.ALGORITHM
    )
    return encodedjwt

def decodetoken(token: str) -> Optional[int]:
    """
    Décode un token JWT.

    Args:
        token: Token JWT

    Returns:
        ID de l'utilisateur ou None
    """
    try:
        payload = jwt.decode(
            token,
            settings.SECRETKEY,
            algorithms=[settings.ALGORITHM]
        )
        userid: Optional[str] = payload.get("sub")
        if userid is None:
            return None
        return int(userid)
    except JWTError:
        return None

app/api/deps.py

"""
app/api/deps.py - Dépendances communes pour les endpoints
"""

from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import getdb
from app.models.user import User
from app.core.security import decodetoken

# Configuration OAuth2
oauth2scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def getcurrentuser(
    token: str = Depends(oauth2scheme),
    db: AsyncSession = Depends(getdb)
) -> User:
    """
    Récupère l'utilisateur actuel à partir du token JWT.

    Args:
        token: Token JWT
        db: Session de base de données

    Returns:
        Utilisateur actuel

    Raises:
        HTTPException: Si le token est invalide
    """
    credentialsexception = HTTPException(
        statuscode=status.HTTP401UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    userid = decodetoken(token)
    if userid is None:
        raise credentialsexception

    result = await db.execute(select(User).where(User.id == userid))
    user = result.scalaroneornone()

    if user is None:
        raise credentialsexception

    if not user.isactive:
        raise HTTPException(
            statuscode=status.HTTP403FORBIDDEN,
            detail="Inactive user"
        )

    return user

async def getcurrentactivesuperuser(
    currentuser: User = Depends(getcurrentuser),
) -> User:
    """
    Vérifie que l'utilisateur est un superutilisateur.

    Args:
        currentuser: Utilisateur actuel

    Returns:
        Utilisateur actuel

    Raises:
        HTTPException: Si l'utilisateur n'est pas superuser
    """
    if not currentuser.issuperuser:
        raise HTTPException(
            statuscode=status.HTTP403FORBIDDEN,
            detail="Not enough permissions"
        )
    return currentuser

class Pagination:
    """Paramètres de pagination."""

    def init(
        self,
        skip: int = 0,
        limit: int = 100
    ) -> None:
        """
        Initialise les paramètres de pagination.

        Args:
            skip: Nombre d'éléments à sauter
            limit: Nombre max d'éléments à retourner
        """
        self.skip = skip
        self.limit = min(limit, 100)  # Max 100 éléments par page

Routes API

app/api/v1/auth.py

"""
app/api/v1/auth.py - Endpoints d'authentification
"""

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import getdb
from app.models.user import User
from app.schemas.user import Token, UserCreate, UserResponse
from app.core.security import (
    verifypassword,
    getpasswordhash,
    createaccesstoken,
    createrefreshtoken
)

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/register", responsemodel=UserResponse, statuscode=status.HTTP201CREATED)
async def register(
    userin: UserCreate,
    db: AsyncSession = Depends(getdb)
) -> dict:
    """
    Inscription d'un nouvel utilisateur.

    Args:
        userin: Données de l'utilisateur
        db: Session de base de données

    Returns:
        Utilisateur créé

    Raises:
        HTTPException: Si email ou username déjà utilisé
    """
    # Vérifier si l'email existe déjà
    result = await db.execute(
        select(User).where(User.email == userin.email)
    )
    if result.scalaroneornone():
        raise HTTPException(
            statuscode=status.HTTP400BADREQUEST,
            detail="Email already registered"
        )

    # Vérifier si le username existe déjà
    result = await db.execute(
        select(User).where(User.username == userin.username)
    )
    if result.scalaroneornone():
        raise HTTPException(
            statuscode=status.HTTP400BADREQUEST,
            detail="Username already taken"
        )

    # Créer l'utilisateur
    dbuser = User(
        email=userin.email,
        username=userin.username,
        hashedpassword=getpasswordhash(userin.password),
        fullname=userin.fullname,
        bio=userin.bio,
        avatarurl=userin.avatarurl
    )

    db.add(dbuser)
    await db.commit()
    await db.refresh(dbuser)

    return {"user": dbuser}

@router.post("/login", responsemodel=Token)
async def login(
    formdata: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(getdb)
) -> dict:
    """
    Connexion et obtention de tokens JWT.

    Args:
        formdata: Formulaire de connexion
        db: Session de base de données

    Returns:
        Tokens d'accès et de rafraîchissement

    Raises:
        HTTPException: Si identifiants incorrects
    """
    # Rechercher l'utilisateur
    result = await db.execute(
        select(User).where(User.username == formdata.username)
    )
    user = result.scalaroneornone()

    # Vérifier les identifiants
    if not user or not verifypassword(formdata.password, user.hashedpassword):
        raise HTTPException(
            statuscode=status.HTTP401UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if not user.isactive:
        raise HTTPException(
            statuscode=status.HTTP403FORBIDDEN,
            detail="Inactive user"
        )

    # Créer les tokens
    accesstoken = createaccesstoken(subject=user.id)
    refreshtoken = createrefreshtoken(subject=user.id)

    return {
        "accesstoken": accesstoken,
        "refreshtoken": refreshtoken,
        "tokentype": "bearer"
    }

app/api/v1/users.py

"""
app/api/v1/users.py - Endpoints pour gérer les utilisateurs
"""

from typing import List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import getdb
from app.models.user import User
from app.schemas.user import (
    User as UserSchema,
    UserPublic,
    UserUpdate,
    UsersResponse,
    UserResponse,
    MessageResponse
)
from app.api.deps import (
    getcurrentuser,
    getcurrentactivesuperuser,
    Pagination
)
from app.core.security import getpasswordhash

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", responsemodel=UserResponse)
async def readcurrentuser(
    currentuser: User = Depends(getcurrentuser)
) -> dict:
    """
    Récupère le profil de l'utilisateur actuel.

    Args:
        currentuser: Utilisateur actuel

    Returns:
        Profil utilisateur
    """
    return {"user": currentuser}

@router.put("/me", responsemodel=UserResponse)
async def updatecurrentuser(
    userupdate: UserUpdate,
    currentuser: User = Depends(getcurrentuser),
    db: AsyncSession = Depends(getdb)
) -> dict:
    """
    Met à jour le profil de l'utilisateur actuel.

    Args:
        userupdate: Données à mettre à jour
        currentuser: Utilisateur actuel
        db: Session de base de données

    Returns:
        Profil utilisateur mis à jour
    """
    updatedata = userupdate.modeldump(excludeunset=True)

    # Si le mot de passe est fourni, le hacher
    if "password" in updatedata:
        updatedata["hashedpassword"] = getpasswordhash(updatedata.pop("password"))

    # Mettre à jour les champs
    for field, value in updatedata.items():
        setattr(currentuser, field, value)

    await db.commit()
    await db.refresh(currentuser)

    return {"user": currentuser}

@router.delete("/me", responsemodel=MessageResponse)
async def deletecurrentuser(
    currentuser: User = Depends(getcurrentuser),
    db: AsyncSession = Depends(getdb)
) -> dict:
    """
    Supprime le compte de l'utilisateur actuel.

    Args:
        currentuser: Utilisateur actuel
        db: Session de base de données

    Returns:
        Message de confirmation
    """
    await db.delete(currentuser)
    await db.commit()

    return {"message": "User deleted successfully"}

@router.get("", responsemodel=UsersResponse)
async def listusers(
    pagination: Pagination = Depends(),
    db: AsyncSession = Depends(getdb),
    : User = Depends(getcurrentactivesuperuser)
) -> dict:
    """
    Liste tous les utilisateurs (superuser uniquement).

    Args:
        pagination: Paramètres de pagination
        db: Session de base de données

    Returns:
        Liste d'utilisateurs avec pagination
    """
    # Compter le total
    countresult = await db.execute(select(func.count(User.id)))
    total = countresult.scalarone()

    # Récupérer les utilisateurs
    result = await db.execute(
        select(User)
        .offset(pagination.skip)
        .limit(pagination.limit)
    )
    users = result.scalars().all()

    return {
        "users": users,
        "total": total,
        "page": pagination.skip // pagination.limit + 1,
        "pagesize": pagination.limit
    }

@router.get("/{userid}", responsemodel=UserResponse)
async def getuser(
    userid: int,
    db: AsyncSession = Depends(getdb),
    : User = Depends(getcurrentuser)
) -> dict:
    """
    Récupère un utilisateur par son ID.

    Args:
        userid: ID de l'utilisateur
        db: Session de base de données

    Returns:
        Utilisateur

    Raises:
        HTTPException: Si utilisateur non trouvé
    """
    result = await db.execute(select(User).where(User.id == userid))
    user = result.scalaroneornone()

    if not user:
        raise HTTPException(
            statuscode=status.HTTP404NOTFOUND,
            detail="User not found"
        )

    return {"user": user}

@router.delete("/{userid}", responsemodel=MessageResponse)
async def deleteuser(
    userid: int,
    db: AsyncSession = Depends(getdb),
    : User = Depends(getcurrentactivesuperuser)
) -> dict:
    """
    Supprime un utilisateur (superuser uniquement).

    Args:
        userid: ID de l'utilisateur
        db: Session de base de données

    Returns:
        Message de confirmation

    Raises:
        HTTPException: Si utilisateur non trouvé
    """
    result = await db.execute(select(User).where(User.id == userid))
    user = result.scalaroneornone()

    if not user:
        raise HTTPException(
            statuscode=status.HTTP404NOTFOUND,
            detail="User not found"
        )

    await db.delete(user)
    await db.commit()

    return {"message": "User deleted successfully"}

Application principale

app/main.py

"""
app/main.py - Point d'entrée de l'application FastAPI
"""

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from app.config import settings
from app.database import initdb, closedb
from app.api.v1 import auth, users

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Gestionnaire du cycle de vie de l'application."""
    # Startup
    await initdb()
    yield
    # Shutdown
    await closedb()

# Créer l'application FastAPI
app = FastAPI(
    title=settings.PROJECTNAME,
    version=settings.VERSION,
    openapiurl=f"{settings.APIV1PREFIX}/openapi.json",
    docsurl="/docs",
    redocurl="/redoc",
    lifespan=lifespan
)

# Middleware CORS
app.addmiddleware(
    CORSMiddleware,
    alloworigins=settings.BACKENDCORSORIGINS,
    allowcredentials=True,
    allowmethods=[""],
    allowheaders=[""],
)

# Middleware de compression
app.addmiddleware(GZipMiddleware, minimumsize=1000)

# Middleware de sécurité
if not settings.DEBUG:
    app.addmiddleware(
        TrustedHostMiddleware,
        allowedhosts=["example.com", ".example.com"]
    )

# Inclure les routers
app.includerouter(auth.router, prefix=settings.APIV1PREFIX)
app.includerouter(users.router, prefix=settings.APIV1PREFIX)

@app.get("/")
async def root():
    """Endpoint racine."""
    return {
        "message": f"Welcome to {settings.PROJECTNAME}",
        "version": settings.VERSION,
        "docs": "/docs"
    }

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

Tests

tests/testapi.py

"""
tests/testapi.py - Tests de l'API
"""

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import createasyncengine, asyncsessionmaker, AsyncSession
from app.main import app
from app.database import Base, getdb
from app.core.security import getpasswordhash

# Base de données de test
TESTDATABASEURL = "postgresql+asyncpg://postgres:postgres@localhost:5432/testdb"

engine = createasyncengine(TESTDATABASEURL, echo=False)
TestingSessionLocal = asyncsessionmaker(engine, class=AsyncSession, expireoncommit=False)

@pytest.fixture(scope="function")
async def dbsession():
    """Fixture pour la session de base de données de test."""
    async with engine.begin() as conn:
        await conn.runsync(Base.metadata.createall)

    async with TestingSessionLocal() as session:
        yield session

    async with engine.begin() as conn:
        await conn.runsync(Base.metadata.dropall)

@pytest.fixture(scope="function")
async def client(dbsession):
    """Fixture pour le client HTTP de test."""
    async def overridegetdb():
        yield dbsession

    app.dependencyoverrides[getdb] = overridegetdb

    async with AsyncClient(app=app, baseurl="http://test") as ac:
        yield ac

    app.dependencyoverrides.clear()

@pytest.mark.asyncio
async def testregisteruser(client: AsyncClient):
    """Test d'inscription d'un utilisateur."""
    response = await client.post(
        "/api/v1/auth/register",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "testpassword123",
            "fullname": "Test User"
        }
    )

    assert response.statuscode == 201
    data = response.json()
    assert data["user"]["email"] == "test@example.com"
    assert data["user"]["username"] == "testuser"
    assert "hashedpassword" not in data["user"]

@pytest.mark.asyncio
async def testlogin(client: AsyncClient):
    """Test de connexion."""
    # Créer un utilisateur
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "testpassword123"
        }
    )

    # Se connecter
    response = await client.post(
        "/api/v1/auth/login",
        data={
            "username": "testuser",
            "password": "testpassword123"
        }
    )

    assert response.statuscode == 200
    data = response.json()
    assert "accesstoken" in data
    assert "refreshtoken" in data
    assert data["tokentype"] == "bearer"

@pytest.mark.asyncio
async def testgetcurrentuser(client: AsyncClient):
    """Test de récupération de l'utilisateur actuel."""
    # Créer et connecter un utilisateur
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "testpassword123"
        }
    )

    loginresponse = await client.post(
        "/api/v1/auth/login",
        data={
            "username": "testuser",
            "password": "testpassword123"
        }
    )
    token = loginresponse.json()["accesstoken"]

    # Récupérer le profil
    response = await client.get(
        "/api/v1/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )

    assert response.statuscode == 200
    data = response.json()
    assert data["user"]["username"] == "testuser"

@pytest.mark.asyncio
async def testupdateuser(client: AsyncClient):
    """Test de mise à jour du profil."""
    # Créer et connecter
    await client.post(
        "/api/v1/auth/register",
        json={
            "email": "test@example.com",
            "username": "testuser",
            "password": "testpassword123"
        }
    )

    loginresponse = await client.post(
        "/api/v1/auth/login",
        data={
            "username": "testuser",
            "password": "testpassword123"
        }
    )
    token = loginresponse.json()["accesstoken"]

    # Mettre à jour
    response = await client.put(
        "/api/v1/users/me",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "fullname": "Updated Name",
            "bio": "This is my bio"
        }
    )

    assert response.statuscode == 200
    data = response.json()
    assert data["user"]["fullname"] == "Updated Name"
    assert data["user"]["bio"] == "This is my bio"

@pytest.mark.asyncio
async def testunauthorizedaccess(client: AsyncClient):
    """Test d'accès non autorisé."""
    response = await client.get("/api/v1/users/me")

    assert response.statuscode == 401

Lancement de l’application

Avec Uvicorn

# Mode développement (avec rechargement automatique)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# Mode production
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Installer les dépendances système
RUN apt-get update && apt-get install -y 
    postgresql-client 
    && rm -rf /var/lib/apt/lists/

# Copier les requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier le code
COPY . .

# Exposer le port
EXPOSE 8000

# Commande de démarrage
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  db:
    image: postgres:15
    environment:
      POSTGRESUSER: postgres
      POSTGRESPASSWORD: postgres
      POSTGRESDB: fastapidb
    ports:
      - "5432:5432"
    volumes:
      - postgresdata:/var/lib/postgresql/data

  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASEURL: postgresql+asyncpg://postgres:postgres@db:5432/fastapidb
      SECRETKEY: your-secret-key
    dependson:
      - db
    volumes:
      - ./app:/app/app

volumes:
  postgres_data:

Lancement:

docker-compose up -d

Conclusion

Vous avez maintenant une API REST complète avec FastAPI incluant:

Configuration robuste avec Pydantic Settings
Base de données asynchrone avec SQLAlchemy 2.0
Authentification JWT sécurisée
Validation automatique avec Pydantic
Documentation interactive (Swagger UI)
Tests asynchrones avec pytest
Déploiement avec Docker

Ressources

FastAPI Documentation: https://fastapi.tiangolo.com/
SQLAlchemy 2.0: https://docs.sqlalchemy.org/
Pydantic: https://docs.pydantic.dev/
pytest-asyncio: https://pytest-asyncio.readthedocs.io/