context-async-sqlalchemy - The best way to use sqlalchemy in an async python application

Source: Dev.to

Short Summary of SQLAlchemy

SQLAlchemy provides an Engine, which manages the database connection pool, and a Session, through which SQL queries are executed. Each session uses a single connection that it obtains from the engine.

The engine should have a long lifespan to keep the connection pool active. Sessions, on the other hand, should be short‑lived, returning their connections to the pool as quickly as possible.

(In the diagram, both engines are shown connected to the same database – this is just for illustration. In practice you typically use one engine per database, e.g., one for the master and another for a replica.)

Integration and Usage in an Application

Direct Usage

Create an engine and a session maker:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(DATABASE_URL)
session_maker = async_sessionmaker(engine, expire_on_commit=False)

A simple endpoint that creates a user:

@app.post("/users/")
async def create_user(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

If the creation process requires two separate queries:

@app.post("/users/")
async def create_user(name):
    await insert_user(name)
    await insert_user_profile(name)

async def insert_user(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

async def insert_user_profile(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

Problems

• Two transactions are opened when only one is desired.
• Code duplication of the session/transaction boilerplate.

A common refactor moves the context managers up:

@app.post("/users/")
async def create_user(name: str):
    async with session_maker() as session:
        async with session.begin():
            await insert_user(name, session)
            await insert_user_profile(name, session)

async def insert_user(name, session):
    await session.execute(stmt)

async def insert_user_profile(name, session):
    await session.execute(stmt)

However, the same duplication appears across multiple handlers:

@app.post("/dogs/")
async def create_dog(name):
    async with session_maker() as session:
        async with session.begin():
            ...

@app.post("/cats/")
async def create_cat(name):
    async with session_maker() as session:
        async with session.begin():
            ...

Dependency Injection

In FastAPI you can encapsulate session/transaction handling in a dependency:

async def get_atomic_session():
    async with session_maker() as session:
        async with session.begin():
            yield session

@app.post("/dogs/")
async def create_dog(name, session=Depends(get_atomic_session)):
    await session.execute(stmt)

@app.post("/cats/")
async def create_cat(name, session=Depends(get_atomic_session)):
    await session.execute(stmt)

This removes duplication, but the session remains open for the entire request lifecycle, preventing early release of the connection back to the pool. It also forces the session to be passed through layers that don’t need it:

@app.post("/some_handler/")
async def some_handler(session=Depends(get_atomic_session)):
    await do_first(session)
    await do_second(session)

async def do_first(session):
    await do_something()
    await insert_to_database(session)

async def insert_to_database(session):
    await session.execute(stmt)

do_first doesn’t use the session directly yet must accept and forward it, which can feel inelegant.

Wrappers Around SQLAlchemy

Various third‑party wrappers add convenience at the cost of a new API surface. Developers already familiar with SQLAlchemy often prefer to stay with the native API rather than learn a different syntax.

The New Library

context-async-sqlalchemy was created to address the shortcomings above. Its goals are:

• Minimal boilerplate and no code duplication
• Automatic commit or rollback when manual control isn’t required
• Ability to manually manage sessions and transactions when needed
• Suitable for both simple CRUD operations and complex logic
• No new syntax – pure SQLAlchemy
• Framework‑agnostic design

Simplest Scenario

Execute a single SQL query inside a handler without worrying about sessions or transactions:

from context_async_sqlalchemy import db_session

async def some_func() -> None:
    session = await db_session(connection)  # new session
    await session.execute(stmt)            # some SQL query
    # commit happens automatically

db_session creates (or reuses) a session and ensures it is closed when the request ends.

Multiple Queries Within One Transaction

@app.post("/users/")
async def create_user(name):
    await insert_user(name)
    await insert_user_profile(name)

async def insert_user(name):
    session = await db_session(connection)  # creates a session
    await session.execute(stmt)              # opens a connection and a transaction

async def insert_user_profile(name):
    session = await db_session(connection)  # gets the same session
    await session.execute(stmt)

In this pattern, db_session returns the same underlying session for the duration of the request, allowing all operations to share a single transaction while keeping the code concise.