FastAPI · Lesson 11 of 15
Background Tasks and WebSockets
Run work after the response and push events over a WebSocket.
- Advanced
- 16 min read
- 3 objectives
Before this lessonLesson 10: Settings with pydantic-settings
What you will learn
- Add a BackgroundTask
- Accept a WebSocket
- Broadcast a message
Your Progress
0 of 15 lessons 0%
- Lessons0 / 15
- Completed0
- Est. time left~ 4 hours
Create a free account to keep your progress on every device.
Two ways to leave the request/response cycle: run a function after the response is sent, or keep a socket open.
BackgroundTasks
from fastapi import BackgroundTasks
def write_audit(user_id: int, action: str):
Path("audit.log").write_text(f"{user_id} {action}\n", encoding="utf-8")
@app.post("/items")
def create(item: ItemIn, bg: BackgroundTasks, user=Depends(get_current_user)):
saved = save_item(item)
bg.add_task(write_audit, user.id, f"created {saved.id}")
return savedThese run in the same process. For retries, schedules or work that must survive a restart, use Redis Queue, Celery or a cloud queue instead.
WebSockets
from fastapi import WebSocket, WebSocketDisconnect
clients: set[WebSocket] = set()
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await ws.accept()
clients.add(ws)
try:
while True:
msg = await ws.receive_text()
for c in list(clients):
await c.send_text(msg)
except WebSocketDisconnect:
clients.discard(ws)