Lesson 15 · FastAPI deep dive

FastAPI Tutorial #15: Async & Background Tasks

Video fifteen of the eighteen-part series: async def handlers and work that runs after the response. When to use async def, why blocking calls inside it are…

⬇ Download notebook

CourseFastAPI deep dive
Lesson15 of 18
FormatJupyter notebook · 10 code cells

FastAPIasyncio

What you'll learn

def vs async def - Both Genuinely Work
async def with await - the Actual Point of Async
The Trap - Blocking Code Inside async def
The Fix - Use def, or a Genuinely Async Library
BackgroundTasks - Work That Runs After the Response
Passing Data Into a Background Task
Multiple Background Tasks Run in Order
BackgroundTasks Via a Dependency

Data

No separate download needed — the notebook creates or downloads everything it uses.

📓 Full notebook

Download .ipynb

FastAPI Deep-Dive, Video 15: Async Endpoints and Background Tasks#

Video fifteen of the eighteen-part series: async def handlers and work that runs after the response.
When to use async def, why blocking calls inside it are dangerous, and BackgroundTasks.
Let's get into it.

Part 1: def vs async def - Both Genuinely Work#

from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get('/sync-hello')
def sync_hello():
    return {'style': 'sync'}
@app.get('/async-hello')
async def async_hello():
    return {'style': 'async'}
client = TestClient(app)
print(client.get('/sync-hello').json())
print(client.get('/async-hello').json())

{'style': 'sync'}
{'style': 'async'}

Part 2: async def with await - the Actual Point of Async#

import asyncio
@app.get('/fetch-remote')
async def fetch_remote():
    await asyncio.sleep(0.01)
    return {'data': 'fetched'}
print(client.get('/fetch-remote').json())

{'data': 'fetched'}

Part 3: The Trap - Blocking Code Inside async def#

import time
@app.get('/blocking-mistake')
async def blocking_mistake():
    time.sleep(0.01)
    return {'warning': 'this genuinely blocks the whole event loop'}
print(client.get('/blocking-mistake').json())

{'warning': 'this genuinely blocks the whole event loop'}

Part 4: The Fix - Use def, or a Genuinely Async Library#

@app.get('/blocking-fixed')
def blocking_fixed():
    time.sleep(0.01)
    return {'note': 'this genuinely runs in a thread pool instead, never blocking the loop'}
print(client.get('/blocking-fixed').json())

{'note': 'this genuinely runs in a thread pool instead, never blocking the loop'}

Part 5: BackgroundTasks - Work That Runs After the Response#

from fastapi import BackgroundTasks
activity_log = []
def write_log(message: str):
    activity_log.append(message)
@app.post('/notify')
def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, 'notification sent')
    return {'status': 'queued'}
print(client.post('/notify').json())
print(activity_log)

{'status': 'queued'}
['notification sent']

Part 6: Passing Data Into a Background Task#

def send_welcome_email(user_id: int, email: str):
    activity_log.append(f'welcome email queued for user {user_id} at {email}')
@app.post('/signup')
def signup(email: str, background_tasks: BackgroundTasks):
    new_id = 42
    background_tasks.add_task(send_welcome_email, new_id, email)
    return {'id': new_id}
print(client.post('/signup', params={'email': 'alex@example.com'}).json())
print(activity_log[-1])

{'id': 42}
welcome email queued for user 42 at alex@example.com

Part 7: Multiple Background Tasks Run in Order#

@app.post('/checkout')
def checkout(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, 'charge processed')
    background_tasks.add_task(write_log, 'receipt emailed')
    background_tasks.add_task(write_log, 'inventory updated')
    return {'status': 'ok'}
activity_log.clear()
print(client.post('/checkout').json())
print(activity_log)

{'status': 'ok'}
['charge processed', 'receipt emailed', 'inventory updated']

Part 8: BackgroundTasks Via a Dependency#

from fastapi import Depends
def audit_dependency(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, 'audit trail recorded')
    return True
@app.post('/secure-action')
def secure_action(audited: bool = Depends(audit_dependency), background_tasks: BackgroundTasks = None):
    background_tasks.add_task(write_log, 'secure action completed')
    return {'audited': audited}
activity_log.clear()
print(client.post('/secure-action').json())
print(activity_log)

{'audited': True}
['audit trail recorded', 'secure action completed']

Part 9: An Async Dependency#

async def get_request_context():
    await asyncio.sleep(0.001)
    return {'traced': True}
@app.get('/traced')
async def traced_endpoint(context: dict = Depends(get_request_context)):
    return context
print(client.get('/traced').json())

{'traced': True}

Part 10: A Real Pattern - Save Data, Then Background-Notify#

orders_app = FastAPI()
ORDERS = {}
order_notifications = []
def send_order_confirmation(order_id: int, email: str):
    order_notifications.append(f'confirmation sent to {email} for order {order_id}')
@orders_app.post('/orders')
async def place_order(email: str, background_tasks: BackgroundTasks):
    await asyncio.sleep(0.001)
    order_id = len(ORDERS) + 1
    ORDERS[order_id] = {'id': order_id, 'email': email}
    background_tasks.add_task(send_order_confirmation, order_id, email)
    return {'order_id': order_id, 'status': 'placed'}
orders_client = TestClient(orders_app)
response = orders_client.post('/orders', params={'email': 'sam@example.com'})
print(response.json())
print(order_notifications)
print(ORDERS)

{'order_id': 1, 'status': 'placed'}
['confirmation sent to sam@example.com for order 1']
{1: {'id': 1, 'email': 'sam@example.com'}}

Wrap-Up: What You Learned#

Both def and async def handlers work; FastAPI runs def handlers in a thread pool automatically.
async def only pays off when the handler actually awaits something, like real I/O.
A blocking call like time.sleep inside async def freezes the whole event loop for everyone.
The fix for blocking code is often just using def instead, which runs safely in a thread.
BackgroundTasks queues work that runs only after the response has already been sent.
add_task accepts positional and keyword arguments, passing handler data straight through.
Multiple queued background tasks run in the exact order they were added.
A dependency can accept BackgroundTasks too, queuing its own work independently.
Dependencies can be async def as well, and compose freely with async handlers.
That wraps up async endpoints and background tasks. Next up: WebSockets.

Found this useful?

All lessons, notebooks and datasets here are free. If they helped you, a coffee keeps new lessons coming.

☕ Buy me a coffee Other ways to help