Pourquoi WebSockets avec FastAPI ?
WebSockets est une technologie qui permet la communication bidirectionnelle en temps réel entre un client et un serveur. Cela signifie que les données peuvent être envoyées du client au serveur et vice versa, sans avoir besoin de rafraîchir la page ou d'envoyer une requête HTTP pour obtenir les nouvelles données.
Dans le contexte professionnel, il existe de nombreux cas d'utilisation où un dev aurait besoin de WebSockets :
- Un chat en direct : chaque message envoyé par un utilisateur est immédiatement visible pour tous les autres.
- Une application de jeu en ligne : les interactions entre les joueurs doivent être instantanées et réactives.
- Un service de notifications en temps réel : des mises à jour instantanées sur le statut d'une tâche ou un message.
Prerequis
Avant de commencer, vous devriez avoir les connaissances suivantes :
- Connaissance approfondie de FastAPI.
- Compréhension des concepts de base de WebSockets, notamment la différence entre HTTP et WebSocket.
- Un environnement Python configuré avec pip pour installer des packages.
Les outils à installer sont :
pip install fastapi uvicorn websockets
Concepts fondamentaux
1. Connexion WebSocket
Un client est connecté au serveur via une connexion WebSocket. Cette connexion est établie grâce à un handshake HTTP et est maintenue ouverte pendant toute la durée de la session.
## Importer les bibliothèques nécessaires
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html_content)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
2. Communication bidirectionnelle
La communication WebSocket est bidirectionnelle, ce qui signifie que le serveur peut envoyer des données au client à tout moment.
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
3. Gestion des exceptions
Il est important de gérer les erreurs et les exceptions pour une connexion WebSocket stables.
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
try:
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except Exception as e:
print(e)
Mise en pratique : projet fil rouge
Étape 1 : Création du projet
Créez un nouveau projet FastAPI.
mkdir fastapi_websockets_project
cd fastapi_websockets_project
python -m venv venv
source venv/bin/activate # Sur Windows utilisez `venv\Scripts\activate`
pip install fastapi uvicorn websockets
Étape 2 : Création des fichiers
Créez les fichiers suivants :
main.py: le fichier principal du projet.templates/index.html: le template HTML pour la page Web.
touch main.py templates/index.html
Étape 3 : Configuration de FastAPI
Ouvrez main.py et ajoutez le code suivant :
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import asyncio
app = FastAPI()
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html_content)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
try:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except Exception as e:
print(e)
Étape 4 : Exécution du serveur
Exécutez le serveur avec uvicorn.
uvicorn main:app --reload
Allez à http://127.0.0.1:8000 dans votre navigateur pour voir l'application en action.
Erreurs frequentes et debugging
1. Connexion WebSocket échoue
Erreur :
Exception in ASGI application
Traceback (most recent call last):
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 450, in serve
await self.run()
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 568, in run
await loop.run_until_complete(self.startup())
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 741, in startup
await self.app().startup()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/applications.py", line 526, in startup
await run_in_threadpool(self._call_on_event, "startup")
File "/home/user/venv/lib/python3.9/site-packages/starlette/applications.py", line 148, in _call_on_event
await callback()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/applications.py", line 523, in _call_on_event
await run_in_threadpool(self._call_on_event, "startup")
File "/home/user/venv/lib/python3.9/site-packages/starlette/applications.py", line 148, in _call_on_event
await callback()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/routing.py", line 657, in startup_event
await self.websocket_routes.startup()
File "/home/user/venv/lib/python3.9/site-packages/starlette/websockets.py", line 40, in startup
raise RuntimeError("WebSocket routes are only supported when using an ASGI server.")
RuntimeError: WebSocket routes are only supported when using an ASGI server.
Correction :
## Remplacer l'importation par :
from fastapi import FastAPI, WebSocket
2. Message non reçu
Erreur : Le client ne reçoit pas les messages du serveur.
Correction : Assurez-vous que le client est correctement connecté et que le serveur envoie bien des messages.
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
3. Erreur lors de la fermeture de la connexion
Erreur :
Exception in ASGI application
Traceback (most recent call last):
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 450, in serve
await self.run()
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 568, in run
await loop.run_until_complete(self.startup())
File "/home/user/venv/lib/python3.9/site-packages/uvicorn/server.py", line 741, in startup
await self.app().startup()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/applications.py", line 526, in startup
await run_in_threadpool(self._call_on_event, "startup")
File "/home/user/venv/lib/python3.9/site-packages/starlette/applications.py", line 148, in _call_on_event
await callback()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/applications.py", line 523, in _call_on_event
await run_in_threadpool(self._call_on_event, "startup")
File "/home/user/venv/lib/python3.9/site-packages/starlette/applications.py", line 148, in _call_on_event
await callback()
File "/home/user/venv/lib/python3.9/site-packages/fastapi/routing.py", line 657, in startup_event
await self.websocket_routes.startup()
File "/home/user/venv/lib/python3.9/site-packages/starlette/websockets.py", line 40, in startup
raise RuntimeError("WebSocket routes are only supported when using an ASGI server.")
RuntimeError: WebSocket routes are only supported when using an ASGI server.
Correction : Gérer les erreurs lors de la fermeture de la connexion.
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except WebSocketDisconnect:
print("Client disconnected")
Pour aller plus loin
1. Authentification et sécurité des connexions WebSocket
Pour une application de production, il est important d'ajouter une authentification et de sécuriser les connexions WebSocket.
2. Gestion du nombre de connexions concurrentes
Il peut être nécessaire de gérer le nombre de connexions simultanées pour éviter un surcharge du serveur.
3. Stockage des données
Pour une application plus complexe, il est utile de stocker les données dans une base de données.
Défi pratique : Création d'une messagerie instantanée entre deux utilisateurs
Créez une application FastAPI qui permet à deux utilisateurs de s'envoyer des messages en temps réel. Utilisez un système de fil de discussion pour stocker les messages et affichez-les dans le navigateur.
Commencez par créer un fichier models.py pour définir le modèle de données :
from pydantic import BaseModel
class Message(BaseModel):
sender: str
content: str
Ensuite, modifiez main.py pour ajouter la gestion des messages et le stockage dans une liste (pour simplifier) :
from fastapi import FastAPI, WebSocket, HTTPException
from pydantic import BaseModel
app = FastAPI()
messages = []
class Message(BaseModel):
sender: str
content: str
@app.get("/")
async def get():
return HTMLResponse(html_content)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
message = Message(**data)
messages.append(message)
for client in app.state.clients:
if client != websocket:
await client.send_text(f"Message from {message.sender}: {message.content}")
except WebSocketDisconnect:
pass
@app.on_event("startup")
async def startup():
app.state.clients = set()
@app.on_event("shutdown")
async def shutdown():
for client in app.state.clients:
await client.close()
Ensuite, modifiez html_content pour inclure le formulaire d'envoi de messages :
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<script>
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(JSON.stringify({ sender: 'User', content: input.value }))
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
Enfin, exécutez le serveur et accédez à http://127.0.0.1:8000 pour tester l'application.
Ce tutoriel vous a montré comment utiliser WebSockets avec FastAPI pour créer des applications en temps réel. Vous avez appris les concepts fondamentaux, comment mettre en pratique le développement avec FastAPI et comment gérer les erreurs courantes. Vous êtes maintenant prêt à approfondir votre connaissance et à développer de nombreuses applications utilisant WebSockets.