Pourquoi FastAPI avec PostgreSQL : guide pratique ?
Contexte réel : pourquoi un dev a besoin de ca au quotidien
FastAPI est une bibliothèque Python moderne pour construire des API web rapides, efficaces et faciles à maintenir. L'utilisation de PostgreSQL comme base de données relationnelle offre des avantages performants et robustes, ce qui rend FastAPI avec PostgreSQL un choix idéal pour les projets nécessitant un accès et une manipulation de données complexes.
Un cas d'usage concret : imaginez que vous travaillez sur une application de gestion de projet. Vous avez besoin d'une API pour gérer les utilisateurs, les projets, les tâches et leurs relations. FastAPI permet de créer rapidement des routes pour ces différentes entités, tandis que PostgreSQL offre un système robuste de stockage et de récupération des données.
Prerequis
- Connaissances en Python (version 3.7+ recommandée)
- Connaissance de base en SQL
- Installation d'un environnement virtuel (
python -m venv env) - Installation de FastAPI :
pip install fastapi - Installation de Starlette (dépendance de FastAPI) :
pip install starlette - Installation de SQLAlchemy (ORM pour PostgreSQL) :
pip install sqlalchemy - Installation de psycopg2 (driver Python pour PostgreSQL) :
pip install psycopg2
Concepts fondamentaux
1. Modèle de données avec SQLAlchemy ORM
SQLAlchemy est un ORM (Object Relational Mapping) qui permet de manipuler les bases de données via des objets Python. Voici un exemple simple d'un modèle de données pour une tâche.
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class Task(Base):
__tablename__ = 'tasks'
id = Column(Integer, primary_key=True)
title = Column(String(100), nullable=False)
description = Column(String(255))
completed = Column(Boolean, default=False)
project_id = Column(Integer, ForeignKey('projects.id'))
project = relationship("Project", back_populates="tasks")
class Project(Base):
__tablename__ = 'projects'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
tasks = relationship("Task", order_by=Task.id, back_populates="project")
2. Connexion à PostgreSQL avec SQLAlchemy
Pour se connecter à une base de données PostgreSQL, utilisez create_engine.
from sqlalchemy import create_engine
DATABASE_URL = "postgresql://username:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)
3. Création d'un FastAPI application
Créez un fichier main.py pour votre application FastAPI.
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import engine, get_db
from .models import Base, Task, Project
from .schemas import TaskCreate, TaskRead
Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.post("/tasks/", response_model=TaskRead)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
# Code à ajouter ici
4. Dépendances pour le gestionnaire de base de données
Créez un fichier database.py pour gérer la connexion et les dépendances.
from sqlalchemy.orm import sessionmaker
from .models import engine
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
5. Schémas de données avec Pydantic
Pydantic est utilisé pour définir les schémas de données.
from pydantic import BaseModel
class TaskCreate(BaseModel):
title: str
description: str = None
completed: bool = False
class TaskRead(BaseModel):
id: int
title: str
description: str
completed: bool
class Config:
orm_mode = True
Mise en pratique : projet fil rouge
Étape 1 : Configuration de la base de données
Créez un fichier database.py avec le code suivant :
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
class Item(Base):
__tablename__ = 'items'
id = Column(Integer, primary_key=True)
title = Column(String(100), nullable=False)
description = Column(String(255))
owner_id = Column(Integer, ForeignKey('users.id'))
owner = relationship("User", back_populates="items")
User.items = relationship("Item", order_by=Item.id, back_populates="owner")
Créez un fichier main.py pour votre application FastAPI :
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import engine, get_db, Base
from .models import User, Item
from .schemas import UserCreate, UserRead, ItemCreate, ItemRead
Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
fake_hashed_password = user.password + "notreallyhashed"
db_user = User(username=user.username, email=user.email, password=fake_hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.post("/items/", response_model=ItemRead)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
fake_hashed_password = item.description + "notreallyhashed"
db_item = Item(title=item.title, description=fake_hashed_password, owner_id=item.owner_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
Créez un fichier schemas.py pour définir les schémas de données :
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
email: str
password: str
class UserRead(BaseModel):
id: int
username: str
email: str
class Config:
orm_mode = True
class ItemCreate(BaseModel):
title: str
description: str
owner_id: int
class ItemRead(BaseModel):
id: int
title: str
description: str
owner_id: int
class Config:
orm_mode = True
Étape 2 : Création des routes
Ajoutez les routes pour lire les utilisateurs et les items.
@app.get("/users/", response_model=list[UserRead])
def read_users(db: Session = Depends(get_db)):
users = db.query(User).all()
return users
@app.get("/items/", response_model=list[ItemRead])
def read_items(db: Session = Depends(get_db)):
items = db.query(Item).all()
return items
Étape 3 : Exécution de l'application
Exécutez votre application avec la commande suivante :
uvicorn main:app --reload
Allez sur http://127.0.0.1:8000/docs pour accéder à l'interface Swagger UI et tester les routes.
Erreurs frequentes et debugging
1. Connexion échoue avec psycopg2
Code incorrect :
from sqlalchemy import create_engine
DATABASE_URL = "postgresql://username:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
Message d'erreur :
psycopg2.OperationalError: could not connect to server: Connection refused
Is the server running on host "localhost" and accepting connections on port 5432?
Code correct :
from sqlalchemy import create_engine
DATABASE_URL = "postgresql://username:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
2. Erreur de validation des données avec Pydantic
Code incorrect :
class UserCreate(BaseModel):
username: str
email: str
password: str
@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
fake_hashed_password = user.password + "notreallyhashed"
db_user = User(username=user.username, email=user.email, password=fake_hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
Message d'erreur :
ValidationError: 1 validation error for UserCreate
password
field required (type=value_error.missing)
Code correct :
class UserCreate(BaseModel):
username: str
email: str
password: str
@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
fake_hashed_password = user.password + "notreallyhashed"
db_user = User(username=user.username, email=user.email, password=fake_hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
3. Erreur de connexion à la base de données avec SQLAlchemy
Code incorrect :
from sqlalchemy.orm import sessionmaker
from .models import engine
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Message d'erreur :
SQLAlchemy.exc.OperationalError: (psycopg2.OperationalError) could not connect to server: Connection refused
Is the server running on host "localhost" and accepting connections on port 5432?
Code correct :
from sqlalchemy.orm import sessionmaker
from .models import engine
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Pour aller plus loin
- Intégration avec Auth0 pour l'authentification : Explorez comment utiliser Auth0 pour ajouter une couche d'authentification à votre application.
- Utilisation de Docker pour le déploiement : Apprenez à containeriser votre application FastAPI et PostgreSQL pour faciliter le déploiement sur un serveur.
- Migrations avec Alembic : Découvrez comment utiliser Alembic pour gérer les migrations de base de données.
Défi pratique : Ajoutez une fonctionnalité permettant d'ajouter des permissions aux utilisateurs pour accéder à certaines ressources de l'API.