FastAPI Async Database with asyncpg SQLAlchemy

Modern web apps demand speed. FastAPI excels here. It supports async operations natively. This includes database interactions. This guide shows you how.

We will combine asyncpg and SQLAlchemy. This duo handles PostgreSQL asynchronously. Your app will be fast and scalable.

Why Async Database Operations?

Traditional database calls are blocking. Your server waits for a response. It cannot handle other requests meanwhile.

Async operations are non-blocking. The server can process other tasks. It waits for the database in the background.

This improves concurrency dramatically. It is key for high-performance applications. FastAPI's async support makes this easy.

Project Setup and Dependencies

First, create a virtual environment. Then install the required packages.

pip install fastapi uvicorn sqlalchemy asyncpg databases

We use the databases package. It provides async support for SQLAlchemy core. It uses asyncpg under the hood.

Configuring the Async Database Engine

Create a database configuration module. This sets up the connection pool.

# database.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, MetaData
from databases import Database

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

# Create database instance
database = Database(DATABASE_URL)

# SQLAlchemy setup
metadata = MetaData()
Base = declarative_base(metadata=metadata)

# Define a sample model
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

The Database class manages connections. The URL uses the `asyncpg` driver. This is crucial for async operations.

Integrating with FastAPI Application

Next, integrate the database with your FastAPI app. Use lifespan events to manage connections.

# main.py
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
from database import database, User
from sqlalchemy import select

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: connect to database
    await database.connect()
    yield
    # Shutdown: disconnect from database
    await database.disconnect()

app = FastAPI(lifespan=lifespan)

@app.get("/users/")
async def get_users():
    query = select(User)
    users = await database.fetch_all(query)
    return users

@app.post("/users/")
async def create_user(username: str, email: str):
    query = User.__table__.insert().values(username=username, email=email)
    last_record_id = await database.execute(query)
    return {"id": last_record_id}

The lifespan manager handles connection lifecycle. The fetch_all and execute methods are async.

Performing CRUD Operations

Let's expand on basic CRUD. We'll add more complex examples.

# More CRUD examples
from sqlalchemy import update, delete

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = select(User).where(User.id == user_id)
    user = await database.fetch_one(query)
    if user is None:
        return {"error": "User not found"}
    return user

@app.put("/users/{user_id}")
async def update_user(user_id: int, username: str = None, email: str = None):
    # Build update dynamically
    update_data = {}
    if username: update_data["username"] = username
    if email: update_data["email"] = email

    query = update(User).where(User.id == user_id).values(**update_data)
    await database.execute(query)
    return {"message": "User updated"}

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    query = delete(User).where(User.id == user_id)
    await database.execute(query)
    return {"message": "User deleted"}

These endpoints show full async CRUD. The fetch_one method gets a single record.

Handling Transactions

Transactions ensure data integrity. Use them for multiple related operations.

from databases import Transaction

@app.post("/users/transfer")
async def transfer_points(sender_id: int, receiver_id: int, points: int):
    async with database.transaction() as transaction:
        # Deduct from sender
        query_sender = update(User).where(User.id == sender_id).values(points=User.points - points)
        await database.execute(query_sender)
        # Add to receiver
        query_receiver = update(User).where(User.id == receiver_id).values(points=User.points + points)
        await database.execute(query_receiver)
    return {"message": "Transfer successful"}

The database.transaction() context manager wraps operations. If any fail, all changes are rolled back.

Error Handling and Best Practices

Always handle database errors gracefully. Use try-except blocks.

import asyncpg
from fastapi import HTTPException

@app.post("/users/safe")
async def create_user_safe(username: str, email: str):
    try:
        query = User.__table__.insert().values(username=username, email=email)
        last_id = await database.execute(query)
        return {"id": last_id}
    except asyncpg.exceptions.UniqueViolationError:
        raise HTTPException(status_code=400, detail="Username or email already exists")
    except Exception as e:
        raise HTTPException(status_code=500, detail="Database error occurred")

Catching UniqueViolationError is important. It handles duplicate entries. For broader performance optimization, review our dedicated guide.

Use connection pooling effectively. The `databases` package handles this. Configure pool size in the DATABASE_URL.

Testing Async Endpoints

Testing is crucial. Use pytest with an async test client. For a deep dive, see our guide on Test FastAPI Endpoints with pytest and TestClient.

# test_main.py (example snippet)
import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/users/", json={"username": "test", "email": "[email protected]"})
    assert response.status_code == 200
    data = response.json()
    assert "id" in data

Scaling and Advanced Patterns

For complex apps, consider repository or service patterns. This separates database logic.

Use alembic for managing schema changes. Our article on FastAPI Database Migrations with Alembic covers this perfectly.

For microservices architecture, containerization is key. Learn more in FastAPI Microservices with Docker Compose.

Conclusion

Async database ops unlock FastAPI's full potential. Combining asyncpg and SQLAlchemy is powerful.

You get non-blocking I/O, high concurrency, and clean code. Remember to handle errors and use transactions.

Start integrating async databases today. Your application's performance will thank you.