Bienvenue dans le monde passionnant de FastAPI et WebSockets ! Aujourd'hui, nous allons plonger au cœur de la mécanique des WebSockets en utilisant FastAPI. Que vous soyez un développeur chevronné ou débutant, ce guide vous présentera les concepts essentiels et les implémentations pratiques des WebSockets. Mais avant de commencer, n'oublions pas un outil pratique qui facilitera grandement le développement de votre API : Apidog. Vous pouvez télécharger Apidog gratuitement et rationaliser vos tests et votre documentation d'API sans effort.
Qu'est-ce que FastAPI ?
FastAPI est un framework web moderne et rapide (haute performance) pour la création d'API avec Python 3.7+ basé sur des indications de type Python standard. Il vous permet de créer rapidement des API web robustes et performantes. L'une des caractéristiques remarquables de FastAPI est sa capacité à gérer la programmation asynchrone et les WebSockets avec facilité.
Que sont les WebSockets ?
WebSockets sont un protocole de communication qui fournit des canaux de communication full-duplex sur une seule connexion TCP. Contrairement à HTTP, qui est un protocole de requête-réponse, les WebSockets permettent une communication bidirectionnelle entre le client et le serveur. Cela les rend parfaits pour les applications en temps réel telles que les applications de chat, les notifications en direct et les jeux.
Pourquoi utiliser FastAPI avec WebSockets ?
La conception de FastAPI facilite la création d'API qui prennent en charge les WebSockets. Avec FastAPI, vous pouvez créer des API qui sont non seulement rapides et fiables, mais qui prennent également en charge la communication asynchrone en utilisant les WebSockets. Cette combinaison est parfaite pour le développement d'applications modernes en temps réel.
Démarrer avec FastAPI WebSockets
Pour commencer, vous devrez installer FastAPI et Uvicorn, qui est un serveur ASGI pour servir vos applications FastAPI.
pip install fastapi uvicorn
Une fois que vous avez installé les packages nécessaires, vous pouvez commencer à créer votre application FastAPI avec la prise en charge des WebSocket.
Création d'une application WebSocket FastAPI de base
Commençons par un exemple simple pour comprendre les bases de la communication WebSocket dans FastAPI.
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>WebSocket Example</title>
</head>
<body>
<h1>WebSocket Example</h1>
<button onclick="connectWebSocket()">Connect</button>
<script>
function connectWebSocket() {
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
const message = event.data;
alert("Message from server: " + message);
};
ws.onopen = function() {
ws.send("Hello Server");
};
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
Dans cet exemple, nous avons une simple page HTML qui se connecte à un serveur WebSocket lorsqu'on clique sur un bouton. Le serveur reçoit le message "Hello Server" et répond avec "Message text was: Hello Server".
Gestion de plusieurs connexions
L'une des forces des WebSockets est la gestion simultanée de plusieurs connexions client. Développons notre exemple pour gérer plusieurs connexions et diffuser des messages à tous les clients connectés.
from fastapi import FastAPI, WebSocket
from typing import List
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client says: {data}")
except:
manager.disconnect(websocket)
Dans cet exemple mis à jour, nous avons introduit une classe ConnectionManager pour gérer plusieurs connexions WebSocket. La méthode broadcast envoie un message à tous les clients connectés. De cette façon, chaque client reçoit tout message envoyé par un autre client.
Passer des paramètres aux points de terminaison WebSocket
Parfois, vous devrez peut-être passer des paramètres à vos points de terminaison WebSocket. FastAPI facilite cela. Voici un exemple :
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await websocket.accept()
await websocket.send_text(f"Hello, Client {client_id}")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message from {client_id}: {data}")
Dans cet exemple, nous définissons un point de terminaison WebSocket qui accepte un paramètre client_id. Ce paramètre peut être utilisé pour identifier le client et personnaliser les messages.
Intégration des WebSockets avec les points de terminaison API
FastAPI vous permet d'intégrer la communication WebSocket avec vos points de terminaison API existants. Ceci est utile lorsque vous devez combiner la communication en temps réel avec les requêtes API traditionnelles.
from fastapi import FastAPI, WebSocket, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 0:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id, "name": "Item name"}
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Received: {data}")
Dans cet exemple, nous avons un point de terminaison API traditionnel /items/{item_id} ainsi qu'un point de terminaison WebSocket /ws. Les deux points de terminaison peuvent être utilisés simultanément, ce qui vous permet de gérer à la fois la communication synchrone et asynchrone dans votre application.
Considérations de sécurité
Lorsque vous travaillez avec les WebSockets, il est important de prendre en compte les aspects de sécurité, tels que l'authentification et l'autorisation. FastAPI fournit des outils pour sécuriser vos connexions WebSocket.
Sécurisation des connexions WebSocket
Vous pouvez utiliser le système d'injection de dépendance de FastAPI pour sécuriser vos points de terminaison WebSocket. Voici un exemple de la façon de sécuriser un point de terminaison WebSocket avec une simple authentification basée sur des jetons.
from fastapi import FastAPI, WebSocket, Depends, HTTPException, status
app = FastAPI()
def get_token_header(token: str = None):
if token != "mysecrettoken":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token_header)):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Received: {data}")
Dans cet exemple, nous utilisons la fonction Depends pour injecter la dépendance get_token_header, qui vérifie la validité du jeton. Si le jeton n'est pas valide, une erreur HTTP 403 Forbidden est levée.
Gestion des données binaires
Les WebSockets prennent également en charge la transmission de données binaires. Cela peut être utile pour envoyer des fichiers ou des images. FastAPI facilite la gestion des données binaires.
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("Connected to WebSocket")
while True:
data = await websocket.receive_bytes()
await websocket.send_bytes(data)
Dans cet exemple, le point de terminaison WebSocket reçoit des données binaires en utilisant receive_bytes et les renvoie au client en utilisant send_bytes.
Applications concrètes de FastAPI WebSockets
Les WebSockets sont incroyablement polyvalents et peuvent être utilisés dans une variété d'applications concrètes. Voici quelques exemples :
Application de chat en temps réel
L'un des cas d'utilisation les plus courants des WebSockets est la création d'applications de chat en temps réel. Avec FastAPI, vous pouvez facilement créer un serveur de chat qui gère plusieurs clients et diffuse des messages en temps réel.
Notifications en direct
Les WebSockets sont parfaits pour la diffusion de notifications en direct. Par exemple, vous pouvez utiliser les WebSockets pour informer les utilisateurs des nouveaux messages, des mises à jour ou des alertes en temps réel.
Jeux en ligne
Les WebSockets sont largement utilisés dans les jeux en ligne pour permettre la communication en temps réel entre les joueurs. La prise en charge de la programmation asynchrone de FastAPI en fait un excellent choix pour la création de serveurs de jeux.
Comment vous pouvez utiliser Apidog pour déboguer un client WebSocket ?
La gestion des API et des données en temps réel peut être difficile. C'est là qu'Apidog entre en jeu. Apidog simplifie le développement, les tests et la gestion des API, ce qui facilite la gestion des requêtes HTTP avec FastAPI et des mises à jour en temps réel avec WebSockets.
- Ouvrez Apidog : Tout d'abord, démarrez l'application Apidog et cliquez sur le bouton « + » sur le côté gauche, un nouveau menu déroulant s'ouvrira. De là, choisissez « Nouvelle API WebSocket » :
2. Établir une connexion : Commencez par saisir l'URL de l'API WebSocket dans la barre d'adresse d'Apidog. Vous pouvez ensuite cliquer sur le bouton « Connecter » pour lancer le processus de négociation et établir une connexion. Apidog vous permet de personnaliser des paramètres tels que Params, Headers et Cookies pendant la négociation.
3. Envoyer et recevoir des messages : Une fois la connexion établie, vous pouvez envoyer des messages sous l'onglet « Message ». Vous pouvez écrire des messages au format texte, JSON, XML, HTML et autres, ainsi que des messages au format binaire en utilisant Base64 ou Hexadécimal. La nouvelle vue chronologique d'Apidog affiche l'état de la connexion, les messages envoyés et les messages reçus par ordre chronologique. Lorsque vous cliquez sur un message, vous pouvez facilement afficher ses détails.
4. Documentation de l'API : Apidog hérite de sa fonctionnalité de documentation d'API robuste pour les API WebSocket, ce qui vous permet de documenter efficacement vos interactions WebSocket.
N'oubliez pas de vérifier les problèmes courants de WebSocket tels que les problèmes de connexion, les erreurs de format de message et les problèmes de sécurité lors de votre processus de débogage.
Utilisation d'Apidog pour envoyer des requêtes HTTP
Apidog offre plusieurs fonctionnalités avancées qui améliorent encore sa capacité à tester les requêtes HTTP. Ces fonctionnalités vous permettent de personnaliser vos requêtes et de gérer des scénarios plus complexes sans effort.
Étape 1 : Ouvrez Apidog et créez une nouvelle requête.
Étape 2 : Trouvez ou saisissez manuellement les détails de l'API pour la requête POST que vous souhaitez effectuer.
Étape 3 : Remplissez les paramètres requis et toutes les données que vous souhaitez inclure dans le corps de la requête.
Conclusion
FastAPI et WebSockets sont une combinaison puissante pour la création d'applications modernes en temps réel. Que vous construisiez une application de chat, un système de notification en direct ou un jeu en ligne, FastAPI fournit les outils dont vous avez besoin pour gérer efficacement la communication WebSocket.
N'oubliez pas de télécharger Apidog gratuitement pour rationaliser votre processus de développement d'API. Apidog facilite les tests et la documentation de vos API, ce qui vous fait gagner du temps et des efforts.
