Authentification avec FastAPI
Ce chapitre est le premier vrai sous-bloc de Advanced FastAPI. Il montre comment faire évoluer le socle posé dans API de base vers une API avec register, login, protection par token et compréhension claire du logout.
Introduction
API de base t'a laissé avec une API CRUD propre, mais encore ouverte : dès qu'un client connaît l'URL, il peut appeler les endpoints sensibles. Le rôle de ce chapitre est d'ajouter la première vraie frontière applicative : savoir qui appelle l'API avant d'autoriser certaines actions.
Ce que ce chapitre va ajouter
À la fin de ce chapitre, tu auras :
- une table
User, - une route d'inscription,
- une route de login qui renvoie un JWT,
- une dépendance
get_current_user, - une route
/user/me, - des routes d'écriture sur les tâches protégées par authentification,
- une compréhension claire de la différence entre login, token, route protégée et logout.
Point de départ et choix pédagogiques
Ce chapitre suppose que tu pars du projet construit dans API de base. On ne repart donc pas de zéro : on étend le socle existant.
Les sources de référence viennent des blocs 11-Register User, 12-Login User et 13-Logout User.
On en reprend la logique essentielle :
- hasher le mot de passe avant l'écriture en base,
- renvoyer un access token au login,
- valider le JWT via une couche sécurité dédiée,
- expliquer séparément le vrai sujet du logout.
Pour garder le tutoriel lisible, on fait deux adaptations volontaires :
- remplacer le domaine
sellerdes sources par un domaine plus générique :user - faire évoluer le routage vers
api/routers/pour accompagner la croissance du projet
Pourquoi ce chapitre arrive maintenant
L'authentification est la première brique avancée logique parce qu'elle change immédiatement la lecture de toute l'API :
- les routes d'écriture ne sont plus publiques,
- certaines réponses dépendent de l'utilisateur courant,
- les chapitres suivants peuvent enfin parler d'ownership, de permissions et de cycle utilisateur.
Nouvelle architecture visée
Après l'ajout de l'auth, l'arborescence devient :
fastapi-base-api/
├── main.py
├── config.py
├── requirements.txt
├── .gitignore
├── .env
├── .env.example
├── core/
│ ├── __init__.py
│ └── security.py
├── api/
│ ├── __init__.py
│ ├── router.py
│ ├── dependencies.py
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── task.py
│ │ └── user.py
│ └── schemas/
│ ├── __init__.py
│ ├── task.py
│ └── user.py
├── database/
│ ├── __init__.py
│ ├── models.py
│ └── session.py
└── services/
├── __init__.py
├── task.py
└── user.py
Repères utiles avant le code
Avant de commencer, retiens ceci :
Userest un modèle de base de données, pas un token.- le mot de passe ne doit jamais être stocké en clair.
- le token JWT n'est pas le mot de passe : c'est une preuve temporaire que l'utilisateur s'est authentifié.
- une route protégée ne lit pas directement le header
Authorizationpartout dans le code : elle passe par une dépendance dédiée. - le logout en JWT n'est pas aussi simple qu'avec une session serveur classique.
Étape 1 — ajouter les dépendances d'authentification
Tu peux partir de ton requirements.txt de API de base et y ajouter les dépendances suivantes :
fastapi
uvicorn[standard]
sqlmodel
sqlalchemy
asyncpg
pydantic-settings
scalar-fastapi
passlib[bcrypt]
PyJWT
python-multipart
email-validator
Pourquoi ces dépendances ?
passlib[bcrypt]: pour hasher et vérifier les mots de passe.PyJWT: pour créer et décoder les JWT.python-multipart: nécessaire pourOAuth2PasswordRequestForm.email-validator: nécessaire si on veut utiliserEmailStrdans les schémas Pydantic.
Installe ensuite :
pip install -r requirements.txt
Étape 2 — préparer les nouveaux fichiers
Ajoute maintenant les nouveaux fichiers :
mkdir -p core api/routers
touch core/__init__.py core/security.py
touch api/routers/__init__.py api/routers/task.py api/routers/user.py
touch api/schemas/user.py
touch services/user.py
Étape 3 — étendre .env.example et .env
L'auth a besoin d'une configuration de sécurité.
.env.example
POSTGRES_SERVER=localhost
POSTGRES_PORT=5432
POSTGRES_DB=taskapi_db
POSTGRES_USER=postgres
POSTGRES_PASSWORD=change_me
JWT_SECRET=change_me_with_a_long_random_value
JWT_ALGORITHM=HS256
.env
POSTGRES_SERVER=localhost
POSTGRES_PORT=5432
POSTGRES_DB=taskapi_db
POSTGRES_USER=postgres
POSTGRES_PASSWORD=your_real_password
JWT_SECRET=your_long_random_secret
JWT_ALGORITHM=HS256
Tu peux générer rapidement un secret avec Python :
python -c "import secrets; print(secrets.token_hex(32))"
JWT_SECRET ne doit pas être versionné, ni partagé publiquement.
Étape 4 — mettre à jour config.py
On garde l'idée du chapitre de base, mais on ajoute maintenant une section sécurité.
from pydantic_settings import BaseSettings, SettingsConfigDict
_base_config = SettingsConfigDict(
env_file=".env",
env_ignore_empty=True,
extra="ignore",
)
class DatabaseSettings(BaseSettings):
POSTGRES_SERVER: str
POSTGRES_PORT: int
POSTGRES_USER: str
POSTGRES_PASSWORD: str
POSTGRES_DB: str
model_config = _base_config
@property
def POSTGRES_URL(self) -> str:
return (
f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
f"@{self.POSTGRES_SERVER}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
)
class SecuritySettings(BaseSettings):
JWT_SECRET: str
JWT_ALGORITHM: str
model_config = _base_config
# Instances prêtes à l'emploi
db_settings = DatabaseSettings()
security_settings = SecuritySettings()
Pourquoi ce changement ?
Dans API de base, config.py ne gérait que la base de données.
Maintenant, l'application a aussi besoin d'une configuration de sécurité.
Séparer db_settings et security_settings rend le code plus clair.
Étape 5 — créer la couche sécurité dans core/security.py
Ici, on regroupe :
- le hash de mot de passe,
- la vérification du mot de passe,
- la création de JWT,
- le décodage des JWT,
- le schéma OAuth2 utilisé par FastAPI.
from datetime import datetime, timedelta, timezone
import jwt
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from config import security_settings
password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/token")
def hash_password(password: str) -> str:
return password_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return password_context.verify(plain_password, hashed_password)
def create_access_token(data: dict, expiry: timedelta = timedelta(days=7)) -> str:
payload = {
**data,
"exp": datetime.now(timezone.utc) + expiry,
}
return jwt.encode(
payload=payload,
key=security_settings.JWT_SECRET,
algorithm=security_settings.JWT_ALGORITHM,
)
def decode_access_token(token: str) -> dict | None:
try:
return jwt.decode(
jwt=token,
key=security_settings.JWT_SECRET,
algorithms=[security_settings.JWT_ALGORITHM],
)
except jwt.PyJWTError:
return None
Pourquoi OAuth2PasswordBearer ?
Les sources montrent plusieurs variantes, dont une approche HTTPBearer et une approche liée à OAuth2PasswordBearer.
Ici, on choisit OAuth2PasswordBearer parce que :
- c'est très bien intégré à FastAPI,
- la doc
/docsle comprend bien, - cela reste cohérent avec une route
/user/token.
Étape 6 — étendre database/models.py
On ajoute maintenant un vrai modèle User.
from enum import Enum
from pydantic import EmailStr
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, SQLModel
class Status_Task(str, Enum):
to_do = "to_do"
in_progress = "in_progress"
completed = "completed"
class Task(SQLModel, table=True):
__tablename__ = "task"
id: int | None = Field(default=None, primary_key=True)
title: str
description: str
assignee: str
status: Status_Task = Field(
sa_type=SAEnum(Status_Task, native_enum=False),
default=Status_Task.to_do,
)
class User(SQLModel, table=True):
__tablename__ = "user"
id: int | None = Field(default=None, primary_key=True)
name: str
email: EmailStr = Field(index=True)
password_hash: str
Pourquoi password_hash et pas password ?
Parce qu'on ne stocke jamais un mot de passe brut en base. On stocke seulement une version hashée.
Étape 7 — créer les schémas utilisateur dans api/schemas/user.py
from pydantic import BaseModel, EmailStr
class BaseUser(BaseModel):
name: str
email: EmailStr
class UserRead(BaseUser):
id: int
class UserCreate(BaseUser):
password: str
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
Pourquoi ces schémas ?
UserCreatesert à l'inscription.UserReadsert à ce que l'API renvoie au client.TokenResponsesert à standardiser la réponse du login.
Étape 8 — créer services/user.py
C'est ici que la logique d'inscription et d'authentification va vivre.
from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from api.schemas.user import UserCreate
from core.security import create_access_token, hash_password, verify_password
from database.models import User
class UserService:
def __init__(self, session: AsyncSession):
self.session = session
async def get_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def get_by_id(self, user_id: int) -> User | None:
return await self.session.get(User, user_id)
async def add_user(self, credentials: UserCreate) -> User:
existing_user = await self.get_by_email(credentials.email)
if existing_user is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already registered",
)
user = User(
**credentials.model_dump(exclude={"password"}),
password_hash=hash_password(credentials.password),
)
self.session.add(user)
await self.session.commit()
await self.session.refresh(user)
return user
async def authenticate(self, email: str, password: str) -> str:
user = await self.get_by_email(email)
if user is None or not verify_password(password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Email or password is incorrect",
)
return create_access_token(
data={
"sub": str(user.id),
"email": user.email,
"name": user.name,
}
)
Ce qu'il faut remarquer
- l'inscription vérifie les doublons,
- le mot de passe est hashé avant l'écriture,
- le login ne renvoie pas l'utilisateur directement,
- le login renvoie un JWT.
Étape 9 — faire évoluer api/dependencies.py
Cette étape est clé : c'est ici qu'on apprend à récupérer l'utilisateur courant à partir du token.
from typing import Annotated
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from core.security import decode_access_token, oauth2_scheme
from database.models import User
from database.session import get_session
from services.task import TaskService
from services.user import UserService
SessionDep = Annotated[AsyncSession, Depends(get_session)]
TokenDep = Annotated[str, Depends(oauth2_scheme)]
def get_task_service(session: SessionDep) -> TaskService:
return TaskService(session)
def get_user_service(session: SessionDep) -> UserService:
return UserService(session)
TaskServiceDep = Annotated[TaskService, Depends(get_task_service)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]
async def get_current_user(token: TokenDep, service: UserServiceDep) -> User:
token_data = decode_access_token(token)
if token_data is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)
user_id = token_data.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload",
)
user = await service.get_by_id(int(user_id))
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
)
return user
CurrentUserDep = Annotated[User, Depends(get_current_user)]
Ce que fait réellement get_current_user
- récupère le token transmis dans le header
Authorization: Bearer ... - décode le JWT
- extrait l'identifiant utilisateur
- recharge l'utilisateur depuis la base
- échoue proprement en
401si quelque chose ne va pas
Étape 10 — faire évoluer le routage
Dans API de base, un seul api/router.py suffisait.
Maintenant que l'application grossit, on passe à des sous-routes.
api/router.py
from fastapi import APIRouter
from .routers import task, user
router = APIRouter()
router.include_router(task.router)
router.include_router(user.router)
api/routers/user.py
from typing import Annotated
from fastapi import APIRouter, Depends, status
from fastapi.security import OAuth2PasswordRequestForm
from api.dependencies import CurrentUserDep, UserServiceDep
from api.schemas.user import TokenResponse, UserCreate, UserRead
router = APIRouter(prefix="/user", tags=["User"])
@router.post("/signup", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def register_user(user: UserCreate, service: UserServiceDep):
return await service.add_user(user)
@router.post("/token", response_model=TokenResponse)
async def login_user(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
service: UserServiceDep,
):
token = await service.authenticate(form_data.username, form_data.password)
return {
"access_token": token,
"token_type": "bearer",
}
@router.get("/me", response_model=UserRead)
async def get_me(current_user: CurrentUserDep):
return current_user
api/routers/task.py
from fastapi import APIRouter, HTTPException, status
from api.dependencies import CurrentUserDep, TaskServiceDep
from api.schemas.task import TaskCreate, TaskRead, TaskUpdate
router = APIRouter(prefix="/task", tags=["Task"])
@router.get("/", response_model=TaskRead)
async def get(id: int, service: TaskServiceDep):
task = await service.get_task(id)
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
return task
@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
async def create(
task_create: TaskCreate,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
return await service.add_task(task_create)
@router.patch("/", response_model=TaskRead)
async def update(
id: int,
task_update: TaskUpdate,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
update_data = task_update.model_dump(exclude_unset=True)
task = await service.update_task(id, update_data)
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
return task
@router.delete("/")
async def delete(
id: int,
service: TaskServiceDep,
current_user: CurrentUserDep,
):
deleted = await service.delete_task(id)
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
return {"message": f"Task with ID {id} has been deleted successfully"}
Pourquoi protéger seulement les routes d'écriture ?
Pour ce premier chapitre auth, c'est un bon compromis pédagogique.
On garde :
GET /task/ouvert pour rester simple,POST,PATCH,DELETEprotégés pour rendre l'auth immédiatement concrète.
Plus tard, tu pourras aller plus loin avec :
- propriété des tâches,
- rôles,
- permissions fines,
- filtrage par utilisateur courant.
Étape 11 — mettre à jour database/session.py
La logique reste la même, mais il faut maintenant importer db_settings.
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel
from config import db_settings
engine = create_async_engine(
url=db_settings.POSTGRES_URL,
echo=True,
)
async_session = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def create_db_tables():
async with engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
async def get_session():
async with async_session() as session:
yield session
Étape 12 — main.py reste presque identique
Le cœur de main.py ne change pas vraiment. C'est un bon signal :
quand l'architecture est saine, on peut faire grandir l'app sans réécrire tout le point d'entrée.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from scalar_fastapi import get_scalar_api_reference
from api.router import router
from database.session import create_db_tables
@asynccontextmanager
async def lifespan(app: FastAPI):
await create_db_tables()
yield
app = FastAPI(
title="Task Organizer API",
description="Une API FastAPI avec authentification",
lifespan=lifespan,
)
app.include_router(router)
@app.get("/scalar", include_in_schema=False)
def get_scalar_docs():
return get_scalar_api_reference(
openapi_url=app.openapi_url,
title="Scalar API",
)
Vérification guidée
Lance l'application :
uvicorn main:app --reload
Ouvre ensuite :
http://127.0.0.1:8000/docshttp://127.0.0.1:8000/scalar
Test 1 — inscrire un utilisateur
curl -X POST http://127.0.0.1:8000/user/signup \
-H "Content-Type: application/json" \
-d '{
"name": "Enes",
"email": "enes@example.com",
"password": "StrongPass123"
}'
Réponse attendue :
{
"name": "Enes",
"email": "enes@example.com",
"id": 1
}
Test 2 — se connecter
curl -X POST http://127.0.0.1:8000/user/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=enes@example.com&password=StrongPass123"
Réponse attendue :
{
"access_token": "...",
"token_type": "bearer"
}
Copie le token dans une variable shell :
TOKEN="colle_ici_le_token"
Test 3 — appeler /user/me
curl http://127.0.0.1:8000/user/me \
-H "Authorization: Bearer $TOKEN"
Réponse attendue :
{
"name": "Enes",
"email": "enes@example.com",
"id": 1
}
Test 4 — essayer de créer une tâche sans token
curl -X POST http://127.0.0.1:8000/task/ \
-H "Content-Type: application/json" \
-d '{
"title": "Tâche protégée",
"description": "Doit échouer sans token",
"assignee": "Enes"
}'
Réponse attendue : 401 Unauthorized.
Test 5 — créer une tâche avec token
curl -X POST http://127.0.0.1:8000/task/ \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{
"title": "Tâche protégée",
"description": "Créée avec un utilisateur connecté",
"assignee": "Enes"
}'
Réponse attendue : 201 Created avec le JSON de la tâche.
Et le logout ?
C'est ici qu'il faut être très clair.
Version minimale
Avec un JWT pur, le logout le plus simple consiste à :
- supprimer le token côté client,
- ne plus l'envoyer.
C'est souvent suffisant pour un premier niveau de compréhension.
Version plus robuste côté serveur
Les sources du dossier 13-Logout User montrent une approche plus avancée :
- chaque token reçoit un
jti, - un store Redis garde la liste des tokens invalidés,
- le backend refuse ensuite les tokens blacklistés.
Option avancée — vrai logout avec blacklist Redis
Cette option est facultative pour ce premier passage, mais elle correspond à une vraie montée en puissance.
Dépendance supplémentaire
Ajoute :
redis
Variables supplémentaires
Dans .env.example :
REDIS_HOST=localhost
REDIS_PORT=6379
Ajouter un jti dans core/security.py
Version inspirée des sources :
from datetime import datetime, timedelta, timezone
from uuid import uuid4
import jwt
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from config import security_settings
password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/token")
def create_access_token(data: dict, expiry: timedelta = timedelta(days=7)) -> str:
payload = {
**data,
"jti": str(uuid4()),
"exp": datetime.now(timezone.utc) + expiry,
}
return jwt.encode(
payload=payload,
key=security_settings.JWT_SECRET,
algorithm=security_settings.JWT_ALGORITHM,
)
Créer database/redis.py
from redis.asyncio import Redis
_token_blacklist = Redis(host="localhost", port=6379, db=0)
async def add_jti_to_blacklist(jti: str):
await _token_blacklist.set(jti, "blacklisted")
async def is_jti_blacklisted(jti: str) -> bool:
return bool(await _token_blacklist.exists(jti))
Ajouter une dépendance pour récupérer le payload du token
Dans api/dependencies.py, tu peux ajouter :
from typing import Annotated
from fastapi import Depends, HTTPException, status
async def get_token_payload(token: TokenDep) -> dict:
token_data = decode_access_token(token)
if token_data is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)
return token_data
TokenPayloadDep = Annotated[dict, Depends(get_token_payload)]
Ajouter une vraie route de logout
from fastapi import APIRouter, Depends
from api.dependencies import CurrentUserDep, TokenPayloadDep
from database.redis import add_jti_to_blacklist
router = APIRouter(prefix="/user", tags=["User"])
@router.post("/logout")
async def logout_user(token_data: TokenPayloadDep, current_user: CurrentUserDep):
await add_jti_to_blacklist(token_data["jti"])
return {"detail": "Successfully logged out"}
Pour rester simple, tu peux très bien garder le logout côté client dans un premier temps, puis revenir plus tard sur cette version Redis quand tu travailleras plus sérieusement la sécurité et l'invalidation de tokens.
Pièges et erreurs fréquentes
- oublier
python-multipartpuis se demander pourquoi le login form ne marche pas - stocker le mot de passe brut au lieu d'un hash
- croire que le JWT remplace la base de données
- oublier de vérifier le token avant d'accéder à une route protégée
- oublier d'ajouter
JWT_SECRETdans.env - oublier d'envoyer
Authorization: Bearer <token> - protéger toutes les routes d'un coup sans garder une progression pédagogique lisible
Bonnes pratiques à retenir
- centraliser la sécurité dans
core/security.py - garder l'authentification dans un service dédié (
UserService) - séparer clairement : inscription, login, utilisateur courant
- faire évoluer l'arborescence quand le projet grossit
- protéger d'abord quelques routes critiques avant d'introduire permissions et rôles plus fins
- toujours expliquer la différence entre password, password_hash et access_token
Variantes et évolutions possibles
À partir de ce chapitre, tu peux ensuite avancer vers :
- relations SQL entre utilisateurs et tâches,
- ownership des tâches,
- rôles et permissions,
- reset de mot de passe,
- confirmation d'email,
- middleware de sécurité,
- tests auth automatisés,
- packaging Docker incluant Redis.
Checklist de validation
Avant de passer à la suite, vérifie que tu peux :
- expliquer la différence entre inscription, login, utilisateur courant et logout
- montrer où sont gérés le hash de mot de passe et la génération du JWT
- appeler correctement la route de token
- protéger une route avec l’utilisateur courant
- expliquer pourquoi un logout stateless ne révoque pas automatiquement un JWT
Résumé final
Dans ce chapitre, tu as ajouté une vraie première couche d'authentification à ton API FastAPI.
Tu as vu comment :
- créer un modèle
User, - hasher les mots de passe,
- générer un JWT,
- récupérer l'utilisateur courant,
- protéger les routes sensibles,
- comprendre ce que veut vraiment dire
logoutavec des tokens.
Si tu retiens une seule chose, retiens celle-ci :
l'authentification n'est pas juste une route
/token; c'est une manière de faire entrer l'identité dans toute l'architecture de ton API.
Pour aller plus loin
- Parcours de lecture — pour replacer ce chapitre dans le bon parcours
- FAQ, erreurs fréquentes et conseils pratiques — pour retrouver les pièges d’auth les plus probables
- Advanced FastAPI — pour garder la vue d’ensemble du bloc avancé
- Relations SQL avec FastAPI — pour relier maintenant les tâches à l’utilisateur authentifié