Aller au contenu principal

Relations SQL avec FastAPI

Repères utiles
remarque

Ce chapitre est la suite logique de Authentification avec FastAPI. Il montre comment passer d'une API avec utilisateurs authentifiés à une API où les données sont réellement liées à ces utilisateurs.

Introduction​

Après Authentification avec FastAPI, l'API sait reconnaître un utilisateur. Mais tant qu'une tâche reste un objet isolé ou relié par un simple champ texte, l'authentification ne contrôle encore rien de concret en base. Le rôle de ce chapitre est donc de transformer l'identité utilisateur en ownership réel des données.

Ce que ce chapitre va accomplir​

À la fin de ce chapitre, tu auras une API où :

  • chaque tâche appartient à un utilisateur,
  • la base de données connaît cette relation,
  • un utilisateur peut récupérer ses propres tâches,
  • un utilisateur ne peut pas lire, modifier ou supprimer les tâches d'un autre utilisateur,
  • tu comprends la différence entre :
    • une simple colonne texte,
    • une clé étrangère,
    • une relation ORM,
    • et un vrai lien métier entre données.

Point de départ et choix pédagogiques​

Ce chapitre suppose que le modèle User, le JWT et get_current_user sont déjà en place. On s'appuie surtout sur 14-SQL Relations/81-models.py, 16-Delivery Partner/93-app.zip et 25-Many to Many/131-models.py.

Les sources parlent de logistique (Shipment, Seller, DeliveryPartner). Ici, on garde uniquement leur logique relationnelle pour l'appliquer à notre projet :

  • User,
  • Task.

Autrement dit, on ne change pas de domaine métier : on fait évoluer le même projet pour montrer comment une clé étrangère et une relation ORM changent le comportement réel de l'API.

Le problème du modèle actuel​

Jusqu'ici, la tâche ressemble encore à quelque chose comme ça :

class Task(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    description: str
    assignee: str
    status: Status_Task

Le problème de cette version, c'est que assignee est juste du texte.

Cela veut dire que :

  • la base ne sait pas si ce nom correspond à un vrai utilisateur,
  • il n'y a aucune garantie d'intégrité,
  • on ne peut pas facilement dire :
    • “montre-moi toutes les tâches de cet utilisateur”
    • “empêche les autres utilisateurs de modifier cette tâche”

La vraie idée à comprendre​

Une relation SQL, ce n'est pas juste “deux objets qui se connaissent”.

C'est d'abord :

  • une clé étrangère en base de données,
  • puis une relation ORM qui rend cette relation agréable à manipuler dans Python.

Dans notre cas, on veut exprimer :

  • un User peut avoir plusieurs tâches,
  • une Task appartient à un seul utilisateur.

C'est une relation one-to-many.

Architecture visée après ce chapitre​

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

Bonne nouvelle : la structure générale ne change pas.

Ce qui change, c'est surtout :

  • le modèle Task,
  • le modèle User,
  • les schémas Task,
  • le service TaskService,
  • les routes task.

Repères utiles avant le code​

Avant de coder, retiens ces idées simples :

  • owner_id est la clé étrangère qui relie la tâche à l'utilisateur.
  • owner est la relation ORM qui te permet d'accéder à l'objet utilisateur.
  • tasks côté User est la relation inverse.
  • Relationship(...) n'est pas magique : il s'appuie sur la clé étrangère réelle.
  • l'ownership doit vivre dans la base, pas seulement dans la logique applicative.

Étape 1 — faire évoluer database/models.py​

Ici, on remplace l'idée d'un simple assignee texte par une vraie appartenance à un utilisateur.

from enum import Enum

from pydantic import EmailStr
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, Relationship, SQLModel


class Status_Task(str, Enum):
    to_do = "to_do"
    in_progress = "in_progress"
    completed = "completed"


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

    tasks: list["Task"] = Relationship(
        back_populates="owner",
        sa_relationship_kwargs={"lazy": "selectin"},
    )


class Task(SQLModel, table=True):
    __tablename__ = "task"

    id: int | None = Field(default=None, primary_key=True)
    title: str
    description: str
    status: Status_Task = Field(
        sa_type=SAEnum(Status_Task, native_enum=False),
        default=Status_Task.to_do,
    )

    owner_id: int = Field(foreign_key="user.id")
    owner: "User" = Relationship(
        back_populates="tasks",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

Pourquoi cette version est plus forte​

Parce qu'elle dit explicitement à la base de données :

  • chaque tâche a un owner_id,
  • cet owner_id doit pointer vers user.id,
  • un utilisateur possède une liste de tâches,
  • une tâche connaît son propriétaire.

Ce n'est plus du texte libre. C'est une structure relationnelle réelle.

Étape 2 — faire évoluer les schémas dans api/schemas/task.py​

Maintenant que la tâche appartient à un utilisateur, on adapte les schémas API.

from pydantic import BaseModel, Field

from database.models import Status_Task


class BaseTask(BaseModel):
    title: str
    description: str


class TaskRead(BaseTask):
    id: int
    status: Status_Task
    owner_id: int


class TaskCreate(BaseTask):
    pass


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None)
    description: str | None = Field(default=None)
    status: Status_Task | None = Field(default=None)

Pourquoi enlever assignee ?​

Parce qu'une fois qu'on a un vrai owner_id, garder un assignee: str devient souvent plus confus qu'utile dans ce mini-projet.

Ici, le but du chapitre est justement de faire sentir la différence entre :

  • une donnée libre,
  • une donnée relationnelle.

Donc on simplifie et on rend l'appartenance explicite.

Étape 3 — faire évoluer services/task.py​

C'est ici que la relation devient utile dans la logique métier.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.schemas.task import TaskCreate
from database.models import Task, User


class TaskService:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_task(self, id: int) -> Task | None:
        return await self.session.get(Task, id)

    async def get_task_for_user(self, task_id: int, user_id: int) -> Task | None:
        result = await self.session.execute(
            select(Task).where(Task.id == task_id, Task.owner_id == user_id)
        )
        return result.scalar_one_or_none()

    async def get_tasks_for_user(self, user_id: int) -> list[Task]:
        result = await self.session.execute(
            select(Task).where(Task.owner_id == user_id)
        )
        return list(result.scalars().all())

    async def add_task(self, task_create: TaskCreate, owner: User) -> Task:
        new_task = Task(
            **task_create.model_dump(),
            owner_id=owner.id,
        )
        self.session.add(new_task)
        await self.session.commit()
        await self.session.refresh(new_task)
        return new_task

    async def update_task(self, task: Task, task_update: dict) -> Task:
        task.sqlmodel_update(task_update)
        self.session.add(task)
        await self.session.commit()
        await self.session.refresh(task)
        return task

    async def delete_task(self, task: Task) -> None:
        await self.session.delete(task)
        await self.session.commit()

Ce qu'il faut remarquer ici​

On a introduit deux idées très importantes :

1. get_tasks_for_user​

Cette méthode permet de dire :

montre-moi toutes les tâches d'un utilisateur donné

2. get_task_for_user​

Cette méthode permet de dire :

récupère cette tâche seulement si elle appartient bien à cet utilisateur

C'est l'un des points les plus importants du chapitre.

L'ownership n'est pas juste une information décorative. Elle devient une condition d'accès aux données.

Étape 4 — garder api/dependencies.py quasiment identique​

Bonne nouvelle : l'infrastructure auth que tu as mise en place au chapitre précédent reste bonne.

Tu peux garder la logique suivante :

  • TaskServiceDep
  • CurrentUserDep

Donc ici, il n'y a pas besoin de tout réécrire.

Le vrai gain d'une bonne architecture, c'est justement ça :

  • on ajoute une relation,
  • mais on ne casse pas toute l'application.

Étape 5 — faire évoluer api/routers/task.py​

Maintenant, on veut que l'utilisateur travaille avec ses tâches.

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("/mine", response_model=list[TaskRead])
async def list_my_tasks(
    service: TaskServiceDep,
    current_user: CurrentUserDep,
):
    return await service.get_tasks_for_user(current_user.id)


@router.get("/", response_model=TaskRead)
async def get(
    id: int,
    service: TaskServiceDep,
    current_user: CurrentUserDep,
):
    task = await service.get_task_for_user(id, current_user.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, current_user)


@router.patch("/", response_model=TaskRead)
async def update(
    id: int,
    task_update: TaskUpdate,
    service: TaskServiceDep,
    current_user: CurrentUserDep,
):
    task = await service.get_task_for_user(id, current_user.id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )

    update_data = task_update.model_dump(exclude_unset=True)
    return await service.update_task(task, update_data)


@router.delete("/")
async def delete(
    id: int,
    service: TaskServiceDep,
    current_user: CurrentUserDep,
):
    task = await service.get_task_for_user(id, current_user.id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found",
        )

    await service.delete_task(task)
    return {"message": f"Task with ID {id} has been deleted successfully"}

Pourquoi cette version est très importante pédagogiquement​

Parce que maintenant l'authentification change réellement le comportement des routes.

Avant, le token servait surtout à protéger l'entrée. Maintenant, le token sert aussi à déterminer :

  • quelles données voir,
  • quelles données modifier,
  • quelles données supprimer.

On passe donc de :

  • une API avec auth à
  • une API avec ownership réel.

Étape 6 — vérifier que main.py et database/session.py restent sains​

Ici encore, pas besoin de révolution.

Tant que :

  • database/models.py contient bien User et Task,
  • create_db_tables() fait un SQLModel.metadata.create_all,
  • main.py inclut toujours le router principal,

alors la nouvelle structure relationnelle sera prise en compte au démarrage.

Vérification guidée​

Lance ton application :

uvicorn main:app --reload

Test 1 — créer deux utilisateurs​

Utilisateur 1​

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"
  }'

Utilisateur 2​

curl -X POST http://127.0.0.1:8000/user/signup \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Alice",
    "email": "alice@example.com",
    "password": "StrongPass456"
  }'

Test 2 — récupérer deux tokens​

Token Enes​

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"
TOKEN_ENES="colle_ici_le_token_enes"

Token Alice​

curl -X POST http://127.0.0.1:8000/user/token \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=alice@example.com&password=StrongPass456"
TOKEN_ALICE="colle_ici_le_token_alice"

Test 3 — créer une tâche avec Enes​

curl -X POST http://127.0.0.1:8000/task/ \
  -H "Authorization: Bearer $TOKEN_ENES" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Task de Enes",
    "description": "Cette tâche doit appartenir à Enes"
  }'

Réponse attendue :

{
  "id": 1,
  "title": "Task de Enes",
  "description": "Cette tâche doit appartenir à Enes",
  "status": "to_do",
  "owner_id": 1
}

Test 4 — lister les tâches de Enes​

curl http://127.0.0.1:8000/task/mine \
  -H "Authorization: Bearer $TOKEN_ENES"

Tu dois voir la tâche créée.

Test 5 — lister les tâches de Alice​

curl http://127.0.0.1:8000/task/mine \
  -H "Authorization: Bearer $TOKEN_ALICE"

Réponse attendue :

  • une liste vide, ou en tout cas pas la tâche de Enes.

Test 6 — Alice essaie de lire la tâche de Enes​

curl "http://127.0.0.1:8000/task/?id=1" \
  -H "Authorization: Bearer $TOKEN_ALICE"

Réponse attendue :

{
  "detail": "Task not found"
}

C'est exactement le comportement que l'on voulait.

La tâche existe bien, mais pas pour Alice.

Test 7 — vérifier directement la relation en base​

psql taskapi_db -c "SELECT id, title, owner_id, status FROM task;"

Tu dois maintenant voir explicitement owner_id dans la table.

Pourquoi retourner 404 ici plutôt que 403 ?​

C'est un choix de design assez courant.

Quand Alice demande la tâche de Enes, on peut considérer que :

  • soit elle n'a pas le droit de la voir,
  • soit, du point de vue de son périmètre d'accès, cette ressource “n'existe pas”.

Dans beaucoup d'APIs, renvoyer 404 évite de révéler l'existence de ressources appartenant à d'autres utilisateurs.

Bonus important — ce que t'enseignent les sources 16-Delivery Partner​

Le dossier 16-Delivery Partner montre une idée très utile : une entité peut avoir plusieurs relations différentes.

Exemple conceptuel observé dans les sources :

  • un Shipment appartient à un Seller,
  • mais il est aussi relié à un DeliveryPartner.

Cela veut dire qu'après avoir compris User -> Task, tu es déjà prêt pour des modèles plus riches du type :

  • Task liée à un Project
  • Task liée à un User
  • Task liée à une Category

Le pattern est le même.

Bonus 2 — aperçu du many-to-many​

Le dossier 25-Many to Many va encore plus loin avec une table d'association.

Dans un projet de tâches, un exemple naturel serait :

  • une tâche peut avoir plusieurs labels,
  • un label peut être utilisé par plusieurs tâches.

Schéma conceptuel :

from sqlmodel import Field, Relationship, SQLModel


class TaskLabelLink(SQLModel, table=True):
    task_id: int = Field(foreign_key="task.id", primary_key=True)
    label_id: int = Field(foreign_key="label.id", primary_key=True)


class Label(SQLModel, table=True):
    __tablename__ = "label"

    id: int | None = Field(default=None, primary_key=True)
    name: str

    tasks: list["Task"] = Relationship(
        back_populates="labels",
        link_model=TaskLabelLink,
    )


class Task(SQLModel, table=True):
    __tablename__ = "task"

    id: int | None = Field(default=None, primary_key=True)
    title: str
    description: str
    owner_id: int = Field(foreign_key="user.id")

    labels: list[Label] = Relationship(
        back_populates="tasks",
        link_model=TaskLabelLink,
    )

Ce n'est pas ce qu'on implémente maintenant, mais c'est la suite naturelle une fois les relations simples bien comprises.

Pièges et erreurs fréquentes​

  • garder un assignee: str tout en croyant avoir déjà une vraie relation utilisateur
  • oublier foreign_key="user.id"
  • croire que Relationship(...) suffit sans clé étrangère
  • oublier d'adapter les schémas API après modification des modèles
  • ne filtrer les tâches que dans les routes sans centraliser la logique dans le service
  • mélanger auth et relation : le token identifie l'utilisateur, mais c'est la relation SQL qui relie réellement la donnée à cet utilisateur

Bonnes pratiques à retenir​

  • faire vivre l'ownership dans la base de données, pas seulement dans le code applicatif
  • écrire des méthodes de service explicites comme get_task_for_user()
  • garder des routes simples et lisibles
  • rendre la relation visible dans le schéma de réponse (owner_id)
  • avancer du one-to-many vers le many-to-many, pas l'inverse

Variantes et évolutions possibles​

À partir d'ici, tu peux naturellement évoluer vers :

  • tâches par utilisateur avec pagination,
  • Project → Task → User,
  • labels many-to-many,
  • permissions plus fines,
  • rôles,
  • historique d'activité,
  • politiques d'accès plus complexes.

Checklist de validation​

Avant de passer à la suite, vérifie que tu peux :

  • expliquer le rôle de owner_id, foreign_key et Relationship
  • montrer comment une tâche est rattachée à un utilisateur
  • expliquer à quoi sert get_task_for_user()
  • appeler /task/mine et comprendre ce qu’il renvoie
  • expliquer le choix entre 404 et 403 dans ce contexte

Résumé final​

Dans ce chapitre, tu as appris à faire le vrai passage entre :

  • une API avec utilisateurs,
  • et une API où les données appartiennent réellement à ces utilisateurs.

Tu as vu comment :

  • créer une relation User -> Task,
  • utiliser foreign_key et Relationship,
  • adapter les schémas,
  • filtrer les données par utilisateur,
  • rendre l'ownership concret dans les routes et dans le service.

Si tu retiens une seule chose, retiens celle-ci :

l'authentification dit qui appelle l'API ; la relation SQL dit à quelles données cette personne est réellement liée.

Pour aller plus loin​