FastAPI Lifespan Events: Stop Using Startup/Shutdown Decorators Wrong

Everyone gets FastAPI startup/shutdown wrong. They slap @app.on_event("startup") on a function that opens a database pool, then wonder why their app crashes at 3am with 'Connection pool exhausted'. The old decorator pattern is a ticking bomb. FastAPI 0.93+ deprecated it for a reason: it doesn't compose, it doesn't handle exceptions cleanly, and it breaks under concurrent startup signals from multiple workers. The lifespan context manager fixes all of that. By the end of this article, you'll be able to write production-grade startup/shutdown code that survives SIGTERM, connection drops, and even your own deployment scripts.

Why the Old @app.on_event Decorators Are Dangerous

The old decorators look clean but hide a nasty flaw: they don't compose. If you have three startup handlers and the second one throws, the first one's resources never get cleaned up. In production, that means leaked connections, open file handles, and zombie threads. I've seen this bring down a payments service when the thread pool was exhausted at 3am because a startup handler crashed after opening a connection pool. The lifespan context manager wraps everything in a single async generator: setup before yield, teardown after. If setup fails, teardown still runs because Python's finally in the context manager triggers. This is not a minor improvement — it's the difference between a self-healing service and a pager storm.

# io.thecodeforge — Python tutorial

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def open_pool():
    # BUG: if this crashes, shutdown never runs
    app.state.db_pool = await create_db_pool()

@app.on_event("startup")
async def load_cache():
    # If this crashes, open_pool's pool is leaked
    raise RuntimeError("Cache unavailable")

@app.on_event("shutdown")
async def close_pool():
    # This never runs if load_cache crashed
    await app.state.db_pool.close()

The Lifespan Context Manager: One Pattern to Rule Them All

FastAPI's lifespan parameter accepts an async generator. Everything before yield is startup; everything after is shutdown. The generator is wrapped in a context manager, so even if startup throws, the cleanup runs. This is the only pattern you should use in production. Here's how a checkout service initializes its database pool, cache client, and external payment gateway client. Notice the try/finally — belt and suspenders.

# io.thecodeforge — Python tutorial

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg  # production-grade PostgreSQL driver

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: acquire resources
    print("Starting up...")
    db_pool = await asyncpg.create_pool(
        min_size=5, max_size=20,
        command_timeout=5  # seconds
    )
    app.state.db_pool = db_pool

    cache = await create_cache_client()
    app.state.cache = cache

    try:
        yield  # app runs here
    finally:
        # Shutdown: release resources — ALWAYS runs
        print("Shutting down...")
        await db_pool.close()
        await cache.close()

app = FastAPI(lifespan=lifespan)

@app.get("/health")
async def health():
    # Use app.state.db_pool safely
    async with app.state.db_pool.acquire() as conn:
        await conn.execute("SELECT 1")
    return {"status": "ok"}

Graceful Shutdown: Handling SIGTERM Without Dropping Requests

When Kubernetes sends SIGTERM, your app has a few seconds to finish in-flight requests and close resources. The lifespan shutdown runs after the server stops accepting new connections but before the process exits. But if your cleanup blocks (e.g., waiting for a database pool to drain), uvicorn will kill you after timeout_graceful_shutdown (default 30 seconds). The classic rookie mistake here is to await a long-running task in shutdown without a timeout. I've seen this cause a full table lock on writes because the cleanup waited indefinitely for a hung query. Always wrap cleanup in asyncio.wait_for with a timeout.

# io.thecodeforge — Python tutorial

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.work_queue = asyncio.Queue()
    app.state.workers = [
        asyncio.create_task(worker(i))
        for i in range(4)
    ]
    yield
    # Shutdown with timeout
    print("Shutting down workers...")
    for task in app.state.workers:
        task.cancel()
    # Wait for tasks to finish, but don't block forever
    await asyncio.wait(
        app.state.workers,
        timeout=5.0,  # seconds
        return_when=asyncio.ALL_COMPLETED
    )
    print("Workers shut down.")

Testing Lifespan Events: How to Verify Startup and Shutdown

You can't test lifespan events with a simple TestClient because it doesn't trigger the lifespan by default. You need TestClient with raise_server_exceptions=False and manually enter/exit the context manager. Or use httpx.AsyncClient with the app directly. Here's how we test the checkout service lifespan: we assert that the pool is created after startup and closed after shutdown. This catches the exact bug that burned us in the incident.

# io.thecodeforge — Python tutorial

import pytest
from httpx import AsyncClient, ASGITransport
from your_app import app  # your FastAPI app with lifespan

@pytest.mark.asyncio
async def test_lifespan():
    # Manually enter lifespan
    async with app.router.lifespan_context(app):
        # Startup should have created db_pool
        assert app.state.db_pool is not None
        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://test"
        ) as client:
            response = await client.get("/health")
            assert response.status_code == 200
    # After lifespan exits, pool should be closed
    assert app.state.db_pool._closed

@pytest.mark.asyncio
async def test_lifespan_startup_failure():
    # Simulate startup failure by patching
    with pytest.raises(RuntimeError):
        async with app.router.lifespan_context(app):
            raise RuntimeError("Boom")
    # Even after exception, cleanup should have run
    # In real test, check that resources are released

When Not to Use Lifespan: Simple Scripts and One-Off Tasks

If you're writing a tiny internal tool that runs one request and exits, don't bother with lifespan. Just open connections in the endpoint and close them in a finally. Lifespan adds complexity that's only justified when you have multiple endpoints sharing resources, or when you need graceful shutdown. For a cron-job style script that calls one API and dies, a simple with block is cleaner. But for any service that stays running and handles concurrent requests, lifespan is non-negotiable.

# io.thecodeforge — Python tutorial

# For a one-off script, lifespan is overkill
import httpx

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        print(response.json())

# No lifespan needed — client is scoped to the function

Advanced: Lifespan with Dependency Injection and Background Tasks

Sometimes you need to pass lifespan resources to background tasks or dependencies. The cleanest pattern is to store them in app.state and access them via request.app.state in dependencies. But beware: background tasks that outlive the request may try to access app.state after shutdown. Always cancel background tasks in the lifespan shutdown, as shown earlier. Another pattern is to use Depends() with a generator that yields a resource from app.state. This keeps your endpoints clean.

# io.thecodeforge — Python tutorial

from fastapi import FastAPI, Depends, Request
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = await create_pool()
    app.state.pool = pool
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

async def get_db(request: Request):
    # Access pool from app.state
    async with request.app.state.pool.acquire() as conn:
        yield conn

@app.get("/items")
async def read_items(db=Depends(get_db)):
    # db is a connection from the pool
    result = await db.fetch("SELECT * FROM items")
    return result