Using a DB dependency in FastAPI without having to pass it through a function tree

Question

I am currently working on a POC using FastAPI on a complex system. This project is heavy in business logic and will interact with 50+ different database tables when completed. Each model has a service, and some of the more complex business logic has its own service (which then interacts/queries with the different tables through the model-specific services).

While everything works, I've gotten some push-back from some members of my team regarding the dependency injection for the Session object. The biggest issue being mainly having to pass the Session from the controller, to a service, to a second service and (in a few cases), a third service further in. In those cases, the intermediary service functions tend to have no database queries but the functions that they call on other services might have some. The complaint mainly lies in this being more difficult to maintain and having to pass the DB object everywhere seems uselessly repetitive.

Example as code:

databases/mysql.py (one of 3 dbs in the project)

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

def get_uri():
    return 'the mysql uri'

engine = create_engine(get_uri())

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


def get_db():
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

controllers/controller1.py

from fastapi import APIRouter, HTTPException, Path, Depends
from sqlalchemy.orm import Session
from services.mysql.bar import get_bar_by_id
from services.mysql.process_x import bar_process
from databases.mysql import get_db

router = APIRouter(prefix='/foo')


@router.get('/bar/{bar_id}')
def process_bar(bar_id: int = Path(..., title='The ID of the bar to process', ge=1),
                          mysql_session: Session = Depends(get_db)):
    # From the crontroller, to a service which only runs a query. This is fine.
    bar = get_bar_by_id(bar_id, mysql_session)

    if bar is None:
        raise HTTPException(status_code=404,
                            detail='Bar not found for id: {bar_id}'.format(bar_id=bar_id))

    # This one calls a function in a service which has a lot of business logic but no queries
    processed_bar = bar_process(bar, mysql_session)

    return processed_bar

services/mysql/process_x.py

from .process_bar import process_the_bar
from models.mysql.w import W
from models.mysql.bar import Bar
from models.mysql.y import Y
from models.mysql.z import Z
from sqlalchemy.orm import Session


def w_process(w: W, mysql_session: Session):
    ...


def bar_process(bar: Bar, mysql_session: Session):
    # Very simplified, there's actually 5 conditional branching service calls here
    return process_the_bar(bar, mysql_session)


def y_process(y: Y, mysql_session: Session):
    ...


def z_process(z: Z, mysql_session: Session):
    ...

services/mysql/process_bar.py

from . import model_service1
from . import model_service2
from . import model_service3
from . import additional_rules_service
from libraries.bar_functions import do_thing_to_bar
from models.mysql.bar import Bar
from sqlalchemy.orm import Session


def process_the_bar(bar: bar, mysql_session: Session):
    process_result = list()

    # Many processing steps, not all of them require db and might work on the bar directly
    process_result.append(process1(bar, mysql_session))
    process_result.append(process2(bar, mysql_session))
    process_result.append(process3(bar, mysql_session))
    process_result.append(process4(bar))
    process_result.append(...(bar))
    process_result.append(processY(bar))


def process1(bar: Bar, mysql_session: Session):
    return model_service1.do_something(bar.val, mysql_session)


def process2(bar: Bar, mysql_session: Session):
    return model_service2.do_something(bar.val, mysql_session)


def process3(bar: Bar, mysql_session: Session):
    return model_service3.do_something(bar.val, mysql_session)

def process4-Y(bar: Bar, mysql_session: Session):
    # do something using the bar library, or maybe on another service with no queries
    return list()

As you can see, we're stuck passing the mysql_session and having it repeat everywhere when using this approach.

Here are a two solutions I have thought of:

1. Adding the DB session to the Starlette request state

I could do this either through the app.startup event ( https://fastapi.tiangolo.com/advanced/events/ ) or a middleware. However, it does mean passing the request state back and forth in a similar fashion (if my understanding of it is correct)

2. Session scope approach using Context Manager

Pretty much, I would turn the get_db function into a context manager instead and not inject it as a dependency. By far the cleanest end result, however it goes completely against the concept of sharing a single db session across the request.

I've considered the fully async approach using encode/databases as shown in the FastAPI documentation ( https://fastapi.tiangolo.com/advanced/async-sql-databases/ ), however one of the databases we are working with on SqlAlchemy is used through a plugin and I am assuming does not support async out of the box (Vertica). If I'm wrong, then I could consider the fully async approach.

So in the end, what I'm wondering is if it's possible to accomplish something "cleaner" without compromising the single session per request approach?

Answer 1

I have gotten some help directly from the FastAPI Github

As user Insomnes mentioned, what I am looking to do can be achieved by using ContextVar. I have tried it in my code and it seems to work just fine.