Dans le développement d'une API RESTful, les opérations CRUD (Créer, Lire, Mettre à jour, Supprimer) constituent le cœur de toute application de gestion des données. En utilisant FastAPI, nous pouvons facilement mettre en place ces opérations et les exposer via des endpoints HTTP. Dans ce chapitre, nous détaillerons le processus de création des fonctions nécessaires pour gérer une base de données de tâches et l'exposition de ces opérations via une API.

Créer des opérations CRUD de base

La première étape dans la gestion des tâches consiste à définir des fonctions permettant de récupérer et manipuler les données. Nous commencerons par la fonction permettant de récupérer le prochain identifiant unique (ID) pour une nouvelle tâche, basé sur les valeurs déjà présentes dans la base de données. Cette fonction inspecte un fichier CSV utilisé comme base de données, cherche le plus grand identifiant et ajoute 1 pour déterminer le nouvel ID :

python
def get_next_id():
try: with open(DATABASE_FILENAME, "r") as csvfile: reader = csv.DictReader(csvfile) max_id = max(
int(row["id"]) for row in reader
)
return max_id + 1 except (FileNotFoundError, ValueError): return 1

Ensuite, il faut définir la fonction qui écrit les données d’une nouvelle tâche dans le fichier CSV. Cette fonction prend un objet TaskWithID, extrait ses données, et les écrit dans le fichier :

python
def write_task_into_csv(task: TaskWithID):
with open(DATABASE_FILENAME, mode="a", newline="") as file:
writer = csv.DictWriter(file, fieldnames=column_fields) writer.writerow(task.model_dump())

À partir de ces deux fonctions de base, nous pouvons créer une fonction qui permet de créer une tâche en générant un ID unique et en écrivant la tâche dans le fichier CSV :

python
def create_task(task: Task) -> TaskWithID: id = get_next_id() task_with_id = TaskWithID(id=id, **task.model_dump()) write_task_into_csv(task_with_id) return task_with_id

De même, une autre fonction permet de modifier une tâche en fonction de son ID, en mettant à jour ses champs dans la base de données. Elle lit d'abord toutes les tâches existantes, met à jour celle qui correspond à l'ID donné, puis réécrit l'ensemble des tâches dans le fichier CSV :

python
def modify_task(id: int, task: dict) -> Optional[TaskWithID]: updated_task: Optional[TaskWithID] = None tasks = read_all_tasks() for number, task_ in enumerate(tasks): if task_.id == id: tasks[number] = updated_task = task_.model_copy(update=task)
with open(DATABASE_FILENAME, mode="w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=column_fields) writer.writeheader()
for task in tasks: writer.writerow(task.model_dump()) if updated_task: return updated_task

Pour supprimer une tâche spécifique, nous devons d'abord rechercher la tâche correspondante à son ID, la supprimer, puis réécrire les tâches restantes dans le fichier CSV :

python
def remove_task(id: int) -> bool: deleted_task: Optional[Task] = None tasks = read_all_tasks()
with open(DATABASE_FILENAME, mode="w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=column_fields) writer.writeheader()
for task in tasks: if task.id == id: deleted_task = task continue writer.writerow(task.model_dump()) if deleted_task: dict_task_without_id = deleted_task.model_dump() del dict_task_without_id["id"] return Task(**dict_task_without_id)

Exposer les opérations via FastAPI

Une fois les fonctions CRUD mises en place, il est temps de les exposer à travers des endpoints HTTP dans FastAPI. Nous allons créer une API permettant de lister, récupérer, ajouter, modifier et supprimer des tâches.

  1. Lister toutes les tâches :
    Un endpoint GET /tasks permet de récupérer toutes les tâches stockées dans la base de données.

    python
    @app.get("/tasks", response_model=list[TaskWithID]) def get_tasks(): tasks = read_all_tasks() return tasks
  2. Récupérer une tâche spécifique :
    Un endpoint GET /tasks/{task_id} permet de récupérer les détails d’une tâche à partir de son ID. Si la tâche n'existe pas, une exception est levée.

    python
    @app.get("/task/{task_id}")
    def get_task(task_id: int): task = read_task(task_id) if not task: raise HTTPException(status_code=404, detail="task not found") return task
  3. Ajouter une nouvelle tâche :
    L'endpoint POST /task permet de créer une nouvelle tâche en envoyant les données nécessaires dans le corps de la requête.

    python
    @app.post("/task", response_model=TaskWithID) def add_task(task: Task): return create_task(task)
  4. Mettre à jour une tâche existante :
    Un endpoint PUT /tasks/{task_id} permet de modifier les informations d'une tâche existante. La mise à jour se fait en envoyant les nouveaux champs à modifier.

    python
    class UpdateTask(BaseModel):
    title: str | None = None description: str | None = None status: str | None = None @app.put("/task/{task_id}", response_model=TaskWithID) def update_task(task_id: int, task_update: UpdateTask): modified = modify_task(task_id, task_update.model_dump(exclude_unset=True)) if not modified: raise HTTPException(status_code=404, detail="task not found") return modified
  5. Supprimer une tâche :
    Enfin, l'endpoint DELETE /tasks/{task_id} permet de supprimer une tâche spécifique en utilisant son ID.

    python
    @app.delete("/task/{task_id}", response_model=Task) def delete_task(task_id: int): removed_task = remove_task(task_id) if not removed_task: raise HTTPException(status_code=404, detail="task not found") return removed_task

Compléments essentiels pour la bonne gestion des tâches

Lors de l’implémentation de ces opérations, plusieurs aspects sont cruciaux pour assurer une gestion efficace des données :

  1. Validation des données : Bien que FastAPI assure une validation automatique des types de données avec les annotations, il est essentiel de vérifier que les données envoyées via les requêtes POST ou PUT respectent les contraintes de l'application, comme la longueur du titre ou la validité du statut.

  2. Gestion des erreurs : Lors de l’interaction avec les fichiers (comme l'ouverture du fichier CSV), il est important de prendre en compte les exceptions potentielles, telles que les erreurs d'I/O ou les erreurs liées à des données mal formées. Un système de gestion d’erreurs robuste peut améliorer l’expérience utilisateur et faciliter le débogage.

  3. Tests unitaires : Pour garantir la stabilité de l’API, il est crucial d'écrire des tests unitaires pour chaque endpoint. Utiliser un cadre de test comme pytest permet de s’assurer que chaque opération fonctionne correctement et de valider la fiabilité des fonctionnalités à long terme.

  4. Performance et Scalabilité : Bien que l’utilisation d’un fichier CSV pour stocker les données soit simple et pratique pour un prototype, il devient rapidement inadapté à des systèmes de plus grande envergure. Il est recommandé de passer à une base de données relationnelle (comme PostgreSQL ou MySQL) pour des applications plus complexes.

Comment sécuriser vos applications FastAPI : Authentification, autorisation et gestion des utilisateurs

La sécurité des applications web modernes est un domaine complexe et essentiel, particulièrement lorsque nous travaillons avec des API. Dans le contexte de FastAPI, une des solutions les plus efficaces consiste à utiliser des techniques d'authentification et d'autorisation robustes, combinées à des méthodes de protection des données sensibles. Ce chapitre vous guidera à travers les différentes étapes nécessaires pour mettre en place un système d'enregistrement d'utilisateurs sécurisé, intégrer OAuth2 avec JSON Web Tokens (JWT), gérer l'authentification multi-facteurs (MFA), et bien plus encore.

L’un des aspects fondamentaux dans la mise en place d’une application SaaS (Software as a Service) est la gestion des utilisateurs, depuis leur inscription jusqu'à la sécurisation des points d’accès avec des clés d’API et des protocoles d’authentification externes. Nous verrons comment implémenter un système d'enregistrement simple, ajouter la gestion des rôles avec contrôle d'accès basé sur les rôles (RBAC), et utiliser des services tiers tels que GitHub pour faciliter l'authentification des utilisateurs. Nous aborderons également la mise en œuvre de la double authentification pour renforcer la sécurité.

Mise en place de l'enregistrement des utilisateurs

Le processus d'enregistrement des utilisateurs dans une application FastAPI commence par la collecte et le stockage sécurisé des informations sensibles, comme le mot de passe. Pour ce faire, il est impératif d'utiliser des techniques de hachage pour transformer le mot de passe en une forme irrécupérable, assurant ainsi la sécurité des données.

Dans cette étape, nous allons stocker les informations des utilisateurs dans une base de données SQL, en utilisant SQLAlchemy comme ORM et bcrypt pour le hachage des mots de passe. Tout d'abord, nous devons créer une classe User qui mappe la table des utilisateurs dans la base de données. Cette table contiendra des champs tels que l'identifiant, le nom d'utilisateur, l'email et le mot de passe haché.

Voici un extrait de code pour cette opération :

python
from passlib.context import CryptContext
from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from models import User pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def add_user(session: Session, username: str, password: str, email: str) -> User | None: hashed_password = pwd_context.hash(password) db_user = User(username=username, email=email, hashed_password=hashed_password) session.add(db_user) try: session.commit() session.refresh(db_user) except IntegrityError: session.rollback() return None return db_user

Dans cet exemple, la fonction add_user ajoute un utilisateur à la base de données après avoir haché son mot de passe. Si un utilisateur avec le même nom d'utilisateur ou email existe déjà, une exception IntegrityError est levée.

Ensuite, nous devons définir un point d'API pour gérer l'enregistrement des utilisateurs. Nous allons utiliser FastAPI pour créer un endpoint de type POST, qui recevra les données de l'utilisateur (nom d'utilisateur, email et mot de passe), effectuera les validations nécessaires et renverra une réponse adéquate.

Le point d'API pourrait ressembler à ceci :

python
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session from models import Base from db_connection import get_session from operations import add_user app = FastAPI() @app.post("/register/user") def register(user: UserCreateBody, session: Session = Depends(get_session)): user = add_user(session=session, **user.model_dump()) if not user: raise HTTPException(status_code=409, detail="Nom d'utilisateur ou email déjà existant")
return {"message": "Utilisateur créé", "user": user}

Ici, UserCreateBody est une structure de données contenant les informations nécessaires pour l'inscription d'un utilisateur. Si un utilisateur avec un email ou un nom d'utilisateur déjà existant tente de s'enregistrer, une exception HTTP 409 (Conflit) sera levée.

Intégration d’OAuth2 et JWT pour l’authentification

Une fois l’enregistrement des utilisateurs mis en place, la prochaine étape consiste à gérer leur authentification. OAuth2 avec JSON Web Tokens (JWT) est une méthode courante et robuste pour gérer l'authentification. OAuth2 permet une gestion centralisée des autorisations, et JWT sert à générer des tokens d’accès sécurisés.

Pour intégrer OAuth2 et JWT, nous devons définir un mécanisme permettant de générer un token pour les utilisateurs une fois leur identité vérifiée. Ce token pourra ensuite être utilisé pour effectuer des requêtes authentifiées vers les points d'API.

Un exemple simplifié de code pour générer et vérifier un JWT serait le suivant :

python
from jose import JWTError, jwt from datetime import datetime, timedelta from passlib.context import CryptContext # Clé secrète pour signer les tokens SECRET_KEY = "votre_clé_secrète" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def create_access_token(data: dict, expires_delta: timedelta = timedelta(minutes=15)):
to_encode = data.copy() expire = datetime.utcnow() + expires_delta to_encode.update({
"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password)

L’utilisation de JWT permet à vos utilisateurs de se connecter en toute sécurité et de maintenir leurs sessions actives sans nécessiter de nouvelles connexions à chaque requête. Le token peut être stocké côté client (dans les cookies ou le stockage local), et transmis via les entêtes HTTP.

Authentification multi-facteurs (MFA)

L'ajout d'une couche de sécurité supplémentaire à l'authentification par mot de passe est fortement recommandé. L'authentification multi-facteurs (MFA) permet de renforcer la sécurité en demandant à l'utilisateur de valider son identité via un second facteur, comme un code envoyé par SMS ou généré par une application d'authentification.

L'implémentation de la MFA nécessite de configurer un mécanisme pour gérer ces codes et vérifier leur validité avant de permettre l'accès à l'application. FastAPI, avec ses points d'API et ses capacités de gestion des sessions, est bien adapté pour cela.

Authentification via des services tiers

De plus en plus d’applications intègrent des services tiers tels que GitHub, Google, ou Facebook pour simplifier l’authentification des utilisateurs. Cela permet à vos utilisateurs de se connecter rapidement sans avoir à créer de nouveaux identifiants, et vous permet également de déléguer la gestion de la sécurité à des plateformes éprouvées.

L’intégration de services tiers se fait en utilisant des protocoles standards comme OAuth2, qui sont pris en charge par des bibliothèques comme Authlib ou FastAPI.

Contrôle d’accès basé sur les rôles (RBAC)

Pour gérer l’accès à vos API de manière granulaire, vous pouvez mettre en place un système de contrôle d’accès basé sur les rôles (RBAC). Ce mécanisme vous permet de définir des rôles (par exemple, administrateur, utilisateur, modérateur) et d’attribuer des permissions spécifiques à chaque rôle.

En utilisant FastAPI, vous pouvez créer un système de vérification des rôles de vos utilisateurs via des dépendances, ce qui vous permet de restreindre l’accès aux points d’API en fonction du rôle de l’utilisateur.

Gestion des clés d’API

Pour protéger vos points d'API et vous assurer que seules des requêtes autorisées peuvent y accéder, vous pouvez mettre en place une authentification par clé d’API. Ce mécanisme est particulièrement utile pour l’intégration de services externes ou la création d’applications front-end qui doivent interagir avec votre API sans nécessiter de gestion complexe des sessions.

Les clés d’API peuvent être attribuées à des utilisateurs ou à des services spécifiques, et peuvent être stockées dans une base de données ou un système de gestion sécurisé.

Comment implémenter des tâches en arrière-plan dans FastAPI pour améliorer les performances

Lors de la création d'applications web modernes, l'optimisation des performances devient un élément clé. En particulier, la gestion des tâches en arrière-plan permet de traiter des processus complexes ou de longue durée sans bloquer l'exécution du serveur ou interférer avec l'expérience utilisateur. Dans le cadre de FastAPI, l'implémentation de tâches en arrière-plan peut grandement améliorer la réactivité de votre application, notamment pour des cas comme le stockage de données dans des bases externes ou l'exécution de traitements lourds.

Imaginons que nous avons une application de plateforme de voyages où nous souhaitons enregistrer des messages envoyés via un endpoint, tel que GET /v2/trips/{category}, dans une base de données externe à des fins d'analyse. L’implémentation de cette fonctionnalité s’effectue en deux étapes simples.

Dans un premier temps, nous définissons une fonction de traitement qui simule l’opération d'enregistrement dans un module dédié, par exemple background_tasks.py, dans le répertoire de l’application. Cette fonction pourrait ressembler à ceci :

python
import asyncio import logging logger = logging.getLogger("uvicorn.error")
async def store_query_to_external_db(message: str):
logger.info(
f"Storing message '{message}'.") await asyncio.sleep(2) # Simule une opération non-bloquante logger.info(f"Message '{message}' stored!")

L'opération de stockage est simulée par asyncio.sleep, une opération non-bloquante permettant de ne pas bloquer le fil d'exécution. Des messages de log sont ajoutés pour suivre l’exécution de cette fonction.

Ensuite, pour exécuter cette tâche en arrière-plan, nous modifions l'endpoint GET /v2/trips/{category} dans notre fichier principal main.py. Voici comment procéder :

python
from fastapi import BackgroundTasks @app.get("/v2/trips/{category}") def get_trips_by_category( background_tasks: BackgroundTasks, category: Annotated[select_category, Depends()], discount_applicable: Annotated[bool, Depends(check_coupon_validity)], ): category = category.replace("-", " ").title() message = f"You requested {category} trips." if discount_applicable: message += "\nThe coupon code is valid! You will get a discount!" background_tasks.add_task(store_query_to_external_db, message) logger.info("Query sent to background task, end of request.") return message

Ici, nous avons ajouté la tâche en arrière-plan avec background_tasks.add_task, ce qui permet à la fonction store_query_to_external_db de s'exécuter de manière asynchrone après la réponse de l'API. Lorsque l'utilisateur effectue une requête, la tâche est mise en file d'attente et sera exécutée de manière concurrente sans bloquer le serveur.

Après avoir démarré le serveur avec uvicorn app.main:app, l'appel de l'endpoint GET /v2/trips/cruises déclenchera la tâche en arrière-plan, et vous pourrez observer dans les logs les messages relatifs à l’enregistrement du message.

bash
INFO: Query sent to background task, end of request. INFO: Storing message 'You requested Cruises trips.' INFO: Message 'You requested Cruises trips.' stored!

Bien que cette approche soit suffisante pour des tâches relativement simples, pour des calculs plus intensifs ou des tâches longues, il peut être pertinent d'utiliser des outils dédiés comme Celery, qui permet de gérer les tâches en arrière-plan dans un processus séparé et ainsi éviter les problèmes de performance liés à l'exécution dans le même processus que le serveur web.

Le fonctionnement des tâches en arrière-plan dans FastAPI repose sur le principe de la boucle d’événements (event loop), qui permet d’exécuter les tâches de manière non-bloquante, en les inscrivant dans un objet BackgroundTasks. Cela garantit que l'exécution des tâches ne bloque pas la réponse HTTP et permet une gestion concurrente des tâches.

Dans certains cas, comme des calculs lourds nécessitant une puissance de traitement importante, utiliser des outils comme Celery peut s'avérer plus adapté. Celery permet de gérer des files d'attente et d'exécuter les tâches dans des processus séparés, assurant ainsi que les performances du serveur ne soient pas affectées par des traitements longs.

Outre l’aspect technique de l’implémentation, il est essentiel de comprendre que l’utilisation de tâches en arrière-plan dans une application FastAPI permet de mieux gérer les opérations de longue durée sans affecter l’expérience utilisateur. Cependant, une telle gestion nécessite également une attention particulière aux ressources du serveur, notamment lorsqu'il s'agit de gérer un grand nombre de tâches simultanées ou des opérations nécessitant des ressources importantes.

En somme, l’utilisation des tâches en arrière-plan dans FastAPI est un moyen efficace d’améliorer la réactivité d'une application. Toutefois, pour des scénarios plus complexes, l'usage de solutions comme Celery peut offrir une meilleure scalabilité et résilience. Une bonne gestion de ces tâches et une surveillance adéquate de leur exécution sont cruciales pour garantir des performances optimales.

Comment gérer les connexions WebSocket dans une application FastAPI

Lorsqu'une connexion WebSocket est établie entre un client et un serveur FastAPI, il est essentiel de bien comprendre le processus sous-jacent et les différentes étapes permettant d'assurer une communication bidirectionnelle fiable et fluide. Dans cet article, nous allons explorer la gestion des connexions WebSocket, l'envoi et la réception de messages en temps réel, ainsi que la gestion des déconnexions, un aspect crucial pour maintenir une interaction continue sans perturbations.

Les connexions WebSocket permettent une communication bidirectionnelle, ce qui signifie que les messages peuvent être envoyés et reçus simultanément entre le client et le serveur. Cela contraste avec le modèle traditionnel de requêtes et réponses HTTP, où la communication est unidirectionnelle. En configurant correctement un point de terminaison WebSocket, le serveur peut accepter des connexions clients, envoyer des messages, recevoir des données et fermer la connexion lorsque cela est nécessaire.

Lorsqu'un serveur reçoit une demande de connexion via WebSocket, il démarre un processus appelé "handshake" (serrement de main HTTP) pour établir la connexion avec le client. Cela commence par la méthode await websocket.accept() sur le serveur. Une fois la connexion établie, le serveur peut envoyer un message d'accueil à l'utilisateur grâce à la méthode await websocket.send_text("Bienvenue dans la salle de chat !"). Ce message informe le client que la connexion est réussie et qu'il peut maintenant interagir avec le serveur. L'étape suivante consiste à attendre des messages entrants, les traiter, puis envoyer une réponse. Pour ce faire, la méthode await websocket.receive_text() permet au serveur de recevoir les messages envoyés par le client, de les stocker dans une variable, puis de les traiter selon les besoins.

L'exemple de code suivant montre comment configurer un tel endpoint dans FastAPI :

python
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
await websocket.accept() await websocket.send_text("Bienvenue dans la salle de chat !") while True: data = await websocket.receive_text() logger.info(f"Message reçu : {data}") await websocket.send_text("Message reçu !")

Dans cet exemple, le serveur attend continuellement des messages du client et répond par un accusé de réception. Cette approche permet de maintenir une communication ouverte sans fermer la connexion après chaque message, ce qui est crucial pour les applications en temps réel comme les chats en ligne.

Il est important de noter que, bien que ce modèle de communication soit très utile, il existe plusieurs défis liés à la gestion de la connexion WebSocket, notamment les déconnexions et les erreurs qui peuvent survenir. Par exemple, lorsqu'un client se déconnecte, une exception WebSocketDisconnect peut être levée si la déconnexion n'est pas correctement gérée. Cela nécessite l'ajout de blocs try-except dans le code pour intercepter cette exception et ainsi éviter des erreurs non capturées sur le serveur. Voici un exemple pour gérer la déconnexion côté client :

python
from fastapi.websockets import WebSocketDisconnect
@app.websocket("/ws") async def ws_endpoint(websocket: WebSocket): await websocket.accept() await websocket.send_text("Bienvenue dans la salle de chat !") try: while True: data = await websocket.receive_text() logger.info(f"Message reçu : {data}") await websocket.send_text("Message reçu !") except WebSocketDisconnect: logger.warning("Connexion fermée par le client")

Cette modification permet au serveur de capter les déconnexions des clients et de les traiter de manière appropriée, en enregistrant un avertissement dans les logs sans provoquer de crash du serveur.

De même, une déconnexion côté serveur peut se produire si le serveur décide de fermer la connexion pour une raison spécifique, comme la réception d’un message particulier. Par exemple, si le serveur reçoit le message "disconnect", il peut alors fermer la connexion. Cette gestion proactive des déconnexions côté serveur assure une meilleure stabilité et un contrôle accru sur le flux de communication.

Voici un exemple de gestion d'une déconnexion côté serveur :

python
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
await websocket.accept() await websocket.send_text("Bienvenue dans la salle de chat !") while True: data = await websocket.receive_text() if data == "disconnect": await websocket.send_text("Déconnexion en cours...") await websocket.close() break logger.info(f"Message reçu : {data}") await websocket.send_text("Message reçu !")

Dans ce scénario, le serveur attend des messages et, lorsqu'il reçoit "disconnect", il répond avec un message et ferme la connexion avec await websocket.close(). Cela permet de contrôler explicitement la durée de la connexion.

L'échange de messages en temps réel via WebSocket est une fonctionnalité puissante dans le développement d'applications modernes. Il permet des interactions rapides et fluides, mais nécessite une gestion attentive des connexions et des erreurs. Lors de la conception de tels systèmes, il est crucial de penser à la gestion des déconnexions, tant côté client que serveur, pour garantir la stabilité et la réactivité de l'application.

Il est également important de noter que, bien que la gestion des WebSockets dans FastAPI soit relativement simple, elle nécessite des outils appropriés pour tester les connexions. Par exemple, Postman peut être utilisé pour simuler des connexions WebSocket et tester l'envoi et la réception de messages en temps réel. De plus, il existe des fonctionnalités avancées comme la validation des messages reçus au format JSON ou binaire, qui peuvent être utilisées pour rendre l'application encore plus robuste.

Enfin, bien que FastAPI ne supporte pas directement les WebSockets dans la documentation Swagger, vous pouvez toujours suivre les discussions en cours sur GitHub pour des mises à jour futures concernant cette fonctionnalité.

Comment intégrer FastAPI avec d'autres bibliothèques Python et GraphQL

Pour débuter l'intégration de FastAPI avec GraphQL, il est essentiel de configurer un environnement de développement adapté en installant les bibliothèques nécessaires. Vous pouvez utiliser le fichier requirements.txt du projet ou la commande pip pour installer FastAPI ainsi que d'autres dépendances comme Uvicorn et Strawberry-GraphQL :

bash
$ pip install fastapi uvicorn strawberry-graphql[fastapi]

Une fois l'installation terminée, vous pouvez entamer la création d'un point d'accès GraphQL en suivant plusieurs étapes.

La première consiste à définir un module database.py, où vous allez simuler une base de données avec une liste d'utilisateurs. Commencez par créer une classe User avec les attributs suivants : identifiant (id), nom d'utilisateur (username), numéro de téléphone (phone_number) et pays (country). Utilisez pydantic.BaseModel pour garantir la validation des données.

python
from pydantic import BaseModel class User(BaseModel): id: int username: str phone_number: str country: str

Ensuite, créez une liste d'objets User qui servira de source de données dans ce module. Par exemple, cette liste pourrait contenir plusieurs utilisateurs provenant de pays différents. Pour cet exemple, voici comment vous pourriez la définir :

python
users_db: list[User] = [
User(id=1, username="user1", phone_number="1234567890", country="USA"), # Autres utilisateurs... ]

Ensuite, dans un module séparé nommé graphql_utils.py, il vous faudra définir une requête GraphQL pour interroger cette "base de données". Pour ce faire, vous devez d'abord définir le modèle qui sera retourné par la requête. Par exemple, vous pouvez utiliser la bibliothèque Strawberry pour définir un type GraphQL correspondant à un utilisateur.

python
import strawberry @strawberry.type class User: username: str phone_number: str country: str

La prochaine étape consiste à définir la requête elle-même, qui permettra de filtrer les utilisateurs en fonction du pays passé en argument. Voici un exemple de cette requête GraphQL :

python
@strawberry.type
class Query: @strawberry.field def users(self, country: str | None) -> list[User]: return [ User(username=user.username, phone_number=user.phone_number, country=user.country) for user in users_db if user.country == country ]

Cette requête retourne tous les utilisateurs d'un pays spécifié. Une fois la requête définie, vous pouvez créer un schéma GraphQL et l'intégrer à FastAPI via le routeur GraphQLRouter de Strawberry.

python
from strawberry.fastapi import GraphQLRouter
schema = strawberry.Schema(Query) graphql_app = GraphQLRouter(schema)

Enfin, dans un module main.py, vous allez inclure ce routeur GraphQL à l'application FastAPI. L'URL de l'endpoint sera /graphql, ce qui permettra d'accéder à l'interface interactive de GraphQL via le navigateur.

python
from fastapi import FastAPI from graphql_utils import graphql_app app = FastAPI() app.include_router(graphql_app, prefix="/graphql")

Lorsque vous lancez le serveur avec Uvicorn :

bash
$ uvicorn main:app

Vous pouvez accéder à l'interface interactive de GraphQL en vous rendant à l'adresse http://localhost:8000/graphql dans votre navigateur. Dans l'éditeur de requêtes GraphQL à gauche, vous pouvez saisir la requête suivante pour obtenir la liste des utilisateurs d'un pays spécifique, par exemple, les utilisateurs des États-Unis :

graphql
{
users(country: "USA") { username country phoneNumber } }

Le panneau de droite affichera les résultats sous forme JSON, vous permettant de visualiser les données extraites de la base.

En combinant les points d'accès RESTful et GraphQL, vous pouvez considérablement étendre les possibilités de requêtes et de modifications de données. Dans des scénarios réels, les points d'accès REST peuvent être utilisés pour modifier la base de données, tandis que GraphQL sert à extraire des informations spécifiques selon des critères définis. Cela permet une plus grande flexibilité dans la gestion et l'extraction des données.


Il est important de noter que l'utilisation conjointe de REST et de GraphQL dans une même application nécessite une gestion soigneuse des endpoints afin de ne pas dupliquer les responsabilités entre les deux technologies. Par ailleurs, l'intégration de GraphQL avec FastAPI permet de bénéficier de la rapidité de FastAPI tout en ayant la souplesse de GraphQL pour des requêtes de données complexes.