Использование зависимости БД в FastAPI без необходимости передавать ее через дерево функций

В настоящее время я работаю над POC с использованием FastAPI в сложной системе. Этот проект насыщен бизнес-логикой и по завершении будет взаимодействовать с более чем 50 различными таблицами базы данных. У каждой модели есть служба, а у некоторых из более сложных бизнес-логик есть собственная служба (которая затем взаимодействует / запрашивает различные таблицы через службы, специфичные для модели).

Хотя все работает, некоторые члены моей команды отказались от внедрения зависимостей для объекта Session. Самая большая проблема заключается в том, что в основном необходимо передать сеанс от контроллера к службе, второй службе и (в некоторых случаях) третьей службе дальше. В этих случаях функции промежуточной службы, как правило, не имеют запросов к базе данных. но функции, которые они вызывают в других службах, могут иметь некоторые из них. Жалоба в основном заключается в том, что это труднее поддерживать, и необходимость передавать объект БД повсюду кажется бесполезно повторяющимся.

Пример в виде кода:

базы данных / mysql.py (один из 3 dbs в проекте)

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

контроллеры / 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

службы / 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):
    ...

службы / 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()

Как видите, при использовании этого подхода мы застряли при передаче mysql_session и повторении его повсюду.

Вот два решения, о которых я подумал:

1. Добавление сеанса БД в состояние запроса Starlette

Я мог бы сделать это либо через событие app.startup (https://fastapi.tiangolo.com/advanced/events/) или промежуточное ПО. Однако это означает передачу состояния запроса туда и обратно аналогичным образом (если я правильно понимаю).

2. Подход к области сеанса с использованием диспетчера контекста

В значительной степени я бы вместо этого превратил функцию get_db в диспетчер контекста, а не вводил бы ее как зависимость. Безусловно, самый чистый конечный результат, однако он полностью противоречит концепции совместного использования одного сеанса БД по запросу.

Я рассмотрел полностью асинхронный подход с использованием encode / databases, как показано в документации FastAPI (https://fastapi.tiangolo.com/advanced/async-sql-databases/), однако одна из баз данных, с которыми мы работаем на SqlAlchemy, используется через плагин, и я предполагаю, что не поддерживает асинхронность вне коробка (Vertica). Если я ошибаюсь, то могу рассмотреть полностью асинхронный подход.

Итак, в конце концов, мне интересно, можно ли сделать что-то более чистое, не ставя под угрозу подход к единственному сеансу на запрос?