INNER CODE UNIT · Python

setup_event_bus_factories

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

def setup_event_bus_factories(
    di_builder: DiBuilder,
    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: rq_connection_pool, scope=DiScope.APP),
            aio_pika.pool.Pool[aio_pika.abc.AbstractConnection],
        ),
    )
    di_builder.bind(
        bind_by_type(
            Dependent(lambda *args: rq_channel_pool, scope=DiScope.APP),
            aio_pika.pool.Pool[aio_pika.abc.AbstractChannel],
        ),
    )
    di_builder.bind(

View source record →