Async Database - fast data access

Welcome! Darwin here with asynchronous databases in FastAPI!

So far we have used an in-memory dict. It is handy for a start, but it has a serious flaw: every server restart wipes out all observations, as if someone tore the pages out of the expedition logbook. Now we will connect a real PostgreSQL database with async SQLAlchemy!

Safari analogy: An async database is like a real-time animal tracking system - thousands of observations recorded at the same time without blocking! A tracker who has sent a radio report does not stand still waiting for confirmation from camp. Meanwhile they watch the next herd, and pick up the reply when it arrives.

Installation

We need two packages: the SQLAlchemy library itself with its asyncio extra, and a driver that talks to PostgreSQL without blocking the event loop:

1pip install sqlalchemy[asyncio] asyncpg

• SQLAlchemy - ORM
• asyncpg - async PostgreSQL driver

An ORM (Object-Relational Mapper) translates Python classes into tables and objects into rows, so instead of writing SQL by hand you work with ordinary objects. In the zsh shell square brackets have a special meaning, so there you should put the name in quotes: "sqlalchemy[asyncio]".

Configuring Async SQLAlchemy

We keep the whole configuration in a single database.py file. First the database URL and the engine, the object that manages connections. The create_async_engine function creates its asynchronous version:

1# database.py
2from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
3from sqlalchemy.orm import DeclarativeBase
4
5DATABASE_URL = "postgresql+asyncpg://user:password@localhost/safari_db"
6
7# Async engine
8engine = create_async_engine(DATABASE_URL, echo=True)

The postgresql+asyncpg prefix in the URL tells SQLAlchemy which driver to use. The echo=True option prints every SQL query to the console - great while learning, but switch it off in production. Creating the engine does not connect to the database yet, the connection is opened on the first query.

The engine is the radio line, but to do work we need a session - a conversation with the database in which we collect changes and commit them together. async_sessionmaker is a factory for such sessions:

1# Session factory
2AsyncSessionLocal = async_sessionmaker(
3    engine,
4    class_=AsyncSession,
5    expire_on_commit=False
6)

AsyncSession is the default class here, so you can omit class_, although spelling it out does no harm. More important is expire_on_commit=False: without it, objects lose their loaded values after commit(), and reading them in async code ends in an error, because SQLAlchemy would try to reload the data from the database implicitly.

What remains is the base class for models and a dependency that hands a session to every request:

1# Base class
2class Base(DeclarativeBase):
3    pass
4
5# Dependency
6async def get_db():
7    async with AsyncSessionLocal() as session:
8        yield session

DeclarativeBase is the class all tables inherit from. The get_db function uses yield, so FastAPI passes the session to the endpoint, and once the response has been sent, async with closes it on its own - even if the endpoint raised an exception.

SQLAlchemy model

A model describes a table. Mapped[int] is the type annotation for a column, and mapped_column supplies the details: the SQL type, the primary key, the default value:

1# models.py
2from sqlalchemy import Integer, String, Boolean
3from sqlalchemy.orm import Mapped, mapped_column
4from database import Base
5
6class SpeciesModel(Base):
7    __tablename__ = "species"
8
9    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
10    name: Mapped[str] = mapped_column(String(100), nullable=False)
11    scientific_name: Mapped[str] = mapped_column(String(150), nullable=False)
12    population: Mapped[int] = mapped_column(Integer, default=0)
13    endangered: Mapped[bool] = mapped_column(Boolean, default=False)

The table name in the database comes from __tablename__. Note that this is not the Pydantic model from the previous lesson - SQLAlchemy describes how data sits in the database, while Pydantic describes what it looks like in requests and responses. In a real project you therefore have both.

CRUD Operations - Async

CRUD stands for the four basic operations: Create, Read, Update, Delete. We build every query with the select function and run it with await db.execute(...). Let's start with reading:

1# crud.py
2from sqlalchemy.ext.asyncio import AsyncSession
3from sqlalchemy import select
4from models import SpeciesModel
5from schemas import SpeciesCreate
6
7async def get_species(db: AsyncSession, species_id: int):
8    result = await db.execute(select(SpeciesModel).where(SpeciesModel.id == species_id))
9    return result.scalar_one_or_none()
10
11async def get_all_species(db: AsyncSession, skip: int = 0, limit: int = 10):
12    result = await db.execute(select(SpeciesModel).offset(skip).limit(limit))
13    return result.scalars().all()

scalar_one_or_none() returns one object, or None when nothing matches, and scalars().all() returns a list of objects. The offset and limit parameters implement pagination, so you do not download the whole reserve at once.

Now writing. We turn the Pydantic model into a dict and unpack it into the constructor of the SQLAlchemy model:

1async def create_species(db: AsyncSession, species: SpeciesCreate):
2    db_species = SpeciesModel(**species.model_dump())
3    db.add(db_species)
4    await db.commit()
5    await db.refresh(db_species)
6    return db_species

db.add() only adds the object to the session - it reaches the database only on await db.commit(). refresh loads the values assigned by the database, such as the generated id. In older tutorials you will see species.dict(), but in Pydantic 2 the correct name is model_dump().

Deleting reuses the read function we already have:

1async def delete_species(db: AsyncSession, species_id: int):
2    species = await get_species(db, species_id)
3    if species:
4        await db.delete(species)
5        await db.commit()
6    return species

In AsyncSession the delete method is a coroutine, so it needs await, just like commit.

FastAPI Endpoints with Async DB

Finally we bring everything together in the endpoints. Depends(get_db) is dependency injection: FastAPI calls get_db by itself and passes the session as the db parameter:

1# main.py
2from fastapi import FastAPI, Depends, HTTPException
3from sqlalchemy.ext.asyncio import AsyncSession
4from database import get_db
5import crud
6import schemas
7
8app = FastAPI()
9
10@app.get("/species", response_model=list[schemas.Species])
11async def list_species(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
12    species = await crud.get_all_species(db, skip=skip, limit=limit)
13    return species

The endpoint neither creates the session nor closes it - the dependency takes care of that. The other two endpoints look the same way:

1@app.get("/species/{species_id}", response_model=schemas.Species)
2async def get_species(species_id: int, db: AsyncSession = Depends(get_db)):
3    species = await crud.get_species(db, species_id)
4    if not species:
5        raise HTTPException(status_code=404, detail="Species not found")
6    return species
7
8@app.post("/species", response_model=schemas.Species)
9async def create_species(species: schemas.SpeciesCreate, db: AsyncSession = Depends(get_db)):
10    return await crud.create_species(db, species)

For response_model to accept a SQLAlchemy object instead of a dict, the Pydantic model needs from_attributes=True, which you know from the Pydantic lesson. A missing species becomes a readable 404 code.

Advantages of an async database:

• Thousands of requests at the same time
• Does not block the event loop
• High performance

To be fair, async does not make a single query faster. The gain appears when many requests wait for the database at the same time - the server handles others instead of sitting idle. My advice: when you write async def endpoints, use an asynchronous driver too, because an ordinary blocking query inside async def stops the whole event loop.

Next lesson: JWT Authentication! Depends will come in handy there, because FastAPI injects the logged-in user the same way.

Remember: the database is the expedition logbook, a session is one entry, and commit is the signature without which the entry does not count.