Async database queries in FastAPI

Introduction

Async database queries let your app handle many tasks at once without waiting for the database to finish. This makes your app faster and smoother.

When your app needs to handle many users at the same time.

When you want to keep the app responsive while waiting for data from the database.

When you build APIs that fetch or save data frequently.

When you want to improve performance by not blocking other tasks during database access.

Syntax

FastAPI

async def function_name():
    async with database_session() as session:
        result = await session.execute(query)
        data = result.scalars().all()
        return data

Use async def to define asynchronous functions.

Use await before database calls to wait for their results without blocking.

Examples

This example fetches all users asynchronously using an async session.

FastAPI

from sqlalchemy.ext.asyncio import AsyncSession

async def get_users(session: AsyncSession):
    result = await session.execute("SELECT * FROM users")
    users = result.scalars().all()
    return users

This example adds a new user and commits the change asynchronously.

FastAPI

async def add_user(session: AsyncSession, name: str):
    new_user = User(name=name)
    session.add(new_user)
    await session.commit()

Sample Program

This FastAPI app uses async database queries to add and get users without blocking. It creates the database on startup, adds users with POST, and lists users with GET.

FastAPI

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select, Integer, String, Column
from sqlalchemy.orm import declarative_base

app = FastAPI()

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

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/users/")
async def create_user(name: str, session: AsyncSession = Depends(get_session)):
    new_user = User(name=name)
    session.add(new_user)
    await session.commit()
    await session.refresh(new_user)
    return {"id": new_user.id, "name": new_user.name}

@app.get("/users/")
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

Important Notes

Always use async with or async_sessionmaker to manage database sessions safely.

Remember to use await before database calls to avoid blocking the app.

SQLite with aiosqlite is good for testing, but for production use a more robust database like PostgreSQL.

Summary

Async queries let your app do other work while waiting for the database.

Use async def and await with async database sessions.

FastAPI works well with async database calls for fast, responsive APIs.