Intégration de base de données FastAPI

FastAPI peut être intégré à diverses bases de données, et l'approche la plus courante consiste à utiliser SQLAlchemy comme ORM. Cette section présente comment construire une API avec les fonctionnalités CRUD (Create, Read, Update, Delete) en utilisant FastAPI + SQLAlchemy.

Installer les dépendances

pip install sqlalchemy

Structure du projet

project/
├── main.py          # FastAPI 应用入口
├── database.py      # 数据库连接配置
├── models.py        # SQLAlchemy 模型
├── schemas.py       # Pydantic 数据模型
└── crud.py          # 数据库操作函数

1. Configuration de la base de données

Dansdatabase.pyConfigurez la connexion à la base de données dans :

Exemple

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# Base de données SQLite (adaptée pour le développement)
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# Créer le moteur de base de données
# connect_args n'est nécessaire que pour SQLite, ce qui autorise l'accès multi-thread
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)

# Créer la fabrique de sessions
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Déclarer la classe de base
class Base(DeclarativeBase):
pass

Pour l'environnement de développement, SQLite suffit. C'est une base de données basée sur des fichiers qui ne nécessite aucune installation supplémentaire de service de base de données. Pour l'environnement de production, PostgreSQL ou MySQL est recommandé ; il vous suffit de modifier leSQLALCHEMY_DATABASE_URLC'est tout.

2. Modèle SQLAlchemy

Dansmodels.pyDéfinissez la structure de la table de base de données dans :

Exemple

# models.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
__tablename__ = "users" # Nom de la table

id = Column(Integer, primary_key=True, index=True) # Clé primaire, auto-incrémentation
email = Column(String, unique=True, index=True) # Unique, créer un index
hashed_password = Column(String) # Mot de passe haché
is_active = Column(Boolean, default=True) # Est actif

class Item(Base):
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True) # Titre, créer un index
description = Column(String) # Description
owner_id = Column(Integer, ForeignKey("users.id")) # Clé étrangère référençant la table users

3. Modèles de données Pydantic

Dansschemas.pydéfinition des structures de données d'entrée/sortie de l'API :

Exemple

# schemas.py
from pydantic import BaseModel

# ===== Modèle User =====
class UserBase(BaseModel):
email: str

class UserCreate(UserBase):
password: str # Mot de passe requis lors de la création

class UserOut(UserBase):
id: int
is_active: bool

model_config = ConfigDict(from_attributes=True) # Prend en charge les objets ORM

# ===== Modèle Product =====
class ItemBase(BaseModel):
title: str
description: str | None = None

class ItemCreate(ItemBase):
pass

class ItemOut(ItemBase):
id: int
owner_id: int

model_config = ConfigDict(from_attributes=True)

Les modèles Pydantic et les modèles SQLAlchemy sont distincts. Pydantic est responsable de la validation des données au niveau de la couche API, tandis que SQLAlchemy est responsable de la structure des tables au niveau de la couche base de données.from_attributes=TruePermettre à Pydantic de lire des données à partir d'objets SQLAlchemy.

4. Fonctions d'opérations de base de données

Danscrud.pyEncapsuler les opérations de base de données dans :

Exemple

# crud.py
from sqlalchemy.orm import Session
import models, schemas

def get_user(db: Session, user_id: int):
"""Récupérer l'utilisateur par ID"""
return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
"""Obtenir un utilisateur par e-mail"""
return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
"""Obtenir la liste des utilisateurs"""
return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
"""Créer un utilisateur"""
fake_hashed_password = user.password + "notreallyhashed" # En pratique, utilisez passlib pour le hachage
db_user = models.User(
email=user.email,
hashed_password=fake_hashed_password
)
db.add(db_user)
db.commit() # Valider la transaction
db.refresh(db_user) # Actualiser l'objet pour obtenir l'id généré par la base de données
return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
"""Obtenir la liste des produits"""
return db.query(models.Item).offset(skip).limit(limit).all()

def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
"""Créer un produit"""
db_item = models.Item(**item.model_dump(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item

5. Point d'entrée de l'application FastAPI

Dansmain.pyIntégrez tous les composants dans :

Exemple

# main.py
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from pydantic import ConfigDict

import models, schemas, crud
from database import engine, SessionLocal, Base

# Créer la table de base de données
Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dépendance : obtenir une session de base de données
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

# ===== Routes utilisateur =====
@app.post("/users/", response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
# Vérifier si l'adresse e-mail existe déjà
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail=“E-mail déjà enregistré”)
return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=list[schemas.UserOut])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users

@app.get("/users/{user_id}", response_model=schemas.UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail=“L’utilisateur n’existe pas”)
return db_user

# ===== Routes Produits =====
@app.post("/users/{user_id}/items/", response_model=schemas.ItemOut)
def create_item_for_user(
user_id: int,
item: schemas.ItemCreate,
db: Session = Depends(get_db),
):
return crud.create_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=list[schemas.ItemOut])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items

Injection de dépendances pour les sessions de base de données

Le modèle central consiste à utiliser l'injection de dépendances pour gérer les sessions de base de données :

def get_db():
    db = SessionLocal()   # 创建会话
    try:
        yield db          # 提供会话给路由函数
    finally:
        db.close()        # 请求完成后关闭会话

Chaque requête obtient une session de base de données indépendante, qui est automatiquement fermée après la fin de la requête, évitant ainsi les fuites de ressources.

Chaînes de connexion pour différentes bases de données

base de données	Exemple de chaîne de connexion
SQLite	`sqlite:///./sql_app.db`
PostgreSQL	`postgresql://user:password@localhost/dbname`
MySQL	`mysql+pymysql://user:password@localhost/dbname`

Le changement de base de données ne nécessite que la modification de la chaîne de connexion ; le reste du code n'a pas besoin d'être modifié. C'est l'un des avantages de l'utilisation d'un ORM.

Résumé

Utiliser SQLAlchemy comme ORM, et FastAPI gère les sessions de base de données par injection de dépendances.
Modèle Pydantic (schemas.py) gère la validation des données de l'API, les modèles SQLAlchemy (models.py) est responsable de la structure des tables de la base de données
Les opérations CRUD sont encapsulées danscrud.pydans, ce qui garde le code propre
Pour le développement, utilisez SQLite ; pour la production, passez à PostgreSQL/MySQL en ne changeant que la chaîne de connexion.
from_attributes=TruePermettre à Pydantic de lire les données à partir d'objets ORM.

autres extensions

Projet pratique

Intégration de base de données FastAPI

Installer les dépendances

Structure du projet

1. Configuration de la base de données

Exemple

2. Modèle SQLAlchemy

Exemple

3. Modèles de données Pydantic

Exemple

4. Fonctions d'opérations de base de données

Exemple

5. Point d'entrée de l'application FastAPI

Exemple

Injection de dépendances pour les sessions de base de données

Chaînes de connexion pour différentes bases de données

Résumé