Fastapi Backend Architecture

On continue la serie d’articles sur FastApi et mes découvertes. Dans celui-ci, je vais partager avec vous ma structure de projet et ma gestion des class de datas. En soi, c’est un mix des structures de projet que j’ai pu voir et de mes lectures. Je suis d'ailleurs en train de lire Domain-driven design.

Influence

Mes premières interactions avec l’architecture logiciel remontent à des projets C++ réalisés durant mon cursus à Epitech, en 3ème année. Puis y a eu la découverte des concepts de clean architecture et d’architecture hexagonale. Et plus récemment, j’ai développé, chez Mycoach, en Scala, de la programmation fonctionnelle avec tous les patterns qui viennent avec : immutabilité de la data, parallélisme ect..

Découpage du projet

Quand j’architecture mon projet je vais chercher plusieurs choses:

• Avoir une seule source de responsabilité : par exemple je vais centraliser tout mes appels à ma db pour une certaine collection ou table dans une seul class. (tout est ajustable en fonction du problème à résoudre)
• Small is buity: plus votre code sera petit plus il sera facile à tester et moins il risquera d’être couplé avec d’autres parties. Par exemple : pour les handlers des routers, les handlers ne contiennent que le code lié à fastApi. La mauvaise pratique serait de faire dans le handler un appel à la db et de rajouter de la business logic dedans.

Project architecture

Qu’est-ce qui doit être isolé?

• La partie routing pour une api
• Un domaine qui va centraliser les definitions des interfaces et des data class.
• Des repositories contenant les interactions avec la db.
• Des services contenant la logic métier
• Des utils, c’est un peu le “fourre-tout”, j y stocke des clients pour service externe ect..

La logique est toujours la même : avoir le moins de code possible pour avoir le moins de code à tester et faciliter sa réalisation.

Pour notre projet cela va ressembler à ça :

main.py
src
├── api
│   ├── __init__.py
│   ├── healthRouter
│   │   ├── __init__.py
│   │   └── handler.py
│   └── server.py
├── domain
│   ├── inputs
│   ├── interfaces
│   └── outputs
│       └── health_reponse.py
├── repository
└── service

Injection de dépendance

Pour bien vous permettre de comprendre comment fonctionne le projet, il me paraît important d’évoquer l’injection de dépendance. Pour faire simple, cela se concretise ainsi : Dans la class du router je vais avoir dans les paramètres du constructeur, les services dont jai besoin qui eux-mêmes auront dans leurs propres paramètres les repository et les utils dont ils ont besoin.

Il faut voir ça comme des poupées russes. Cest en partie ce qui rend possible la découpe de notre projet.

Attention pas tous les langages supportent l’injection de dépendance. Par exemple : le JS et le TS ne le supportent pas nativement. Il faut avoir recours à des solutions comme inversify ou des framework comme nest.js.

Data structure

Avant d’attaquer le code il me semble important de prendre d’évoquer rapidement les class.

Le python a un système de class vraiment basique, trop à mon goût. En cherchant, j’ai fini par découvrir la lib Pydentic ayant un bon couplage à fastapi et les dataclass(freezed).

L’intégration de Pydentic par fastapi permet entre autre d’avoir un check des input et la génération d’un swagger avec la définition des contrats.

Voici un exemple de class BaseModel de pydentic. Cette class, me permet de définir l’output renvoyé dans mon endpoint de healthcheck.

from pydantic import BaseModel

class HealthResponse(BaseModel):
    status: str = 'ok'

L’utilisation des dataclass Freezed permet qu’une class devienne immuable après son initialisation un peu comme un const en JS.

Du coup, j’utilise les deux les pydentic BaseModel pour faire toute la data validation et la génération de la doc swagger côté API et les dataclass Freezed partout où c’est possible.

Nous allons mettre ça en pratique.

Code

Nous allons à présent realiser le Create et le Get de livre et le stocker en DB.

Nous allons realiser de l’injection de dépendance. Nous allons commencer par déveloper le repository, puis le service et pour finir le router.

Repository

Data schéma

Pour commencer, nous allons définir ce que nous voulons stocker. Pour l’exemple ce sera sera un livre. Pour cela on va créer un fichier book dans src/domain/mongo/book.py.

from dataclasses import dataclass

from bson import ObjectId

@dataclass(frozen=True)
class Book:
    _id: ObjectId
    title: str
    author: str
    year: int

En choisissant d’en faire une class frozen, donc une class immuable, cela va permettre de structurer le code produit et de limiter le risque d’erreur avec les instances d’un object.

Protocol

Les Protocol sont les types pour définir les interfaces en python. Ils permettent la rédaction d’un contrat pour les class. Dans certains langages comme en GO l’utilisation des interfaces est très fortement conseillée pour se faciliter la vie lors des tests unitaires.

Revenons à notre python, la définition de notre interface se limitera à un insert et find_by_id. Notre protocol sera créé à cet endroit dans le repository : src/domain/interfaces/I_book_service.py.

from typing import Optional, Protocol

from src.domain.mongo.book import Book

class IBookService(Protocol):
    def insert(self, document: Book) -> Optional[Book]:
        ...

    def find_by_id(self, document_id: str) -> Optional[Book]:
        ...

Repository

Nous allons créer notre src/repository/books_repository.py. Le but de ce fichier est de centraliser tout les calls faits à la db au sujet des livres.

from typing import Optional

from bson import ObjectId
from pymongo import MongoClient

from src.domain.interfaces.i_book_repository import IBookRepository
from src.domain.mongo.book import Book

class BooksRepository(IBookRepository):
    def __init__(self, db_url: str, db_name: str, collection_name: str):
        self.client = MongoClient(db_url)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    def insert(self, document: Book) -> Optional[Book]:
        result = self.collection.insert_one(vars(document))
        if result.inserted_id:
            return document
        return None

    def find_by_id(self, document_id: ObjectId) -> Optional[Book]:
        print(document_id)
        document_data = self.collection.find_one({"_id": ObjectId(document_id)})
        if document_data:
            return Book(**document_data)
        return None

Pour l’exemple, j’ai juste codé un insert et un find_by_id.

Service

Pour éviter toute redondance, je ne vais pas développer la creation du protocole, c’est en effet exactement le même workflow : création du protocol → création du service → création du fichier de test.

from typing import Optional

from bson import ObjectId

from src.domain.interfaces.i_book_repository import IBookRepository
from src.domain.interfaces.i_book_service import IBooksService
from src.domain.mongo.book import Book

class BookService(IBooksService):
    def __init__(self, repository: IBookRepository):
        self.repository = repository

    def insert_book(self, title: str, author: str, year: int) -> Optional[Book]:
        new_object_id = ObjectId()
        book = Book(_id=new_object_id, title=title, author=author, year=year)
        return self.repository.insert(book)

    def find_book_by_id(self, document_id: str) -> Optional[Book]:
        return self.repository.find_by_id(document_id)

La partie intéressante dans cet exemple est dans la définition du constructeur. On vient injecter le repository, ce qui va se traduire par une initialisation du repository au niveau du main.py ou du router. Cela va différer en fonction des langages et des implémantations des Framework.

Dans le service, vous y centraliserez la partie business logic. L’application de cette architecture vous permettra au besoin de changer facilement soit de db, soit de framework http.

Api

Pour se qui est de l’Api, j’aime découper ça par router. Le router centralisera tout les endpoints pour un prefix.

On va créer notre src/api/booksRouter/handler.py. FastApi met en avant un système de définitions de décorateurs pour définir les endpoints et la documentation swagger. Du coup je me suis adapté pour faire au mieux. L’injection de dépendance que je fais n’est pas optimale mais elle fera l’affaire.

books_router = APIRouter(prefix="/books", tags=["Books"])

def get_books_service() -> IBooksService:
    repository: IBookRepository = BooksRepository("mongodb://localhost:27017/", "mydatabase", "mycollection")
    return BookService(repository)

@books_router.post("", response_model=OutputBook)
async def insert_book(book: InputBook,
                      service: IBooksService = Depends()):
    res = service.insert_book(book.title, book.author, book.year)
    if res:
        return OutputBook(id=str(res._id), title=res.title, author=res.author, year=res.year)
    else:
        raise HTTPException(status_code=500, detail="book insert fail")

@books_router.get("/{book_id}")
async def get_book_by_id(book_id: ObjectIdStr,
                         service: IBooksService = Depends(get_books_service)):
    book = service.find_book_by_id(str(book_id))
    if book:
        return OutputBook(id=str(book._id), title=book.title, author=book.author, year=book.year)
    else:
        raise HTTPException(status_code=404, detail="book not found")

Grace au décorateur @books_router on va pouvoir définir les endpoints, FastApi provide propose plein de paramètres optionnels pour la documentation. Il nous suffira d’importer le books_router et de le lier à notre app.

On va créer un src/api/server.py qui nous servira à centraliser tout nos routers.

from fastapi import FastAPI
from fastapi.applications import AppType

from src.api import healthRouter
from src.api.booksRouter.handler import books_router

def initServer(app: FastAPI) -> AppType:
    app.include_router(healthRouter.router)
    app.include_router(books_router)
    return app

Et pour finir, on appelle initServer dans le main.py.

from fastapi import FastAPI

pfrom src.api.server import initServer

app = FastAPI(title="Mon API FastAPI",
              description="Ceci est une API de démonstration",
              version="1.0.0")

initServer(app)

Et voila c’est fini un aperçu du repository (il manque les tests unitaires). J’espère que cela pourra en aider certains ou alimenter votre réflexion sur vos architectures backend.

A bientôt!