Modèles et pratiques d'utilisation de SQLAlchemy 2.0 avec FastAPI

Alors que Django et Flask restent les premiers choix pour de nombreux ingénieurs Python, FastAPI a déjà été reconnu comme un choix indéniablement fiable. Il s'agit d'un framework hautement flexible, bien optimisé et structuré qui offre au développeur des possibilités infinies pour créer des applications backend.

Travailler avec des bases de données est un aspect essentiel de la plupart des applications backend. Par conséquent, l'ORM joue un rôle essentiel dans le code backend. Cependant, contrairement à Django, FastAPI n'a pas d'ORM intégré. Il est entièrement de la responsabilité du développeur de sélectionner une bibliothèque appropriée et de l'intégrer dans la base de code.

Les ingénieurs Python considèrent largement SQLAlchemy comme l'ORM le plus populaire disponible. C'est une bibliothèque légendaire qui est utilisée depuis 2006 et qui a été adoptée par des milliers de projets. En 2023, il a reçu une mise à jour majeure vers la version 2.0. Semblable à FastAPI, SQLAlchemy fournit aux développeurs des fonctionnalités et des utilitaires puissants sans les forcer à les utiliser d'une manière spécifique. Il s'agit essentiellement d'une boîte à outils polyvalente qui permet aux développeurs de l'utiliser comme bon leur semble.

FastAPI et SQLAlchemy sont un match fait au paradis. Ce sont des technologies à la fois fiables, performantes et modernes, qui permettent de créer des applications puissantes et uniques. Cet article explore la création d'une application backend FastAPI qui utilise SQLAlchemy 2.0 comme ORM. Le contenu couvre :

• construction de modèles à l'aide de Mapped et mapped_column
• définir un modèle abstrait
• gérer une session de base de données
• en utilisant l'ORM
• créer une classe de référentiel commune pour tous les modèles
• préparer une configuration de test et ajouter des tests

Ensuite, vous pourrez facilement combiner l'application FastAPI avec SQLAlchemy ORM. De plus, vous vous familiariserez avec les meilleures pratiques et modèles pour créer des applications bien structurées, robustes et performantes.

Conditions préalables

Les exemples de code inclus dans l'article proviennent du projet alchemist , qui est une API de base pour créer et lire des objets d'ingrédients et de potions. L'objectif principal de l'article est d'explorer la combinaison de FastAPI et SQLAlchemy. Il ne couvre pas d'autres sujets, tels que :

• configuration de la configuration de Docker
• démarrage du serveur uvicorn
• mise en place de peluches

Si ces sujets vous intéressent, vous pouvez les explorer par vous-même en examinant la base de code. Pour accéder au dépôt de code du projet alchemist, veuillez suivre ce lien ici . De plus, vous pouvez trouver la structure de fichiers du projet ci-dessous :

 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml

Bien que l'arborescence puisse sembler grande, certains contenus ne sont pas pertinents pour le point principal de cet article. De plus, le code peut sembler plus simple que ce qui est nécessaire dans certains domaines. Par exemple, le projet manque :

• étape de production dans le Dockerfile
• configuration de l'alambic pour les migrations
• sous-répertoires pour les tests

Cela a été fait intentionnellement pour réduire la complexité et éviter les frais généraux inutiles. Cependant, il est important de garder ces facteurs à l'esprit s'il s'agit d'un projet plus prêt pour la production.

Exigences de l'API

Lorsque vous commencez à développer une application, il est essentiel de prendre en compte les modèles que votre application utilisera. Ces modèles représenteront les objets et les entités avec lesquels votre application fonctionnera et seront exposés dans l'API. Dans le cas de l'application alchimiste, il y a deux entités : les ingrédients et les potions. L'API doit permettre de créer et de récupérer ces entités. Le fichier alchemist/api/models.py contient les modèles qui seront utilisés dans l'API :

 import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1)

L'API renverra des modèles Ingredient et Potion . Définir orm_mode sur True dans la configuration facilitera le travail avec les objets SQLAlchemy à l'avenir. Des modèles Payload seront utilisés pour créer de nouveaux objets.

L'utilisation de pydantic rend les classes plus détaillées et plus claires dans leurs rôles et fonctions. Il est maintenant temps de créer les modèles de base de données.

Déclarer des modèles

Un modèle est essentiellement une représentation de quelque chose. Dans le contexte des API, les modèles représentent ce que le backend attend dans le corps de la requête et ce qu'il renverra dans les données de réponse. Les modèles de base de données, en revanche, sont plus complexes et représentent les structures de données stockées dans la base de données et les types de relations entre elles.

Le fichier alchemist/database/models.py contient des modèles pour les objets ingrédient et potion :

 import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", )

Chaque modèle dans SQLAlchemy commence par la classe DeclarativeBase . En hériter permet de construire des modèles de base de données compatibles avec les vérificateurs de type Python.

Il est également recommandé de créer un modèle abstrait (classe Base dans ce cas) qui inclut les champs requis dans tous les modèles. Ces champs incluent la clé primaire, qui est un identifiant unique de chaque objet. Le modèle abstrait stocke également souvent les dates de création et de mise à jour d'un objet, qui sont définies automatiquement, lorsqu'un objet est créé ou mis à jour. Cependant, le modèle Base restera simple.

Passant au modèle Ingredient , l'attribut __tablename__ spécifie le nom de la table de base de données, tandis que le champ name utilise la nouvelle syntaxe SQLAlchemy, permettant aux champs de modèle d'être déclarés avec des annotations de type. Cette approche concise et moderne est à la fois puissante et avantageuse pour les vérificateurs de type et les IDE, car elle reconnaît le champ name comme une chaîne.

Les choses deviennent plus complexes dans le modèle Potion . Il inclut également les attributs __tablename__ et name , mais en plus de cela, il stocke la relation avec les ingrédients. L'utilisation de Mapped[list["Ingredient"]] indique que la potion peut contenir plusieurs ingrédients, et dans ce cas, la relation est plusieurs à plusieurs (M2M). Cela signifie qu'un seul ingrédient peut être affecté à plusieurs potions.

Le M2M nécessite une configuration supplémentaire, impliquant généralement la création d'une table d'association qui stocke les connexions entre les deux entités. Dans ce cas, l'objet potion_ingredient_association stocke uniquement les identifiants de l'ingrédient et de la potion, mais il peut également inclure des attributs supplémentaires, tels que la quantité d'un ingrédient spécifique nécessaire pour la potion.

La fonction relationship configure la relation entre la potion et ses ingrédients. L'argument lazy spécifie comment les éléments associés doivent être chargés. En d'autres termes : que doit faire SQLAlchemy avec les ingrédients associés lorsque vous récupérez une potion ? Le régler sur selectin signifie que les ingrédients seront chargés avec la potion, éliminant ainsi le besoin de requêtes supplémentaires dans le code.

Construire des modèles bien conçus est crucial lorsque vous travaillez avec un ORM. Une fois cela fait, l'étape suivante consiste à établir la connexion avec la base de données.

Gestionnaire de sessions

Lorsque vous travaillez avec une base de données, en particulier lors de l'utilisation de SQLAlchemy, il est essentiel de comprendre les concepts suivants :

• dialecte
• moteur
• connexion
• pool de connexion
• session

De tous ces termes, le plus important est le moteur . Selon la documentation SQLAlchemy, l'objet moteur est chargé de connecter le Pool et Dialect pour faciliter la connectivité et le comportement de la base de données. En termes plus simples, l'objet moteur est la source de la connexion à la base de données, tandis que la connexion fournit des fonctionnalités de haut niveau telles que l'exécution d'instructions SQL, la gestion des transactions et la récupération des résultats de la base de données.

Une session est une unité de travail qui regroupe des opérations connexes au sein d'une même transaction. Il s'agit d'une abstraction des connexions de base de données sous-jacentes et gère efficacement les connexions et le comportement transactionnel.

Dialect est un composant qui prend en charge un backend de base de données spécifique. Il agit comme un intermédiaire entre SQLAlchemy et la base de données, gérant les détails de la communication. Le projet alchemist utilise Postgres comme base de données, donc le dialecte doit être compatible avec ce type de base de données spécifique.

Le dernier point d'interrogation est le pool de connexion . Dans le contexte de SQLAlchemy, un pool de connexions est un mécanisme qui gère une collection de connexions de base de données. Il est conçu pour améliorer les performances et l'efficacité des opérations de base de données en réutilisant les connexions existantes plutôt qu'en en créant de nouvelles pour chaque requête. En réutilisant les connexions, le pool de connexions réduit la surcharge liée à l'établissement de nouvelles connexions et à leur suppression, ce qui améliore les performances.

Une fois ces connaissances couvertes, vous pouvez maintenant jeter un œil au fichier alchemist/database/session.py , qui contient une fonction qui sera utilisée comme dépendance pour se connecter à la base de données :

 from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise

Le premier détail important à noter est que la fonction get_db_session est une fonction génératrice. En effet, le système de dépendance FastAPI prend en charge les générateurs. Par conséquent, cette fonction peut gérer à la fois les scénarios de réussite et d'échec.

Les deux premières lignes de la fonction get_db_session créent un moteur de base de données et une session. Cependant, l'objet de session peut également être utilisé comme gestionnaire de contexte. Cela vous donne plus de contrôle sur les exceptions potentielles et les résultats positifs.

Bien que SQLAlchemy gère la fermeture des connexions, il est recommandé de déclarer explicitement comment gérer la connexion une fois qu'elle est terminée. Dans la fonction get_db_session , la session est validée si tout se passe bien et annulée si une exception est levée.

Il est important de noter que ce code est construit autour de l'extension asyncio. Cette fonctionnalité de SQLAlchemy permet à l'application d'interagir avec la base de données de manière asynchrone. Cela signifie que les demandes adressées à la base de données ne bloqueront pas les autres demandes d'API, ce qui rendra l'application beaucoup plus efficace.

Une fois les modèles et la connexion configurés, l'étape suivante consiste à s'assurer que les modèles sont ajoutés à la base de données.

Migrations rapides

Les modèles SQLAlchemy représentent les structures d'une base de données. Cependant, leur simple création n'entraîne pas de modifications immédiates de la base de données. Pour apporter des modifications, vous devez d'abord les appliquer . Cela se fait généralement à l'aide d'une bibliothèque de migration telle qu'alembic, qui suit chaque modèle et met à jour la base de données en conséquence.

Étant donné qu'aucune autre modification des modèles n'est prévue dans ce scénario, un script de migration de base suffira. Vous trouverez ci-dessous un exemple de code du fichier scripts/migrate.py .

 import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables())

Pour faire simple, la fonction migrate_tables lit la structure des modèles et la recrée dans la base de données à l'aide du moteur SQLAlchemy. Pour exécuter ce script, utilisez la commande python scripts/migrate.py .

Les modèles sont maintenant présents à la fois dans le code et dans la base de données et get_db_session peut faciliter les interactions avec la base de données. Vous pouvez maintenant commencer à travailler sur la logique de l'API.

API avec l'ORM

Comme mentionné précédemment, l'API pour les ingrédients et les potions est censée prendre en charge trois opérations :

• création d'objets
• lister des objets
• récupération d'objets par ID

Grâce aux préparations préalables, toutes ces fonctionnalités peuvent déjà être implémentées avec SQLAlchemy comme ORM et FastAPI comme framework web. Pour commencer, passez en revue l'API des ingrédients située dans le fichier alchemist/api/v1/routes.py :

 import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient)

Sous l'API /ingredients , trois routes sont disponibles. Le point de terminaison POST prend une charge utile d'ingrédient en tant qu'objet à partir d'un modèle créé précédemment et d'une session de base de données. La fonction de générateur get_db_session initialise la session et active les interactions de base de données.

Dans le corps de la fonction proprement dite, cinq étapes se déroulent :

1. Un objet ingrédient est créé à partir de la charge utile entrante.
2. La méthode add de l'objet session ajoute l'objet ingrédient au système de suivi de session et le marque comme en attente d'insertion dans la base de données.
3. La séance est validée.
4. L'objet ingrédient est actualisé pour s'assurer que ses attributs correspondent à l'état de la base de données.
5. L'instance d'ingrédient de base de données est convertie en instance de modèle d'API à l'aide de la méthode from_orm .

Pour un test rapide, une boucle simple peut être exécutée sur l'application en cours d'exécution :

 curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}'

Dans la réponse, il doit y avoir un objet ingrédient dont l'ID provient de la base de données :

 { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" }

Bien que les multiples couches d'abstraction de SQLAlchemy puissent sembler inutiles pour une API simple, elles maintiennent les détails ORM séparés et contribuent à l'efficacité et à l'évolutivité de SQLAlchemy. Lorsqu'elles sont combinées avec asyncio, les fonctionnalités ORM fonctionnent exceptionnellement bien dans l'API.

Les deux paramètres restants sont moins complexes et partagent des similitudes. Une partie qui mérite une explication plus approfondie est l'utilisation de la méthode scalars dans la fonction get_ingredients . Lors de l'interrogation de la base de données à l'aide de SQLAlchemy, la méthode execute est souvent utilisée avec une requête comme argument. Alors que la méthode execute renvoie des tuples de type ligne, scalars renvoient directement les entités ORM, ce qui rend le point de terminaison plus propre.

Considérons maintenant l'API des potions, dans le même fichier :

 @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion)

Les points de terminaison GET pour les potions sont identiques à ceux des ingrédients. Cependant, la fonction POST nécessite du code supplémentaire. En effet, la création de potions implique d'inclure au moins un identifiant d'ingrédient, ce qui signifie que les ingrédients doivent être récupérés et liés à la potion nouvellement créée. Pour ce faire, la méthode scalars est à nouveau utilisée, mais cette fois avec une requête qui spécifie les ID des ingrédients récupérés. La partie restante du processus de création de la potion est identique à celle des ingrédients.

Pour tester le point de terminaison, une commande curl peut à nouveau être exécutée.

 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

Il en résulte la réponse suivante :

 { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] }

Il est important de noter que chaque ingrédient est représenté comme un objet complet dans la potion, grâce à l'argument lazy="selectin" spécifié dans la relation.

Les API sont fonctionnelles, mais il y a un problème majeur avec le code. Bien que SQLAlchemy vous donne la liberté d'interagir avec la base de données à votre guise, il n'offre aucun utilitaire de "gestionnaire" de haut niveau similaire à Model.objects de Django. Par conséquent, vous devrez le créer vous-même, ce qui est essentiellement la logique utilisée dans les API d'ingrédients et de potions. Cependant, si vous conservez cette logique directement dans les points de terminaison sans l'extraire dans un espace séparé, vous vous retrouverez avec beaucoup de code dupliqué. De plus, apporter des modifications aux requêtes ou aux modèles deviendra de plus en plus difficile à gérer.

Le prochain chapitre présente le modèle de référentiel : une solution élégante pour extraire le code ORM.

Dépôt

Le modèle de référentiel permet d'abstraire les détails de l'utilisation de la base de données. Dans le cas de l'utilisation de SQLAlchemy, comme dans l'exemple de l'alchimiste, la classe de référentiel serait responsable de la gestion de plusieurs modèles et de l'interaction avec la session de base de données.

Jetez un œil au code suivant du fichier alchemist/database/repository.py :

 import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query))

La classe DatabaseRepository contient toute la logique précédemment incluse dans les points de terminaison. La différence est qu'il permet de transmettre la classe de modèle spécifique dans la méthode __init__ , ce qui permet de réutiliser le code pour tous les modèles au lieu de le dupliquer dans chaque point de terminaison.

De plus, DatabaseRepository utilise des génériques Python, avec le type générique Model lié au modèle abstrait de base de données. Cela permet à la classe de référentiel de bénéficier davantage de la vérification de type statique. Lorsqu'ils sont utilisés avec un modèle spécifique, les types de retour des méthodes du référentiel reflètent ce modèle spécifique.

Étant donné que le référentiel doit utiliser la session de base de données, il doit être initialisé avec la dépendance get_db_session . Considérez la nouvelle dépendance dans le fichier alchemist/api/v2/dependencies.py :

 from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository(  model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]:  def func(session: AsyncSession = Depends(session.get_db_session)):    return repository.DatabaseRepository(model, session)  return func

En termes simples, la fonction get_repository est une fabrique de dépendances. Il prend d'abord le modèle de base de données avec lequel vous utiliserez le référentiel. Ensuite, il renvoie la dépendance, qui sera utilisée pour recevoir la session de base de données et initialiser l'objet de référentiel. Pour mieux comprendre, consultez la nouvelle API à partir du fichier alchemist/api/v2/routes.py . Il ne montre que les points de terminaison POST, mais cela devrait suffire à vous donner une idée plus claire de la façon dont le code est amélioré :

 from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[  DatabaseRepository[db_models.Ingredient],  Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[  DatabaseRepository[db_models.Potion],  Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion)

La première fonctionnalité importante à noter est l'utilisation de Annotated , une nouvelle façon de travailler avec les dépendances FastAPI. En spécifiant le type de retour de la dépendance en tant que DatabaseRepository[db_models.Ingredient] et en déclarant son utilisation avec Depends(get_repository(db_models.Ingredient)) vous pouvez finir par utiliser des annotations de type simples dans le point de terminaison : repository: IngredientRepository .

Grâce au référentiel, les terminaux n'ont pas à stocker toute la charge liée à l'ORM. Même dans le cas des potions plus compliquées, tout ce que vous avez à faire est d'utiliser deux référentiels en même temps.

Vous pouvez vous demander si l'initialisation de deux référentiels initialisera la session deux fois. La réponse est non. Le système de dépendance FastAPI met en cache les mêmes appels de dépendance dans une seule requête. Cela signifie que l'initialisation de la session est mise en cache et que les deux référentiels utilisent exactement le même objet de session. Encore une autre grande fonctionnalité de la combinaison de SQLAlchemy et FastAPI.

L'API est entièrement fonctionnelle et dispose d'une couche d'accès aux données réutilisable et hautement performante. L'étape suivante consiste à s'assurer que les exigences sont satisfaites en écrivant des tests de bout en bout.

Essai

Les tests jouent un rôle crucial dans le développement de logiciels. Les projets peuvent contenir des tests unitaires, d'intégration et de bout en bout (E2E). Bien qu'il soit généralement préférable d'avoir un nombre élevé de tests unitaires significatifs, il est également bon d'écrire au moins quelques tests E2E pour s'assurer que l'ensemble du flux de travail fonctionne correctement.

Pour créer des tests E2E pour l'application alchemist, deux bibliothèques supplémentaires sont nécessaires :

• pytest pour réellement créer et exécuter les tests
• httpx pour faire des requêtes asynchrones dans les tests

Une fois ceux-ci installés, l'étape suivante consiste à mettre en place une base de données de test distincte. Vous ne voulez pas que votre base de données par défaut soit polluée ou abandonnée. Comme alchemist inclut une configuration Docker, seul un simple script est nécessaire pour créer une deuxième base de données. Jetez un œil au code du fichier scripts/create_test_db.sh :

 #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test"

Pour que le script soit exécuté, il doit être ajouté en tant que volume au conteneur Postgres. Ceci peut être réalisé en l'incluant dans la section volumes du fichier docker-compose.yaml .

La dernière étape de la préparation consiste à créer des appareils pytest dans le fichier tests/conftest.py :

 from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client

Une chose qu'il est essentiel de changer dans les tests, c'est la façon dont l'application interagit avec la base de données. Cela inclut non seulement la modification de l'URL de la base de données, mais également la garantie que chaque test est isolé en commençant par une base de données vide.

Le luminaire db_session remplit ces deux objectifs. Son corps suit les étapes suivantes :

1. Créez un moteur avec une URL de base de données modifiée.
2. Supprimez toutes les tables existantes pour vous assurer que le test dispose d'une base de données propre.
3. Créez toutes les tables à l'intérieur de la base de données (même code que dans le script de migration).
4. Créez et produisez un objet de session.
5. Fermez manuellement la session lorsque le test est terminé.

Bien que la dernière étape puisse également être implémentée en tant que gestionnaire de contexte, la fermeture manuelle fonctionne très bien dans ce cas.

Les deux appareils restants devraient être assez explicites :

• test_app est l'instance FastAPI du fichier alchemist/app.py , avec la dépendance get_db_session remplacée par le luminaire db_session
• client est le httpx AsyncClient qui fera des requêtes API contre le test_app

Avec tout cela mis en place, les tests réels peuvent enfin être écrits. Par souci de concision, l'exemple ci-dessous du fichier tests/test_api.py montre uniquement un test de création d'un ingrédient :

 from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk

Le test utilise un objet client créé dans un appareil, qui envoie des requêtes à l'instance FastAPI avec une dépendance remplacée. En conséquence, le test est capable d'interagir avec une base de données distincte qui sera effacée une fois le test terminé. La structure de la suite de tests restante pour les deux API est à peu près la même.

Résumé

FastAPI et SQLAlchemy sont d'excellentes technologies pour créer des applications backend modernes et puissantes. La liberté, la simplicité et la flexibilité qu'ils offrent en font l'une des meilleures options pour les projets basés sur Python. Si les développeurs suivent les meilleures pratiques et modèles, ils peuvent créer des applications performantes, robustes et bien structurées qui gèrent facilement les opérations de base de données et la logique d'API. Cet article visait à vous fournir une bonne compréhension de la façon de mettre en place et de maintenir cette incroyable combinaison.

Sources

Le code source du projet alchemist peut être trouvé ici : lien .