Skip to content

SQLAlchemy Session

Lifecycle

SQLAlchemy documentation recommends the following:

  • Keep the lifecycle of the session separate and external from functions and objects that access and/or manipulate database data.
  • Make sure you have a clear notion of where transactions begin and end, and keep transactions short, meaning, they end at the series of a sequence of operations, instead of being held open indefinitely.

FastSQLA automatically manages the session lifecycle:

  • If the request is successful, the session is committed.
  • If the request fails, the session is rolled back.
  • In all cases, at the end of the request, the session is closed and the associated connection is returned to the connection pool.

To learn more about SQLAlchemy sessions:

fastsqla.Session dependency

A dependency used exclusively in endpoints to get an SQLAlchemy session.

Session is a FastAPI dependency that provides an asynchronous SQLAlchemy session. By defining an argument with type Session in an endpoint, FastAPI will automatically inject an SQLAlchemy async session into the endpoint.

At the end of request handling:

  • If no exceptions are raised, the session is automatically committed.
  • If an exception is raised, the session is automatically rolled back.
  • In alls cases, the session is closed and the associated connection is returned to the connection pool.

Example:

example.py
@app.get("/heros/{hero_id}", response_model=Item[HeroItem])
async def get_items(
    session: Session, # (1)!
    item_id: int,
):
    hero = await session.get(Hero, hero_id)
    return {"data": hero}
  1. Just define an argument with type Session to get an async session injected in your endpoint.

Recommendation: Unless there is a good reason to do so, avoid committing the session manually, as FastSQLA handles it automatically.

If you need data generated by the database server, such as auto-incremented IDs, flush the session instead:

@app.post("/heros", response_model=Item[HeroItem])
async def create_item(session: Session, new_hero: HeroBase):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    await session.flush()
    return {"data": hero}

Or use the session context manager instead.

fastsqla.open_session context manager

An asynchronous context manager that opens a new SQLAlchemy async session.

To the contrary of the Session dependency which can only be used in endpoints, open_session can be used anywhere such as in background tasks.

On exit, it automatically commits the session if no errors occur inside the context, or rolls back when an exception is raised. In all cases, it closes the session and returns the associated connection to the connection pool.

from fastsqla import open_session

async def example():
    async with open_session() as session:
        await session.execute(...)