INNER CODE UNIT · Python

init_di_builder

SamWarden/user_service · src/user_service/main/di/main.py:27

def init_di_builder() -> DiBuilder:
    di_container = Container()
    di_executor = AsyncExecutor()
    di_scopes = [DiScope.APP, DiScope.REQUEST]
    di_builder = DiBuilderImpl(di_container, di_executor, di_scopes=di_scopes)
    return di_builder


def setup_di_builder(
    di_builder: DiBuilder,
    di_engine: AsyncEngine,
    session_factory: async_sessionmaker[AsyncSession],
    rq_connection_pool: aio_pika.pool.Pool[aio_pika.abc.AbstractConnection],
    rq_channel_pool: aio_pika.pool.Pool[aio_pika.abc.AbstractChannel],
) -> None:
    di_builder.bind(bind_by_type(Dependent(lambda *args: di_builder, scope=DiScope.APP), DiBuilder))
    di_builder.bind(bind_by_type(Dependent(build_uow, scope=DiScope.REQUEST), UnitOfWork))
    setup_mediator_factory(di_builder, get_mediator, DiScope.REQUEST)

View source record →