title: Using a DB dependency in FastAPI without having to pass it through a function tree
body: <p>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).</p>
  <p>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.</p>
  <p>Example as code:</p>
  <p>databases/mysql.py (one of 3 dbs in the project)</p>
  <pre class="lang-py prettyprint-override"><code>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()
  
  </code></pre>
  <p>controllers/controller1.py</p>
  <pre class="lang-py prettyprint-override"><code>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
  
  </code></pre>
  <p>services/mysql/process_x.py</p>
  <pre class="lang-py prettyprint-override"><code>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):
      ...
  
  </code></pre>
  <p>services/mysql/process_bar.py</p>
  <pre class="lang-py prettyprint-override"><code>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()
  </code></pre>
  <p>As you can see, we're stuck passing the mysql_session and having it repeat everywhere when using this approach.</p>
  <p>Here are a two solutions I have thought of:</p>
  <ol>
  <li>Adding the DB session to the Starlette request state</li>
  </ol>
  <p>I could do this either through the app.startup event ( <a href="https://fastapi.tiangolo.com/advanced/events/" rel="noreferrer">https://fastapi.tiangolo.com/advanced/events/</a> ) 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)</p>
  <ol start="2">
  <li>Session scope approach using Context Manager</li>
  </ol>
  <p>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.</p>
  <p>I've considered the fully async approach using encode/databases as shown in the FastAPI documentation ( <a href="https://fastapi.tiangolo.com/advanced/async-sql-databases/" rel="noreferrer">https://fastapi.tiangolo.com/advanced/async-sql-databases/</a> ), 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.</p>
  <p>So in the end, what I'm wondering is if it's possible to accomplish something &quot;cleaner&quot; without compromising the single session per request approach?</p>