INNER CODE UNIT · Python

setup_di_builder

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

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)
    setup_db_factories(di_builder, di_engine, session_factory)
    setup_event_bus_factories(di_builder, rq_connection_pool, rq_channel_pool)


def setup_mediator_factory(
    di_builder: DiBuilder,
    mediator_factory: DependencyProviderType,
    scope: Scope,

View source record →