How to Use Async Database in FastAPI: Simple Guide

To use an async database in FastAPI, use an async ORM like SQLAlchemy 2.0 with async support or Databases library. Define async session and use async def endpoints with await for database calls to keep your app responsive.

Syntax

Use async functions for your API endpoints and database operations. Create an async engine and session with SQLAlchemy's async API. Use await to call async database methods.

• async def: defines an async function
• AsyncSession: manages async DB sessions
• async with: ensures proper session closing
• await: waits for async DB calls to complete

python

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

async def get_user(user_id: int):
    async with async_session() as session:
        result = await session.execute(
            "SELECT * FROM users WHERE id = :id", {"id": user_id}
        )
        user = result.fetchone()
        return user

Example

This example shows a FastAPI app using async SQLAlchemy to fetch a user by ID from a SQLite database asynchronously.

python

from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

app = FastAPI()

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=False)
async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    async with async_session() as session:
        result = await session.execute(text("SELECT id, name FROM users WHERE id = :id"), {"id": user_id})
        user = result.fetchone()
        if user is None:
            raise HTTPException(status_code=404, detail="User not found")
        return {"id": user.id, "name": user.name}

Output

GET /users/1
Response: {"id": 1, "name": "Alice"}  (if user exists)

Common Pitfalls

• Not using async def for endpoints or DB functions causes blocking.
• Forgetting to await async DB calls leads to errors or no execution.
• Using synchronous DB drivers (like plain sqlite3) blocks the event loop.
• Not closing sessions properly can cause connection leaks; use async with for sessions.

python

from fastapi import FastAPI

app = FastAPI()

# Wrong: synchronous function blocks event loop
@app.get("/wrong")
def wrong_endpoint():
    # synchronous DB call here (not shown)
    return {"message": "This blocks"}

# Right: async function with await
@app.get("/right")
async def right_endpoint():
    # await async DB call here
    return {"message": "This is async and non-blocking"}

Quick Reference

Remember these key points when using async databases in FastAPI:

• Use async def for all async functions.
• Use async database drivers like aiosqlite or async SQLAlchemy.
• Always await async calls.
• Use async with to manage sessions safely.
• Test endpoints to ensure non-blocking behavior.

Key Takeaways

Use async database drivers and async SQLAlchemy for non-blocking DB access in FastAPI.

Define API endpoints with async functions and await database calls.

Manage database sessions with async context managers to avoid leaks.

Avoid synchronous DB calls to keep FastAPI responsive.

Test your async endpoints to confirm they work without blocking.