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

Question

I have a CRUD with insert and update functions with commit at the end of the each one as follows:

@staticmethod
def insert(db: Session, item: Item) -> None:
    db.add(item)
    db.commit()
   
   
@staticmethod
def update(db: Session, item: Item) -> None:
    ...
    db.commit()

I have an endpoint which receives a sqlalchemy session from a FastAPI dependency and needs to insert and update atomically (DB transaction).

What's the best practice when working with transactions? I can't work with the CRUD since it does more than one commit.

How should I handle the transactions? Where do you commit your session? in the CRUD? or only once in the FastAPI dependency function for each request?

Efficiency

You can do multiple transactions in a single request as long as it is not a problem for your business logic. Doing flush() doesn't do much. The changes are not persisted, just communicated with the database. If you need a transactional safety you need to make sure you are using SELECT .. FOR UPDATE correctly and doing a single transaction after you finished doing the updates. That is very safe but error prone. I'd say if your app won't see tons of requests where this kind of race condition is a real risk than you just ignore it and do commit() as you already have. – Kellda

Answer 1

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.

Estis

@JosephAsafGardin great to hear! If that's the case can you mark the answer as the accepted one? Cheers! – Estis

Thanks for sharing! I've been searching for a better way of doing this, and also how to do tests for these. – Roter

Answer 2

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()

Catenate

Answer 3

I solved this problem by an „fastapi middleware“ based approach:

• you define a middleware class that interceptyour http requests . It creats firsta new db session and put it in the local thread as variable. After request propagation the middleware can execute either a commit or rollback in error case. This way all the tx handling is encapsulated in the middleware and don’t need to deal with in your other code.
• you have then to register your middleware by your fastapi app
• Now you can every where in your code get the session from the local thread an just use it… you don’t have to care commits and rollbacks. All the staff is handled by the middleware. If needed I can post some example code here…

Flews