FastAPI est un framework python pour concevoir des API robustes, performantes, documentées et faciles à maintenir.

La nouvelle génération des frameworks

Le framework FastAPI a été créé par Sebastian Ramirez, alias @tiangolo, en 2018 à l’issue d’un constat simple : il n’existe aucun framework python orienté API à la fois facile à prendre en main et performant.

Le framework nouvelle génération met en avant les points suivants :

  • Rapidité : des performances exceptionnelles qui le font passer devant Node.js ou encore Go
  • Développement : facilité d’apprentissage et d’adaptation
  • Qualité de code : réduction du code dupliqué et génération automatique de la documentation
  • Déploiement : prêt à aller en production sans effort
  • Standards : basé sur les standards des APIs : OpenAPI et JSON Schema
  • Documentation : le code suffit à lui-même pour automatiquement générer la documentation OpenAPI

La puissance du framework

FastAPI structure le code pour éviter les erreurs humaines et la duplication de code. Le framework guide le développeur et reste strict sur les entrants/sortants des étapes successives du code : de la base de données au rendu HTTP.

Structure de validation

Le framework fournit les outils pour structurer et valider les données qui transitent dans le code. Cette partie est réalisée par la librairie pydantic.

La force de pydantic est créer un cadre aux données qui sont manipulées dans le code. Cette structure comporte des champs qui doivent être typés, sur lesquels on peut ajouter des validateurs customs.

Les structures forcent les développeurs à se conformer au cadre définit sur la donnée. Cela empêche toute torsion du code et réduit considérablement les dérives opérées sur la donnée.

Les références : Documentation Pydantic, Github Pydantic, Pypi Pydantic.

Base de données

La base de données est représentée dans le code par une structure.

Elle contient des données (enregistrements ou documents) qui sont requêtées et rapatriées dans le code au travers d’un ORM (SQLAlchemy pour une base relationnelle).
L’ORM permet de simplifier et fluidifier la manipulation des données stockées en base en s’appuyant sur des objets appelés models fournit par pydantic.

Les données manipulées dans le code sont accessibles sous forme d’objets définis par des structures pydantic appelés schemas. Le schéma est étroitement lié au model, les champs doivent concorder en nombre et en forme pour faire transiter les données du code vers la base, et inversement.

Les références : Documentation FastAPI-SQLAlchemy, Pypi FastAPI-SQLAlchemy.

Tests unitaires

Les tests unitaires sont la clé de voute de la sérénité d’un projet.
FastAPI associe les avantages de Starlette, Requests et Pytest.

L’élaboration des tests se voit simplifiée et compréhensible. On peut dédier l’utilisation d’une base de données aux tests unitaires afin de ne pas polluer la base principale, ainsi que la création d’un jeu de données au lancement des tests.

Les références : Documentation FastAPI Testing // Documentation Starlette, Github Starlette, Pypi Starlette // Documentation Pytest, Github Pytest, Pypi Pytest.

Example as code

Quoi de mieux que des exemples pour bien comprendre ? Let’s go !

FastAPI – Les premiers pas

Objectif : Créer un Endpoint API avec la documentation OpenAPI associée.

Installer les dépendances fastapi et uvicorn :

$ pipenv install fastapi uvicorn[standard]

Créer le fichier fastapi_example01.py :

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def hello_world():
    return {"message": "Hello World"}

Lancer l’application sur le port par défaut (port 8000) :

$ pipenv run uvicorn fastapi_example01:app --reload

Résultats

Console
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [19340] using watchgod
INFO:     Started server process [19350]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
Endpoint API

Se rendre sur l’URL : http://127.0.0.1:8000/

{"message":"Hello World"}
API Documentation

Se rendre sur l’URL : http://127.0.0.1:8000/docs

FastAPI - Premiers Pas - API Documentation

Pydantic

Objectif : Structurer la donnée

Installer la dépendance pydantic :

$ pipenv install pydantic

Créer le fichier pydantic_example01.py :

from typing import List, Optional
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name = 'John Doe'
    friends: List[int] = []

user_johnny_dict = dict(
    id=1,
    name="Johnny",
)
user_johnny = User(**user_johnny_dict)
user_marc = User(id=2, name="Marc")
user_robert = User(id=3, name="Robert", friends=[1, 2])

print(dict(user_johnny))
print(dict(user_marc))
print(dict(user_robert))
print(user_robert.friends)

Lancer l’application :

$ pipenv run python pydantic_example01.py

Résultats

Console
{'id': 1, 'friends': [], 'name': 'Johnny'}
{'id': 2, 'friends': [], 'name': 'Marc'}
{'id': 3, 'friends': [1, 2], 'name': 'Robert'}
[1, 2]

FastAPI – Base de données SQL

Objectif : Manipuler de la donnée provenant d’une base

Arborescence de fichiers :

.
└── sql_app
    ├── __init__.py (empty file)
    ├── database.py
    ├── models.py
    ├── schemas.py
    ├── crud.py
    └── main.py

Installer les dépendances fastapi, uvicorn, fastapi-sqlalchemy :

$ pipenv install fastapi uvicorn fastapi-sqlalchemy

Contenu des fichiers :

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")
from typing import List, Optional

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


class ItemCreate(ItemBase):
    pass


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

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True
from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

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

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
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.User)
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="User not found")
    return db_user


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


@app.get("/items/", response_model=List[schemas.Item])
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

Résultats

Console
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [19568] using watchgod
INFO:     Started server process [19580]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
API Endpoint
Documentation

Se rendre sur l’URL : http://127.0.0.1:8000/docs

FastAPI - Base de données SQL - API Documentation

On comprend maintenant mieux l’avantage des schemas et des models, qui font le liant entre la base de données et la manipulation de cette donnée.

Les méthodes annotées @app.get, @app.post, etc. permettent d’autogénérer la documentation, ce qui fait gagner du temps et surtout empêche les erreurs humaines.

La connexion à la base de données provient de la librairie flask-sqlalchemy, ceux qui ont déjà manipulé flask l’auront remarqué.

Conclusion : le projet devient structuré (schemas, models, database, crud, main), ce qui augmente la lisibilité. La séparation des domaines permet de mieux se repérer et améliore la maintenabilité.

FastAPI – Tests unitaires

Objectif : Écrire les tests unitaires sur la base du projet précédent FastAPI – Base de données SQL.

Les fichiers créés s’ajoutent au projet précédent.

Arborescence de fichiers :

.
└── sql_app
    ├── ...
    └── tests/
        ├── __init__.py (empty file)
        ├── conftest.py
        ├── utils.py
        ├── test_crud_users.py
        └── test_api_users.py

Installer les dépendances fastapi, uvicorn, fastapi-sqlalchemy :

$ pipenv install fastapi uvicorn fastapi-sqlalchemy pytest requests

Contenu des fichiers :

from typing import  Generator

import pytest
from fastapi.testclient import TestClient

from sql_app.main import app
from sql_app.database import SessionLocal


@pytest.fixture(scope="session")
def db() -> Generator:
    yield SessionLocal()


@pytest.fixture(scope="module")
def client() -> Generator:
    with TestClient(app) as c:
        yield c
import random
import string

from fastapi.testclient import TestClient


def random_lower_string() -> str:
    return "".join(random.choices(string.ascii_lowercase, k=32))

def random_email() -> str:
    return f"{random_lower_string()}@{random_lower_string()}.com"
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session

from sql_app import crud
from sql_app.schemas import UserCreate
from sql_app.tests.utils import random_email, random_lower_string


def test_create_user(db: Session) -> None:
    email = random_email()
    password = random_lower_string()
    user_in = UserCreate(email=email, password=password)
    user = crud.create_user(db, user=user_in)
    assert user.email == email
    assert hasattr(user, "hashed_password")

def test_get_user(db: Session) -> None:
    password = random_lower_string()
    email = random_email()
    user_in = UserCreate(email=email, password=password, is_superuser=True)
    user = crud.create_user(db, user=user_in)
    user_2 = crud.get_user(db, user_id=user.id)
    assert user_2
    assert user.email == user_2.email
    assert jsonable_encoder(user) == jsonable_encoder(user_2)
from typing import Dict

from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from sql_app import crud
from sql_app.schemas import UserCreate
from sql_app.tests.utils import random_email, random_lower_string


def test_create_user_by_normal_user(
    client: TestClient
) -> None:
    email = random_email()
    password = random_lower_string()
    data = {"email": email, "password": password}
    r = client.post(
        f"/users/", json=data,
    )
    assert r.status_code == 200

Lancer les tests unitaires :

$ pipenv run pytest -sv sql_app/tests

Résultats

Console
====================================== test session starts ======================================
platform linux -- Python 3.8.0, pytest-6.2.2, py-1.10.0, pluggy-0.13.1 -- /home/germainlef/.local/share/virtualenvs/fastapi-beginning-1VT2EOFy/bin/python
cachedir: .pytest_cache
rootdir: /home/germainlef/ineat/blog/fastapi-beginning
collected 3 items

sql_app/tests/test_api_users.py::test_create_user_by_normal_user PASSED
sql_app/tests/test_crud_users.py::test_create_user PASSED
sql_app/tests/test_crud_users.py::test_get_user PASSED

====================================== 2 passed in 0.31s ======================================

On a ainsi testé le bon fonctionnement des méthodes CRUD de la classe User et un Endpoint API associé.

Conclusion

Nous venons de voir quelques fonctionnalités du framework en 3 exemples simples. On peut bien sûr aller beaucoup plus loin, je vous laisse parcourir la documentation de FastAPI, qui par ailleurs est d’une qualité exceptionnelle.

FastAPI, le framework nouvelle génération, est un sérieux challenger qui s’ajoute au marché des Frameworks Python. Ses fonctionnalités puissantes et sa facilité d’apprentissage en font un adversaire redoutable. Ses concurrents, principalement Django et Flask, vont avoir du fil à retordre face à cette fusée.

On peut remarquer quelques références à FastAPI : Microsoft, Uber, Netflix. En regardant au-delà des apparences marketing, on retrouve beaucoup d’équipes autour de l’IA et du ML qui utilisent le framework.
Connaissant leur rigueur et leur amour des données (bien calibrées), ça ne peut être qu’un gage de qualité.

Prochainement, nous verrons comment bien démarrer un projet d’envergure avec FastAPI. De l’arborescence de fichiers à la validation des entrants, en passant par les réponses API, nous aborderons des points beaucoup plus techniques.