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
│ └── test_api.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, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Configuration de l'application."""
# Configuration de l'API
API_V1_PREFIX: str = "/api/v1"
PROJECT_NAME: str = "FastAPI Application"
VERSION: str = "1.0.0"
DEBUG: bool = False
# CORS
BACKEND_CORS_ORIGINS: list[str] = ["http://localhost:3000"]
# Base de données
POSTGRES_SERVER: str = "localhost"
POSTGRES_USER: str = "postgres"
POSTGRES_PASSWORD: str = "postgres"
POSTGRES_DB: str = "fastapi_db"
POSTGRES_PORT: str = "5432"
DATABASE_URL: Optional[PostgresDsn] = None
@field_validator("DATABASE_URL", mode="before")
@classmethod
def assemble_db_connection(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("POSTGRES_USER"),
password=values.get("POSTGRES_PASSWORD"),
host=values.get("POSTGRES_SERVER"),
port=int(values.get("POSTGRES_PORT", 5432)),
path=f"{values.get('POSTGRES_DB') or ''}",
)
)
# JWT
SECRET_KEY: str = "your-secret-key-change-in-production"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# Configuration Pydantic
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=True
)
# Instance globale des settings
settings = Settings()
.env.example
# .env.example - Fichier d'exemple de configuration
# Database
POSTGRES_SERVER=localhost
POSTGRES_USER=postgres
POSTGRES_PASSWORD=your_password
POSTGRES_DB=fastapi_db
POSTGRES_PORT=5432
# Security
SECRET_KEY=your-super-secret-key-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
# API
DEBUG=False
PROJECT_NAME=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,
create_async_engine,
async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase
from app.config import settings
# Créer le moteur async
engine = create_async_engine(
str(settings.DATABASE_URL),
echo=settings.DEBUG,
future=True,
pool_pre_ping=True,
pool_size=10,
max_overflow=20
)
# Session factory
AsyncSessionLocal = async_sessionmaker(
engine,
class*=AsyncSession,
expire_on_commit=False,
autoflush=False,
autocommit=False
)
class Base(DeclarativeBase):
"""Classe de base pour tous les modèles."""
pass
async def get_db() -> 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 init_db() -> None:
"""Initialise la base de données."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def close_db() -> 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, mapped_column
from app.database import Base
class User(Base):
"""Modèle utilisateur."""
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
email: Mapped[str] = mapped_column(
String(255),
unique=True,
index=True,
nullable=False
)
username: Mapped[str] = mapped_column(
String(50),
unique=True,
index=True,
nullable=False
)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
full_name: Mapped[Optional[str]] = mapped_column(String(100))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)
bio: Mapped[Optional[str]] = mapped_column(Text)
avatar_url: Mapped[Optional[str]] = mapped_column(String(500))
created_at: Mapped[datetime] = mapped_column(
DateTime,
default=datetime.utcnow,
nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
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(..., min_length=3, max_length=50)
full_name: Optional[str] = Field(None, max_length=100)
bio: Optional[str] = None
avatar_url: Optional[str] = None
class UserCreate(UserBase):
"""Propriétés pour créer un utilisateur."""
password: str = Field(..., min_length=8, max_length=100)
class UserUpdate(BaseModel):
"""Propriétés pour mettre à jour un utilisateur."""
email: Optional[EmailStr] = None
username: Optional[str] = Field(None, min_length=3, max_length=50)
full_name: Optional[str] = Field(None, max_length=100)
bio: Optional[str] = None
avatar_url: Optional[str] = None
password: Optional[str] = Field(None, min_length=8, max_length=100)
class UserInDB(UserBase):
"""Propriétés stockées en base de données."""
id: int
is_active: bool
is_superuser: bool
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class User(UserInDB):
"""Propriétés retournées au client."""
pass
class UserPublic(BaseModel):
"""Profil public d'un utilisateur."""
id: int
username: str
full_name: Optional[str]
bio: Optional[str]
avatar_url: Optional[str]
created_at: datetime
model_config = ConfigDict(from_attributes=True)
# Schémas d'authentification
class Token(BaseModel):
"""Token JWT."""
access_token: str
refresh_token: str
token_type: 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
page_size: 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
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Vérifie un mot de passe.
Args:
plain_password: Mot de passe en clair
hashed_password: Mot de passe haché
Returns:
True si le mot de passe correspond
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Hache un mot de passe.
Args:
password: Mot de passe en clair
Returns:
Mot de passe haché
"""
return pwd_context.hash(password)
def create_access_token(
subject: int,
expires_delta: Optional[timedelta] = None
) -> str:
"""
Crée un token d'accès JWT.
Args:
subject: ID de l'utilisateur
expires_delta: Durée de validité
Returns:
Token JWT
"""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(
to_encode,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
return encoded_jwt
def create_refresh_token(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.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(
to_encode,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
return encoded_jwt
def decode_token(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.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
user_id: Optional[str] = payload.get("sub")
if user_id is None:
return None
return int(user_id)
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 get_db
from app.models.user import User
from app.core.security import decode_token
# Configuration OAuth2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> 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
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
user_id = decode_token(token)
if user_id is None:
raise credentials_exception
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user"
)
return user
async def get_current_active_superuser(
current_user: User = Depends(get_current_user),
) -> User:
"""
Vérifie que l'utilisateur est un superutilisateur.
Args:
current_user: Utilisateur actuel
Returns:
Utilisateur actuel
Raises:
HTTPException: Si l'utilisateur n'est pas superuser
"""
if not current_user.is_superuser:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
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 get_db
from app.models.user import User
from app.schemas.user import Token, UserCreate, UserResponse
from app.core.security import (
verify_password,
get_password_hash,
create_access_token,
create_refresh_token
)
router = APIRouter(prefix="/auth", tags=["authentication"])
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(
user_in: UserCreate,
db: AsyncSession = Depends(get_db)
) -> dict:
"""
Inscription d'un nouvel utilisateur.
Args:
user_in: 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 == user_in.email)
)
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Vérifier si le username existe déjà
result = await db.execute(
select(User).where(User.username == user_in.username)
)
if result.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already taken"
)
# Créer l'utilisateur
db_user = User(
email=user_in.email,
username=user_in.username,
hashed_password=get_password_hash(user_in.password),
full_name=user_in.full_name,
bio=user_in.bio,
avatar_url=user_in.avatar_url
)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return {"user": db_user}
@router.post("/login", response_model=Token)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
) -> dict:
"""
Connexion et obtention de tokens JWT.
Args:
form_data: 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 == form_data.username)
)
user = result.scalar_one_or_none()
# Vérifier les identifiants
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user"
)
# Créer les tokens
access_token = create_access_token(subject=user.id)
refresh_token = create_refresh_token(subject=user.id)
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "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 get_db
from app.models.user import User
from app.schemas.user import (
User as UserSchema,
UserPublic,
UserUpdate,
UsersResponse,
UserResponse,
MessageResponse
)
from app.api.deps import (
get_current_user,
get_current_active_superuser,
Pagination
)
from app.core.security import get_password_hash
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/me", response_model=UserResponse)
async def read_current_user(
current_user: User = Depends(get_current_user)
) -> dict:
"""
Récupère le profil de l'utilisateur actuel.
Args:
current_user: Utilisateur actuel
Returns:
Profil utilisateur
"""
return {"user": current_user}
@router.put("/me", response_model=UserResponse)
async def update_current_user(
user_update: UserUpdate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
) -> dict:
"""
Met à jour le profil de l'utilisateur actuel.
Args:
user_update: Données à mettre à jour
current_user: Utilisateur actuel
db: Session de base de données
Returns:
Profil utilisateur mis à jour
"""
update_data = user_update.model_dump(exclude_unset=True)
# Si le mot de passe est fourni, le hacher
if "password" in update_data:
update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
# Mettre à jour les champs
for field, value in update_data.items():
setattr(current_user, field, value)
await db.commit()
await db.refresh(current_user)
return {"user": current_user}
@router.delete("/me", response_model=MessageResponse)
async def delete_current_user(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
) -> dict:
"""
Supprime le compte de l'utilisateur actuel.
Args:
current_user: Utilisateur actuel
db: Session de base de données
Returns:
Message de confirmation
"""
await db.delete(current_user)
await db.commit()
return {"message": "User deleted successfully"}
@router.get("", response_model=UsersResponse)
async def list_users(
pagination: Pagination = Depends(),
db: AsyncSession = Depends(get_db),
*: User = Depends(get_current_active_superuser)
) -> 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
count_result = await db.execute(select(func.count(User.id)))
total = count_result.scalar_one()
# 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,
"page_size": pagination.limit
}
@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db),
*: User = Depends(get_current_user)
) -> dict:
"""
Récupère un utilisateur par son ID.
Args:
user_id: 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 == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return {"user": user}
@router.delete("/{user_id}", response_model=MessageResponse)
async def delete_user(
user_id: int,
db: AsyncSession = Depends(get_db),
*: User = Depends(get_current_active_superuser)
) -> dict:
"""
Supprime un utilisateur (superuser uniquement).
Args:
user_id: 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 == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
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 init_db, close_db
from app.api.v1 import auth, users
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Gestionnaire du cycle de vie de l'application."""
# Startup
await init_db()
yield
# Shutdown
await close_db()
# Créer l'application FastAPI
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
openapi_url=f"{settings.API_V1_PREFIX}/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
lifespan=lifespan
)
# Middleware CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.BACKEND_CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Middleware de compression
app.add_middleware(GZipMiddleware, minimum_size=1000)
# Middleware de sécurité
if not settings.DEBUG:
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["example.com", "*.example.com"]
)
# Inclure les routers
app.include_router(auth.router, prefix=settings.API_V1_PREFIX)
app.include_router(users.router, prefix=settings.API_V1_PREFIX)
@app.get("/")
async def root():
"""Endpoint racine."""
return {
"message": f"Welcome to {settings.PROJECT_NAME}",
"version": settings.VERSION,
"docs": "/docs"
}
@app.get("/health")
async def health_check():
"""Endpoint de health check."""
return {"status": "healthy"}
Tests
tests/test_api.py
"""
tests/test_api.py - Tests de l'API
"""
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.main import app
from app.database import Base, get_db
from app.core.security import get_password_hash
# Base de données de test
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/test_db"
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestingSessionLocal = async_sessionmaker(engine, class*=AsyncSession, expire_on_commit=False)
@pytest.fixture(scope="function")
async def db_session():
"""Fixture pour la session de base de données de test."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with TestingSessionLocal() as session:
yield session
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture(scope="function")
async def client(db_session):
"""Fixture pour le client HTTP de test."""
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()
@pytest.mark.asyncio
async def test_register_user(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",
"full_name": "Test User"
}
)
assert response.status_code == 201
data = response.json()
assert data["user"]["email"] == "test@example.com"
assert data["user"]["username"] == "testuser"
assert "hashed_password" not in data["user"]
@pytest.mark.asyncio
async def test_login(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.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
@pytest.mark.asyncio
async def test_get_current_user(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"
}
)
login_response = await client.post(
"/api/v1/auth/login",
data={
"username": "testuser",
"password": "testpassword123"
}
)
token = login_response.json()["access_token"]
# Récupérer le profil
response = await client.get(
"/api/v1/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
data = response.json()
assert data["user"]["username"] == "testuser"
@pytest.mark.asyncio
async def test_update_user(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"
}
)
login_response = await client.post(
"/api/v1/auth/login",
data={
"username": "testuser",
"password": "testpassword123"
}
)
token = login_response.json()["access_token"]
# Mettre à jour
response = await client.put(
"/api/v1/users/me",
headers={"Authorization": f"Bearer {token}"},
json={
"full_name": "Updated Name",
"bio": "This is my bio"
}
)
assert response.status_code == 200
data = response.json()
assert data["user"]["full_name"] == "Updated Name"
assert data["user"]["bio"] == "This is my bio"
@pytest.mark.asyncio
async def test_unauthorized_access(client: AsyncClient):
"""Test d'accès non autorisé."""
response = await client.get("/api/v1/users/me")
assert response.status_code == 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:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: fastapi_db
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
api:
build: .
ports:
- "8000:8000"
environment:
DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/fastapi_db
SECRET_KEY: your-secret-key
depends_on:
- 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
- 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/