Mastering Database Dependencies in FastAPI: A Deep Dive

FastAPI’s built‑in dependency injection system is one of its most praised features, especially when it comes to wiring up a database. By turning the database session into a dependency, you let FastAPI handle the lifecycle of connections, keep your endpoints clean, and make testing a breeze. In this article we’ll explore the why and how of FastAPI database dependencies, walk through a concrete async SQLAlchemy setup, and flag the common traps that catch newcomers.

Why Dependency Injection Matters for Your Database Layer

At first glance, opening a connection inside each endpoint seems harmless, but it quickly spirals into duplicated code, leaked connections, and tangled transaction logic. Dependency injection (DI) solves these problems by centralising the creation, configuration, and teardown of resources. With FastAPI, a single Depends call can provide a ready‑to‑use session to any route function, while the framework ensures the session is closed once the request finishes.

Beyond resource management, DI encourages a clear separation of concerns: route handlers stay focused on request/response handling, whereas the database logic lives in its own module. This separation also simplifies mocking the session during unit tests, because you can replace the dependency with a lightweight stub.

Setting Up an Async SQLAlchemy Engine

FastAPI works naturally with async libraries, and SQLAlchemy 1.4+ offers a solid async API. Below is a minimal configuration that you can drop into database.py:

• Engine creation: use create_async_engine with a URL like postgresql+asyncpg://user:pass@localhost/dbname.
• Session factory: sessionmaker with class_=AsyncSession and expire_on_commit=False keeps objects usable after a commit.
• Base model: declarative_base() remains unchanged; just inherit from it as usual.

Here’s the code snippet:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
Base = declarative_base()

With the engine ready, the next step is to expose a dependency that yields a session.

Creating a Dependency That Yields a Session

FastAPI expects a callable that either returns a value or yields it. Yielding lets you perform cleanup after the response is sent. A typical dependency looks like this:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()

Because the async with block already closes the session, the explicit await session.close() is optional but clarifies intent. Any route can now receive session: AsyncSession = Depends(get_db) and start issuing queries immediately.

Using the Dependency in Route Handlers

Let’s illustrate with a simple CRUD endpoint for a User model:

from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.future import select
router = APIRouter()
@router.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalars().first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user

The handler stays tiny: it declares the dependency, runs a query, and returns the model. No manual session.commit() or session.close() is required, because the dependency handles it automatically.

Transaction Management with Dependencies

Sometimes you need more control, for example when multiple writes must either all succeed or all roll back. One pattern is to create a separate dependency that starts a transaction, yields the session, and commits or rolls back based on whether an exception bubbled up:

async def get_db_transaction() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
async with session.begin():
yield session
# If an exception occurs, session.begin() rolls back automatically.

Endpoints that perform several writes can now depend on get_db_transaction, guaranteeing atomicity without cluttering the view logic.

Common Pitfalls and How to Avoid Them

• Mixing sync and async sessions: Passing a synchronous Session into an async endpoint can block the event loop. Always stick to AsyncSession when your routes are declared with async def.
• Forgotten await on queries: The async API returns awaitable objects. Forgetting await leads to confusing “coroutine not awaited” errors.
• Leaking connections in background tasks: Background tasks run after the request finishes, so they don’t inherit the request‑scoped dependency. Explicitly create a new session inside the task or pass a session that you manage yourself.
• Using the same session across concurrent requests: Because dependencies are called per‑request, each request gets its own session. Sharing a session object globally defeats the purpose and can cause race conditions.

Testing Your Dependency Without a Real Database

One of the biggest advantages of FastAPI’s DI is the ability to replace the real database with a mock. In your test suite, define a simple in‑memory SQLite engine and a dependency override:

from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine
from myapp.main import app, get_db
test_engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async def override_get_db():
async with AsyncSessionLocal(bind=test_engine) as session:
yield session
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
# Now you can call endpoints; they’ll hit the in‑memory DB.

This approach keeps tests fast, deterministic, and free from external side effects.

FAQ

Do I have to use SQLAlchemy with FastAPI?

No. FastAPI works with any async‑compatible ORM or raw driver. SQLAlchemy is popular because it offers a rich declarative model and integrates cleanly with FastAPI’s dependency system, but you can swap it for Tortoise‑ORM, GINO, or even a simple asyncpg connection pool.

Can I share a single session across multiple endpoints?

Generally you should not. Each request gets its own session to avoid cross‑request contamination. If you need to share state, store it in a cache or a separate service rather than reusing the session.

What happens if an exception is raised after yielding a session?

FastAPI’s dependency handling ensures that the finally block runs, so the session is closed even when the endpoint raises an error. When you wrap the session in session.begin(), the transaction is rolled back automatically on exceptions.

Is it possible to use the same dependency for both sync and async routes?

Yes, but you need two separate dependencies: one returning a synchronous Session and another yielding an AsyncSession. Mixing them can lead to subtle bugs, so keeping them distinct is the safest route.