Background Tasks and Async Operations¶

🎯 What You’ll Learn¶

• How to write async endpoints in FastAPI and why they matter
• How FastAPI integrates with Python’s asyncio for non-blocking I/O
• How to run background jobs (e.g., sending emails, logging, cleanup) without blocking the main request
• Best practices for using async and background tasks in production

⚡ Step 1: Async Endpoints¶

🧠 Why Async?¶

FastAPI is built on Starlette and asyncio, which means it can handle thousands of concurrent requests efficiently. Async endpoints are useful when:

• You’re calling external APIs
• You’re querying databases asynchronously
• You’re performing I/O-bound tasks (file operations, network calls)

📄 Example: Async Endpoint¶

from fastapi import APIRouter
import httpx

router = APIRouter(prefix="/async", tags=["Async"])

@router.get("/weather")
async def get_weather():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.weatherapi.com/v1/current.json?key=demo&q=Tokyo")
    return response.json()

🔍 What’s happening?¶

• async def makes the endpoint asynchronous
• httpx.AsyncClient performs non-blocking HTTP requests
• await ensures the coroutine completes before returning

⚡ Step 2: Background Jobs¶

🧠 Why Background Jobs?¶

Sometimes you need to perform tasks after returning a response — without making the user wait. Examples:

• Sending confirmation emails
• Logging analytics
• Cleaning up temporary files

FastAPI provides a BackgroundTasks utility for this.

📄 Example: Background Task¶

from fastapi import APIRouter, BackgroundTasks

router = APIRouter(prefix="/jobs", tags=["Jobs"])

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@router.post("/signup")
def signup(username: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"New signup: {username}")
    return {"message": f"User {username} created"}

🔍 What’s happening?¶

• BackgroundTasks is injected into the endpoint
• background_tasks.add_task() schedules write_log() to run after the response is sent
• The user gets a fast response, while the log is written in the background

⚡ Step 3: Combining Async + Background Tasks¶

You can mix async endpoints with background jobs:

from fastapi import APIRouter, BackgroundTasks
import httpx

router = APIRouter(prefix="/combo", tags=["Combo"])

async def notify_admin(user: str):
    async with httpx.AsyncClient() as client:
        await client.post("https://example.com/notify", json={"user": user})

@router.post("/register")
async def register(user: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(notify_admin, user)
    return {"message": f"User {user} registered"}

Here:

• The endpoint is async (non-blocking)
• The background task (notify_admin) is also async

🧠 Best Practices¶

• Use async endpoints for I/O-bound tasks (API calls, DB queries)
• Use background tasks for non-critical work that can happen after response
• Don’t use background tasks for heavy CPU-bound work — offload those to task queues (Celery, RQ)
• Always monitor logs and errors in background jobs

🧪 Practice Challenge¶

1. Create an async endpoint that fetches data from two APIs in parallel.
2. Add a background task that logs the response time of each request.
3. Test how quickly the endpoint responds compared to a synchronous version.

🧠 Recap¶

You now know how to:

• Write async endpoints for non-blocking I/O
• Schedule background jobs with FastAPI’s BackgroundTasks
• Combine async + background tasks for efficient workflows