How to return error response when database failed to commit

Question

Introduction:

In our FastAPI app, we implemented dependency which is being used to commit changes in the database (according to https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/).

Issue:

The issue we are facing (which is also partially mentioned on the website provided above) is that the router response is 200 even though the commit is not succeeded. This is simply because in our case commit or rollback functions are being called after the response is sent to the requestor.

Example:

Database dependency:

def __with_db(request: Request):
    db = Session()
    try:
        yield db
        db.commit()
    except Exception as e:
        db.rollback()
        raise e
    finally:
        db.close()

As an endpoint example, we import csv file with records, then create db model instances and then add them to the db session (for simplicity, irrelevant things deleted).

from models import Movies
...
@router.post("/import")
async def upload_movies(file: UploadFile, db: DbType = db_dependency):
    df = await read_csv(file)
    new_records = [Movies(**item) for item in df.to_dict("records")]
    db.add_all(new_records)         # this is still valid operation, no error here

    return "OK"

Everything within the endpoint doesn't raise an error, so the endpoint returns a positive response, however, once the rest of the dependency code is being executed, then it throws an error (ie. whenever one of the records has a null value).

Question:

Is there any solution, to how to actually get an error when the database failed to commit the changes? Of course, the simplest one would be to add db.commit() or even db.flush() to each endpoint but because of the fact we have a lot of endpoints, we want to avoid this repetition in each of them (if it is even possible).

Best regards,

Answer 1

This is the solution we have implemented for this individual use case.

As a reminder, the main purpose was to catch a database error and react to it by sending proper response to the client. The tricky part was that we wanted to omit the scenario of adding the same line of code to every endpoint as we have plenty of them.

We managed to solve it with middleware.

Updated dependency.py

def __with_db(request: Request):
   db = Session()

   #assign db to request state for middleware to be able to acces it 
   request.state.db = db

   yield db

Added one line to the app.py

# fastAPI version: 0.79.0
from starlette.middleware.base import BaseHTTPMiddleware
from middlewares import unit_of_work_middleware
...
app = FastAPI()
...
app.add_middleware(BaseHTTPMiddleware, dispatch=unit_of_work_middleware) #new line
...

And created main middleware logic in middlewares.py

from fastapi import Request, Response

async def unit_of_work_middleware(request: Request, call_next) -> Response:
    try:
        response = await call_next(request)

        # Committing the DB transaction after the API endpoint has finished successfully
        # So that all the changes made as part of the router are written into the database all together
        # This is an implementation of the Unit of Work pattern https://martinfowler.com/eaaCatalog/unitOfWork.html
        if "db" in request.state._state:
            request.state.db.commit()

        return response

    except:
        # Rolling back the database state to the version before the API endpoint call
        # As the exception happened, all the database changes made as part of the API call
        # should be reverted to keep data consistency
        if "db" in request.state._state:
            request.state.db.rollback()
        raise

    finally:
        if "db" in request.state._state:
            request.state.db.close()

The middleware logic is applied to every endpoint so each request that is coming is going through it.

I think it's relatively easy way to implement it and get this case resolved.

Answer 2

I don't know your FastAPI version. But as i know, from 0.74.0, dependencies with yield can catch HTTPException and custom exceptions before response was sent(i test 0.80.0 is okay):

async def get_database():
    with Session() as session:
        try:
            yield session
        except Exception as e:
            # rollback or other operation.
            raise e
        finally:
            session.close()

If one HTTPException raised, the flow is:
endpoint -> dependency catch exception -> ExceptionMiddleware catch exception -> respond

Get more info https://fastapi.tiangolo.com/release-notes/?h=asyncexi#breaking-changes_1

Additional, about commit only in one code block,

Solution 1, can use decorator:

def decorator(func):
    @wraps(func)
    async def wrapper(*arg, **kwargs):
        rsp = await func(*arg, **kwargs)
        if 'db' in kwargs:
            kwargs['db'].commit()
        return rsp

    return wrapper

@router.post("/import")
@decorator
async def upload_movies(file: UploadFile, db: DbType = db_dependency):
    df = await read_csv(file)
    new_records = [Movies(**item) for item in df.to_dict("records")]
    db.add_all(new_records)         # this is still valid operation, no error here

    return "OK"

@Barosh I think decorator is the easiest way. I also thought about middleware but it's not possible.

Solution 2, just an idea:

1. save session in request

async def get_database(request: Request):
    with Session() as session:
        request.state.session = session
        try:
            yield session
        except Exception as e:
            # rollback or other operation.
            raise e
        finally:
            session.close()

2. custom starlette.request_response

def custom_request_response(func: typing.Callable) -> ASGIApp:
    """
    Takes a function or coroutine `func(request) -> response`,
    and returns an ASGI application.
    """
    is_coroutine = iscoroutinefunction_or_partial(func)

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        request = Request(scope, receive=receive, send=send)
        if is_coroutine:
            response = await func(request)
        else:
            response = await run_in_threadpool(func, request)
        request.state.session.commit() # or other operation
        await response(scope, receive, send)

    return app

3. custom FastAPI.APIRoute.app

class CustomRoute(APIRoute):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.app = custom_request_response(self.get_route_handler())

4. create router with CustomRoute

router = APIRouter(route_class=CustomRoute)

I think this is a executable idea. You can test it.

Hope this is useful.