FastApi Sqlalchemy how to manage transaction (session and multiple commits)

I had the same problem while using FastAPI. I couldn't find a way to use commit in separate methods and have them behave transactionally. What I ended up doing was a flush instead of the commit, which sends the changes to the db, but doesn't commit the transaction.

One thing to note, is that in FastAPI every request opens a new session and closes it once its done. This would be a rough example of what is happening using the example in the SQLAlchemy docs.

def run_my_program():
    # This happens in the `database = SessionLocal()` of the `get_db` method below
    session = Session()
    try:
        ThingOne().go(session)
        ThingTwo().go(session)

        session.commit()
    except:
        session.rollback()
        raise
    finally:
        # This is the same as the `get_db` method below
        session.close()

The session that is generated for the request is already a transaction. When you commit that session what is actually doing is this

When using the Session in its default mode of autocommit=False, a new transaction will be begun immediately after the commit, but note that the newly begun transaction does not use any connection resources until the first SQL is actually emitted.

In my opinion after reading that it makes sense handling the commit and rollback at the endpoint scope.

I created a dummy example of how this would work. I use everything form the FastAPI guide.

def create_user(db: Session, user: UserCreate):
    """
    Create user record
    """
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.flush() # Changed this to a flush
    return db_user

And then use the crud operations in the endpoint as follows

from typing import List
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

...

def get_db():
    """
    Get SQLAlchemy database session
    """
    database = SessionLocal()
    try:
        yield database
    finally:
        database.close()

@router.post("/users", response_model=List[schemas.User])
def create_users(user_1: schemas.UserCreate, user_2: schemas.UserCreate, db: Session = Depends(get_db)):
    """
    Create two users
    """
    try:
        user_1 = crud.create_user(db=db, user=user_1)
        user_2 = crud.create_user(db=db, user=user_2)
        db.commit()
        return [user_1, user_2]
    except:
        db.rollback()
        raise HTTPException(status_code=400, detail="Duplicated user")

In the future I might investigate moving this to a middleware, but I don't think that using commit you can get the behavior you want.

A more pythonic approach is to let a context manager perform a commit or rollback depending on whether or not there was an exception.

A Transaction is a nice abstraction of what we are trying to accomplish.

class Transaction:
    def __init__(self, session: Session = Depends(get_session)):
        self.session = session

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # rollback and let the exception propagate
            self.session.rollback()
            return False

        self.session.commit()
        return True

And, use it in your APIs, like so:

def some_api(tx: Transaction = Depends(Transaction)):
    with tx:
        ThingOne().go()
        ThingTwo().go()

No need to pass session to ThingOne and ThingTwo. Inject it into them, like so:

class ThingOne:
   def __init__(self, session: Session = Depends(get_session)):
       ...

class ThingTwo:
   def __init__(self, session: Session = Depends(get_session)):
       ...

I would also inject ThingOne and ThingTwo in the APIs as well:

def some_api(tx: Transaction = Depends(Transaction), 
            one: ThingOne = Depends(ThingOne), 
            two: ThingTwo = Depends(ThingTwo)):
    with tx:
        one.go()
        two.go()