Mastering Database Connections in FastAPI: Best Practices and Code Samples

When building APIs with FastAPI, database management is a key component. Efficiently handling database connections ensures your application is robust, scalable, and avoids common issues such as connection leaks or performance bottlenecks. This article introduces best practices for managing database connections in FastAPI.

Why proper database connection management is needed?

Databases are resource‑intensive systems. Improper handling can cause:

Connection leaks : Unclosed connections exhaust the database connection pool.

Performance issues : Frequently reopening connections slows the application.

State inconsistency : Unclosed connections may retain stale or incomplete data.

Properly managing connections maintains performance and ensures data integrity.

Configuring FastAPI with a database

FastAPI does not include built‑in database integration, but works seamlessly with ORM tools such as SQLAlchemy, Tortoise‑ORM, and databases like PostgreSQL, MySQL, or SQLite. This tutorial uses SQLAlchemy.

Install dependencies

<code>pip install sqlalchemy psycopg2</code>

Database configuration

Create a database.py file to handle connection logic.

<code>from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# Replace with your database URL
DATABASE_URL = "postgresql://user:password@localhost/mydatabase"

# Create engine
engine = create_engine(DATABASE_URL)

# Create session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for ORM models
Base = declarative_base()</code>

The parts serve the following purposes:

engine : Manages the connection to the database.

SessionLocal : Provides sessions for interacting with the database.

Base : Base class for ORM models.

Using dependency injection to manage database sessions

FastAPI’s dependency injection system is ideal for managing database sessions. Create a dependency that provides and closes a database session.

<code>from fastapi import Depends
from sqlalchemy.orm import Session
from .database import SessionLocal

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()</code>

The logic of the code:

yield db : Provides a session to routes.

db.close() : Ensures the session is closed after use, even on exceptions.

Using the database in routes

Integrate the get_db dependency into route handlers to access a database session.

<code>from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from .database import Base, engine, get_db

# Create all tables
Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    items = db.execute("SELECT * FROM items LIMIT :limit OFFSET :skip", {"limit": limit, "skip": skip}).fetchall()
    return {"items": items}</code>

Key points:

Depends(get_db) : Injects the database session into the route.

Automatic closing : The session is automatically closed after the route completes.

Handling connections in asynchronous code

For async frameworks like Tortoise‑ORM or async SQLAlchemy, adapt to async connections. Below is an example using async SQLAlchemy.

Install async libraries

<code>pip install sqlalchemy[asyncio] asyncpg</code>

Update database.py to support async

<code>from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydatabase"

# Async engine
async_engine = create_async_engine(DATABASE_URL, echo=True)

# Async session factory
AsyncSessionLocal = sessionmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False)

# Base class for ORM models
Base = declarative_base()</code>

Update async session dependency

<code>async def get_db():
    async with AsyncSessionLocal() as session:
        yield session</code>

Update async route

<code>@app.get("/async-items/")
async def read_items(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
    result = await db.execute("SELECT * FROM items LIMIT :limit OFFSET :skip", {"limit": limit, "skip": skip})
    items = result.fetchall()
    return {"items": items}</code>

Test database connection

Testing routes that interact with the database requires proper setup:

Test database : Use a separate database for testing.

Session override : Override get_db to use a mock database.

Example:

<code>from fastapi.testclient import TestClient
from .main import app
from .database import SessionLocal

# Override dependency
def override_get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200
    assert "items" in response.json()</code>

Summary

When managing database connections in FastAPI, follow these best practices:

Use a centralized database.py file for configuration.

Leverage dependency injection for session management.

Ensure sessions are properly closed to prevent leaks.

For asynchronous applications, use async configuration.

Implementing these techniques lets you build efficient, scalable, and reliable FastAPI applications.