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:

• Session Basics

fastsqla.Session dependency

Dependency used exclusively in endpoints to get an SQLAlchemy or SQLModel session.

Session is a FastAPI dependency that provides an asynchronous SQLAlchemy session or SQLModel one if it's installed. By defining an argument with type Session in an endpoint, FastAPI will automatically inject an 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 all 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

Async context manager that opens a new SQLAlchemy or SQLModel 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.

Returns:

Type	Description
AsyncGenerator[AsyncSession, None]	When SQLModel is not installed, an async generator that yields an SQLAlchemy AsyncSession.
AsyncGenerator[AsyncSession, None]	When SQLModel is installed, an async generator that yields an SQLModel AsyncSession which inherits from SQLAlchemy AsyncSession.

from fastsqla import open_session

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