SQLAlchemy 2.x and AsyncSession for FastAPI APIs (Part 1)

SQLAlchemy is the most widely used ORM (Object Relational Mapper) in the Python ecosystem and is the de facto choice for FastAPI applications.

With the introduction of SQLAlchemy 2.x, the library adopted a cleaner, more explicit API while providing first-class support for asynchronous database programming through AsyncSession.

In this article, we will build a complete understanding of SQLAlchemy 2.x and AsyncSession by creating a data access layer that integrates seamlessly with FastAPI.

SQLAlchemy is responsible for bridging the gap between Python objects and relational database tables, allowing developers to work with Python models instead of writing SQL for every operation.

The Problem SQLAlchemy Solves

Suppose we want to retrieve all users from a database. Without an ORM, we might write something like:

connection.execute(
    "SELECT * FROM users"
)

As the application grows, every endpoint must create database connections, execute SQL statements, convert database rows into Python objects, manage transactions, and close connections.

This repetitive code quickly becomes difficult to maintain.

Instead of working directly with SQL, it is preferable to work with Python objects. Instead of writing:

SELECT *
FROM users
WHERE id = 1;

we would rather write:

user = await session.get(User, 1)

SQLAlchemy performs the conversion between Python objects and SQL automatically.

SQLAlchemy Architecture

Although FastAPI endpoints typically interact only with AsyncSession, several components work behind the scenes to execute every database operation.

When a request reaches a FastAPI endpoint, the framework resolves the get_db() dependency and injects an AsyncSession into the endpoint.

The endpoint performs ORM operations using this session without worrying about how database connections are created or managed.

The AsyncSession represents a unit of work. It tracks ORM objects, manages transactions, executes SQL statements, and coordinates communication with the database.

Whenever the session needs to execute a query, it requests a database connection from the Engine.

The Engine is the core of SQLAlchemy. It manages database communication, understands the database dialect (PostgreSQL, MySQL, SQLite, etc.), and maintains the application's connection pool.

Instead of creating a new database connection for every request, the Engine reuses existing connections by borrowing one from the pool whenever possible.

The Connection Pool maintains a collection of reusable database connections.

Once a query finishes executing, the connection is returned to the pool instead of being closed, making it immediately available for future requests.

Installing SQLAlchemy

Install SQLAlchemy along with the asynchronous PostgreSQL driver.

pip install sqlalchemy asyncpg

For SQLite during development:

pip install aiosqlite

Every SQLAlchemy application starts by creating an Engine.

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = (
    "postgresql+asyncpg://user:password@localhost/demo"
)

engine = create_async_engine(
    DATABASE_URL,
    echo=True
)

The Engine does not immediately connect to the database. Instead, it stores the configuration required to create connections whenever they are needed.

Setting echo=True instructs SQLAlchemy to log every SQL statement, which is useful during development.

Creating AsyncSession

Applications usually do not instantiate AsyncSession directly. Instead, SQLAlchemy provides async_sessionmaker to create and configure AsyncSession instances.

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker
)

SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

This factory creates properly configured AsyncSession objects whenever a request requires one.

Creating sessions through async_sessionmaker ensures that every request receives its own isolated AsyncSession.

Creating Your First Model

Suppose our application manages users. A model might look like this.

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(
        primary_key=True
    )

    name: Mapped[str]
    email: Mapped[str]

Unlike a Pydantic model, this class represents a database table. Each instance corresponds to a single row inside the database.

Creating Database Tables

Once the models have been defined, SQLAlchemy can create the database schema.

async with engine.begin() as conn:
    await conn.run_sync(
        Base.metadata.create_all
    )

Although production systems typically use Alembic for migrations, understanding how SQLAlchemy creates tables is useful during development.

Injecting AsyncSession into FastAPI

One of the biggest advantages of FastAPI is its Dependency Injection system. We can inject an AsyncSession into every endpoint.

async def get_db():
    async with SessionLocal() as session:
        yield session

Endpoint:

from fastapi import Depends

@app.get("/users")
async def get_users(
    db: AsyncSession = Depends(get_db)
):
    ...

When a request arrives, FastAPI creates an AsyncSession using SessionLocal, injects it into the endpoint, and automatically closes the session after the response has been sent.

The endpoint never needs to worry about opening or closing database connections.

CRUD Operations

Now that we have an AsyncSession injected into our endpoint, we can perform the four basic CRUD operations.

Creating Records

Let's create a new user.

@app.post("/users")
async def create_user(
    db: AsyncSession = Depends(get_db)
):
    user = User(
        name="John",
        email="john@example.com"
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

The sequence is straightforward. First, a new User object is created. The object is added to the session using add().

Calling commit() persists the changes to the database, while refresh() reloads the object so that database-generated values, such as the primary key, become available.

Reading Records

Fetching data uses the SQLAlchemy 2.x select() API.

from sqlalchemy import select

@app.get("/users")
async def get_users(
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(User)
    )
    users = result.scalars().all()
    return users

Instead of returning raw database rows, SQLAlchemy converts the results into fully populated User objects.

Updating Records

Updating a record begins by retrieving the existing entity from the database.

After modifying one or more attributes, committing the transaction automatically generates the appropriate UPDATE statement.

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    user = await db.get(User, user_id)
    if user is None:
        return {"message": "User not found"}

    user.name = "Alice"
    user.email = "alice@example.com"

    await db.commit()
    await db.refresh(user)

    return user

SQLAlchemy automatically tracks changes made to ORM objects. When commit() is called, it detects the modified fields and generates the appropriate UPDATE statement.

Deleting Records

Deleting follows a similar pattern. The entity is first loaded into the session and then marked for deletion.

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    user = await db.get(User, user_id)
    if user is None:
        return {"message": "User not found"}

    await db.delete(user)
    await db.commit()
    return {
        "message": "User deleted successfully"
    }

When delete() is called, SQLAlchemy marks the entity for deletion.

The corresponding DELETE statement is executed only after the transaction is committed. Once the commit succeeds, the row is permanently removed from the database.

The Lifecycle of AsyncSession

For every incoming request, a new AsyncSession is created.

The AsyncSession exists only for the lifetime of a single request. After the response is returned, FastAPI automatically cleans up the session.

A production application should never share the same AsyncSession across multiple requests. Every request should receive its own session instance.

One common mistake is creating a session manually inside every endpoint.

session = AsyncSession(...)

Doing so bypasses FastAPI's Dependency Injection system and often leads to connection leaks.

Another common mistake is forgetting to commit transactions after modifying data. Without calling commit(), changes remain only inside the session and are never persisted to the database.

Conclusion

SQLAlchemy 2.x and AsyncSession provide the foundation for building asynchronous, production-grade FastAPI applications.

Together, they simplify database access, transaction management, and object-relational mapping while integrating seamlessly with FastAPI's Dependency Injection system.

In the next article, we will build on this foundation by exploring transactions, relationship mapping, lazy loading, eager loading, connection pooling, and performance optimization techniques that are essential for large-scale production systems.