INNER CODE UNIT · Python

setup_db_factories

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

def setup_db_factories(
    di_builder: DiBuilder, db_engine: AsyncEngine, session_factory: async_sessionmaker[AsyncSession]
) -> None:
    di_builder.bind(bind_by_type(Dependent(lambda *args: db_engine, scope=DiScope.APP), AsyncEngine))
    di_builder.bind(
        bind_by_type(
            Dependent(lambda *args: session_factory, scope=DiScope.APP),
            async_sessionmaker[AsyncSession],
        ),
    )
    di_builder.bind(bind_by_type(Dependent(build_sa_session, scope=DiScope.REQUEST), AsyncSession))
    di_builder.bind(bind_by_type(Dependent(UserRepoImpl, scope=DiScope.REQUEST), UserRepo, covariant=True))
    di_builder.bind(bind_by_type(Dependent(UserReaderImpl, scope=DiScope.REQUEST), UserReader, covariant=True))


def setup_event_bus_factories(
    di_builder: DiBuilder,
    rq_connection_pool: aio_pika.pool.Pool[aio_pika.abc.AbstractConnection],

View source record →