fastsqla-session

name: fastsqla-session description: > Manages async SQLAlchemy sessions in FastSQLA endpoints and background tasks. Covers the Session dependency (auto-commit/rollback lifecycle), flush vs commit rules, IntegrityError handling, and the open_session() context manager. Use when writing FastAPI endpoints or background tasks that interact with a database through FastSQLA.

FastSQLA Session Management

FastSQLA provides two ways to get an async SQLAlchemy session:

1. Session — A FastAPI dependency for endpoints.
2. open_session() — An async context manager for non-endpoint code.

Both follow the same lifecycle: auto-commit on success, auto-rollback on exception, always close.

Session Dependency

Session is a FastAPI dependency. Type-annotate an endpoint parameter as Session and FastAPI injects an async session automatically.

from fastsqla import Session, Item

@app.get("/users/{user_id}", response_model=Item[UserModel])
async def get_user(session: Session, user_id: int):
    user = await session.get(User, user_id)
    return {"data": user}

Lifecycle

You do not need to manage any of this yourself.

Critical: flush() vs commit()

NEVER call session.commit() inside an endpoint. FastSQLA commits automatically when the endpoint returns without error. Calling commit() manually breaks the transactional guarantee — if an error occurs after your manual commit, the already-committed changes cannot be rolled back.

Use session.flush() when you need server-generated data (e.g., auto-increment IDs) before the response is returned. Flushing sends pending changes to the database within the current transaction without finalizing it.

CORRECT — use flush()

from fastsqla import Session, Item

@app.post("/heroes", response_model=Item[HeroItem])
async def create_hero(session: Session, new_hero: HeroBase):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    await session.flush()  # hero.id is now populated
    return {"data": hero}
# FastSQLA auto-commits here

INCORRECT — do not call commit()

from fastsqla import Session, Item

@app.post("/heroes", response_model=Item[HeroItem])
async def create_hero(session: Session, new_hero: HeroBase):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    await session.commit()  # WRONG: breaks auto-commit lifecycle
    return {"data": hero}

If you call commit() and a later step raises an exception, the committed data cannot be rolled back. Let FastSQLA handle the commit.

IntegrityError Handling

When a flush() triggers a constraint violation (unique, foreign key, etc.), SQLAlchemy raises IntegrityError. The session is invalidated after this — you cannot continue using it for further queries.

The correct pattern is to catch IntegrityError after flush() and re-raise it as an HTTPException. The raised exception triggers FastSQLA's automatic rollback.

from http import HTTPStatus

from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from fastsqla import Session, Item

@app.post("/heroes", response_model=Item[HeroItem])
async def create_hero(session: Session, new_hero: HeroBase):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    try:
        await session.flush()
    except IntegrityError:
        raise HTTPException(
            status_code=HTTPStatus.CONFLICT, detail="Hero already exists"
        )
    return {"data": hero}

Rules for IntegrityError

• Always re-raise as an exception. Do not catch and silently ignore — the session is broken after an IntegrityError and cannot be used for further operations.
• Use flush(), not commit(), so the error is caught within the transaction.
• The HTTPException propagates up, triggering the automatic rollback, which is the correct behavior.

open_session()

For code that runs outside FastAPI endpoints (background tasks, CLI scripts, scheduled jobs), use open_session():

from fastsqla import open_session

async def sync_external_data():
    async with open_session() as session:
        result = await session.execute(select(Hero))
        heroes = result.scalars().all()
        for hero in heroes:
            hero.synced = True
    # auto-commit on successful exit

Lifecycle

open_session() follows the same pattern as the Session dependency:

• Context body succeeds — session is committed, then closed.
• Context body raises — session is rolled back, then closed. The exception is re-raised.
• Commit itself fails — session is rolled back, then closed. The commit exception is re-raised.

The third case is important: if everything in your async with block succeeds but the commit() call at exit fails (e.g., a deferred constraint violation), open_session() rolls back and re-raises the commit exception. You do not get a silent failure.

Summary Rules

1. Use Session for endpoints — type-annotate a parameter and FastAPI injects it. Never instantiate sessions manually in endpoint code.
2. Never call session.commit() in endpoints — FastSQLA auto-commits on success. Use session.flush() to get server-generated values.
3. Catch IntegrityError after flush() and re-raise as HTTPException — the session is broken after an integrity error; do not attempt further operations on it.
4. Use open_session() outside endpoints — background tasks, scripts, and other non-request code should use this async context manager.
5. Trust the lifecycle — success commits, exceptions roll back, sessions always close. Do not add manual commit/rollback/close calls.