Connection lifecycle management in FastAPI

Introduction

Connection lifecycle management helps you open and close resources like database connections safely during a web request. It keeps your app fast and avoids errors.

When you need to connect to a database for each web request.

When you want to open a file or external service only while handling a request.

When you want to clean up resources automatically after a request finishes.

When you want to avoid keeping connections open longer than needed.

When you want to share a connection setup across multiple routes safely.

Syntax

FastAPI

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

@asynccontextmanager
def lifespan(app: FastAPI):
    # setup code here
    yield
    # cleanup code here

app = FastAPI(lifespan=lifespan)

# Or use dependencies with yield
async def get_db():
    db = connect_to_db()
    try:
        yield db
    finally:
        db.close()

Use @asynccontextmanager to define setup and cleanup around app lifespan.

Use yield in dependencies to manage resource opening and closing per request.

Examples

This runs code when the app starts and stops.

FastAPI

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
def lifespan(app: FastAPI):
    print('Starting app')
    yield
    print('Stopping app')

app = FastAPI(lifespan=lifespan)

This opens a DB connection for each request and closes it after.

FastAPI

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_db():
    db = open_db_connection()
    try:
        yield db
    finally:
        db.close()

@app.get('/')
async def read_root(db=Depends(get_db)):
    return {'message': 'Using DB'}

Sample Program

This example shows a fake database connection opening when a request starts and closing after it ends.

FastAPI

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

class FakeDB:
    def __init__(self):
        print('DB connected')
    def close(self):
        print('DB closed')

async def get_db():
    db = FakeDB()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.get('/')
async def read_root(db=Depends(get_db)):
    return {'message': 'Hello with DB'}

Important Notes

Always close connections to avoid resource leaks.

Use yield in dependencies to run cleanup code after the request.

FastAPI runs lifespan code only once when the app starts and stops.

Summary

Connection lifecycle management opens and closes resources safely.

Use yield in dependencies for per-request setup and cleanup.

Use @asynccontextmanager for app startup and shutdown tasks.