Background Tasks and WebSockets (Optional)

Introduction

Not every side effect belongs in the main request path. BackgroundTasks runs lightweight work after sending a response—email notifications, audit logs. WebSockets enable bidirectional real-time communication. This chapter covers both; skip until you need them—Celery handles heavy jobs (Flask caching/tasks).

Prerequisites

BackgroundTasks

python
from fastapi import BackgroundTasks, FastAPI
 
app = FastAPI()
 
def write_audit_log(message: str):
    with open("audit.log", "a", encoding="utf-8") as f:
        f.write(message + "\n")
 
@app.post("/orders")
def create_order(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_audit_log, "order created")
    return {"status": "accepted"}

Code explanation:

  • Tasks run after the response is sent to the client
  • Same process, same event loop—not durable across restarts

Email-style task (concept):

python
def send_welcome_email(email: str):
    # SMTP or provider API
    pass
 
@app.post("/users")
def register_user(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"registered": email}

Warning

BackgroundTasks Are Not a Job Queue

Long reports, video encoding, or guaranteed delivery need Celery, RQ, or cloud queues.

Async Route With Background Task

python
@app.post("/notify")
async def notify(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_audit_log, "async route fired")
    return {"ok": True}

Background function can be sync or async—FastAPI runs accordingly.

WebSocket Echo

Test with browser console or websocat:

bash
websocat ws://127.0.0.1:8000/ws/echo

Connection Manager (Broadcast Concept)

Production needs room isolation, auth, and scaling with Redis pub/sub—beyond this intro.

WebSocket and JWT (Concept)

Pass token as query param or first message—validate before accept():

python
from fastapi import Query, WebSocket, WebSocketDisconnect
 
@app.websocket("/ws/secure")
async def secure_ws(websocket: WebSocket, token: str = Query(...)):
    user_id = decode_token(token)
    if user_id is None:
        await websocket.close(code=1008)
        return
    await websocket.accept()
    ...

Prefer short-lived tokens for WebSocket URLs.

Nginx WebSocket Proxy

Production requires upgrade headers—see deployment chapter:

nginx
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";

FAQ

Background task failed silently?

Exceptions may log to server—wrap task body in try/except and log.

Multiple tasks order?

Run in order added—no parallelism guarantee.

WebSocket vs SSE?

SSE is one-way server→client over HTTP—simpler for live feeds.

Load test WebSockets?

Use dedicated tools; one connection per client—different from HTTP QPS.

async def required for WebSocket?

Yes—WebSocket endpoints should be async.

Replace Celery?

No for retries, scheduling, and worker pools.