60.7 Background Tasks and Lifespan Events
Right, let’s talk about the stuff that happens around your request. You’re not just building a fancy request-response vending machine. A real application needs to do work after it’s sent a response, or needs to set up and tear down expensive resources gracefully. This is where FastAPI’s background tasks and lifespan events come in, and they are two of the most elegantly designed features in the framework. They solve different problems, but both with a refreshing lack of ceremony.
When to Use Background Tasks (And When Not To)
Imagine a user signs up on your site. You don’t want them staring at a loading spinner while you send a welcome email, resize their profile picture, and update a CRM. You send them a “Welcome!” response immediately and queue up that other work to happen after you’ve told their browser everything is done.
That’s a background task. The key insight is this: the task is directly tied to a specific request, but its completion is not required for the response.
Here’s the beautiful part: you can just declare them as a parameter in your path operation function. It’s almost suspiciously simple.
from fastapi import BackgroundTasks, FastAPI
from .email import send_welcome_email # your fake email function
app = FastAPI()
def log_registration(username: str):
# Imagine this does some slow database logging
print(f"User {username} registered successfully!")
@app.post("/register/{username}")
async def register_user(username: str, background_tasks: BackgroundTasks):
# Add the function you want to run and the arguments to pass to it
background_tasks.add_task(log_registration, username)
background_tasks.add_task(send_welcome_email, username)
return {"message": "User registered. Check your email soon!"}
FastAPI injects the background_tasks object for you. You call add_task, passing it the function and its arguments. The framework then takes care of running them once the response is sent. It’s a fire-and-forget mechanism, which is both its greatest strength and its primary weakness.
The Pitfall: Background tasks are great for “best effort” work. But they are in-process and ephemeral. If your server process crashes or is restarted right after sending the response, those tasks are gone forever. They never ran. For mission-critical work (like taking payment or fulfilling an order), you need a proper, persistent task queue like Celery or RQ. Use background tasks for things where failure is annoying but not catastrophic—logging, sending non-critical notifications, triggering non-essential analytics.
Managing Your Application’s Lifespan
While background tasks are per-request, lifespan events are per-application. They let you run code before your application starts taking requests and after it has stopped. This is your chance to be a responsible adult and manage expensive resources like database connection pools, HTTP client sessions, or in-memory caches.
The design is wonderfully modern: it uses an async context manager. You define an async function with yield, and everything before the yield runs on startup, everything after runs on shutdown.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .database import create_connection_pool, close_connection_pool
# This is the lifespan function
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: Initialize the connection pool
print("Starting up... creating resources.")
app.state.db_pool = await create_connection_pool()
yield # This is where the application runs live
# Shutdown: Close the pool cleanly
print("Shutting down... cleaning up resources.")
await close_connection_pool(app.state.db_pool)
# Pass the lifespan to your FastAPI app
app = FastAPI(lifespan=lifespan)
@app.get("/data")
async def get_data():
# Now any endpoint can access the pool stored on app.state
data = await app.state.db_pool.fetch("SELECT * FROM table")
return data
Why this is brilliant: It’s explicit, async-first, and keeps your resource management neatly contained. The app.state attribute is your best friend here. It’s a dedicated place to hang these application-global objects so every endpoint can access them. No more flailing around with global variables.
The Critical Edge Case: Shutdown. When your application receives a shutdown signal (e.g., Ctrl+C or a process manager like Kubernetes sends a SIGTERM), FastAPI stops accepting new requests. It then waits for any ongoing requests to finish before running the shutdown part of your lifespan. This is crucial for data integrity. You won’t yank the database connection pool out from under a request that’s still using it. The shutdown code only runs once all in-flight requests are complete. It’s the polite way to exit.
The Gotcha: Blocking the Event Loop
Here’s the trap everyone falls into, and it’s the most important thing to remember: Your background tasks and lifespan events run in the same process, on the same event loop, as your main application.
If you do this, you are committing a cardinal sin:
def some_background_task():
# This is a CPU-intensive calculation or a blocking call
time.sleep(10) # 🚨 NEVER DO THIS IN AN ASYNC APP 🚨
print("I just blocked the entire server for 10 seconds.")
time.sleep(10) is a blocking operation. It halts the entire Python thread. In an async framework, this means all other requests and tasks are frozen solid for ten seconds. The solution is simple: don’t call blocking code. If you must, run it in a thread pool using asyncio.run_in_executor.
import asyncio
def some_sync_blocking_thing():
time.sleep(10)
return "Done"
async def some_background_task(background_tasks: BackgroundTasks):
# Offload the blocking function to a thread pool
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, some_sync_blocking_thing)
print(result)
The lifespan manager, being async, is less prone to this mistake, but the same rule applies. Always use non-blocking, async libraries for your startup/shutdown logic (e.g., asyncpg for databases, httpx.AsyncClient for HTTP) to keep your application nimble from birth to death.