FastAPI: Async & Real-time Features 🚀

The Restaurant That Never Sleeps

Imagine you run a very busy restaurant. You have ONE waiter. Every time a customer orders food, should the waiter stand at the kitchen door waiting for the chef to finish cooking? Or should the waiter take orders from other tables while the food is being prepared?

That’s exactly what async programming is about!

In FastAPI, async means your server can handle multiple requests at the same time, without getting stuck waiting.

🎭 Part 1: Async Path Operations

What is an Async Path Operation?

Think of a path operation as a door to your restaurant. When someone knocks (makes a request), you answer.

Regular (Sync) Door:

@app.get("/food")
def get_food():
    # Waiter stands and waits
    food = cook_slowly()
    return food

Async Door:

@app.get("/food")
async def get_food():
    # Waiter goes to help others
    food = await cook_slowly()
    return food

The Magic Word: await

When you see await, think of it as telling the waiter:

“Go help other customers. Come back when this is ready!”

from fastapi import FastAPI

app = FastAPI()

@app.get("/weather")
async def get_weather():
    # Go do other things while waiting
    data = await fetch_weather_api()
    return {"weather": data}

When to Use async?

graph TD
    A[Does your code wait for something?] --> B{What kind?}
    B --> C[Database query]
    B --> D[External API call]
    B --> E[File reading]
    C --> F[Use async + await]
    D --> F
    E --> F
    A --> G[Just calculations?]
    G --> H[Regular def is fine]

Simple Example: Two Customers

Without async:

• Customer A orders → waits 5 seconds → served
• Customer B orders → waits 5 seconds → served
• Total time: 10 seconds

With async:

• Customer A orders → cooking starts
• Customer B orders → cooking starts
• Both served around the same time!
• Total time: ~5 seconds

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/order/{item}")
async def place_order(item: str):
    # This simulates cooking time
    await asyncio.sleep(3)
    return {"order": item, "status": "ready!"}

🎒 Part 2: Background Tasks

The Busy Bee Helper

Sometimes you want to tell the customer “Your order is placed!” immediately, but then do extra work in the background.

Real example: When you sign up for a website, it says “Welcome!” right away. But the email confirmation is sent in the background.

How Background Tasks Work

from fastapi import BackgroundTasks

@app.post("/signup")
async def signup(
    email: str,
    background_tasks: BackgroundTasks
):
    # This happens AFTER the response
    background_tasks.add_task(
        send_welcome_email,
        email
    )
    # Customer sees this immediately!
    return {"message": "Welcome!"}

def send_welcome_email(email: str):
    # This runs in background
    print(f"Sending email to {email}")

The Kitchen Ticket System

graph TD
    A[Customer Orders] --> B[Waiter says: Coming right up!]
    B --> C[Response sent immediately]
    B --> D[Kitchen gets ticket]
    D --> E[Background: Prepare food]
    E --> F[Background: Pack order]

Multiple Background Tasks

You can add many tasks! They run one after another.

@app.post("/order")
async def create_order(
    order_id: str,
    background_tasks: BackgroundTasks
):
    # Add multiple background jobs
    background_tasks.add_task(
        save_to_database,
        order_id
    )
    background_tasks.add_task(
        notify_kitchen,
        order_id
    )
    background_tasks.add_task(
        send_receipt,
        order_id
    )

    return {"order_id": order_id}

When to Use Background Tasks?

🔌 Part 3: WebSocket Endpoints

The Two-Way Walkie-Talkie

Regular HTTP is like sending letters:

• You send a letter → Wait → Get a reply
• For every question, you send a new letter

WebSocket is like a phone call:

• You connect once
• Both sides can talk anytime
• No need to reconnect!

Why WebSockets Matter

graph LR
    subgraph Regular HTTP
    A[Client] -->|Request 1| B[Server]
    B -->|Response 1| A
    A -->|Request 2| B
    B -->|Response 2| A
    end

graph LR
    subgraph WebSocket
    C[Client] <-->|Always connected| D[Server]
    end

Your First WebSocket

from fastapi import WebSocket

@app.websocket("/chat")
async def chat(websocket: WebSocket):
    # Step 1: Accept the connection
    await websocket.accept()

    while True:
        # Step 2: Wait for message
        message = await websocket.receive_text()

        # Step 3: Send response
        await websocket.send_text(
            f"You said: {message}"
        )

The Chat Room Example

Think of it like this:

1. 📱 Phone rings (client connects)
2. 📞 You pick up (accept connection)
3. 🗣️ They talk (receive message)
4. 💬 You reply (send message)
5. 🔁 Keep talking (loop continues)
6. 📴 Hang up (connection closes)

Handling Multiple Users

# Keep track of all connected users
connected_users = []

@app.websocket("/chat-room")
async def chat_room(websocket: WebSocket):
    await websocket.accept()
    connected_users.append(websocket)

    try:
        while True:
            message = await websocket.receive_text()
            # Send to ALL users
            for user in connected_users:
                await user.send_text(message)
    except:
        connected_users.remove(websocket)

WebSocket Message Types

# Text messages (like chat)
text = await websocket.receive_text()
await websocket.send_text("Hello!")

# JSON data (structured info)
data = await websocket.receive_json()
await websocket.send_json({"status": "ok"})

# Binary (files, images)
bytes_data = await websocket.receive_bytes()
await websocket.send_bytes(image_data)

🎯 Putting It All Together

Real-World Scenario: Live Game Score

from fastapi import FastAPI, WebSocket
from fastapi import BackgroundTasks

app = FastAPI()
game_subscribers = []

# WebSocket: Live updates
@app.websocket("/live-score")
async def live_score(websocket: WebSocket):
    await websocket.accept()
    game_subscribers.append(websocket)

    try:
        while True:
            await websocket.receive_text()
    except:
        game_subscribers.remove(websocket)

# Async endpoint: Update score
@app.post("/update-score")
async def update_score(
    team: str,
    score: int,
    background_tasks: BackgroundTasks
):
    # Background: Notify all watchers
    background_tasks.add_task(
        broadcast_score,
        team,
        score
    )
    return {"updated": True}

async def broadcast_score(team, score):
    for subscriber in game_subscribers:
        await subscriber.send_json({
            "team": team,
            "score": score
        })

🌟 Remember This!

The Golden Rules

1. async = “I can do other things while waiting”
2. await = “Start this, come back when done”
3. BackgroundTasks = “Do this later, reply now”
4. WebSocket = “Stay connected, talk anytime”

🎉 You Did It!

You now understand FastAPI’s real-time superpowers:

✅ Async Path Operations - Handle many requests without blocking ✅ Background Tasks - Do work after responding ✅ WebSockets - Real-time two-way communication

Your FastAPI apps can now be fast, responsive, and real-time! 🚀