Installation

pip install deps-injection

Compatibility between web frameworks and injection features

Quickstart with FastAPI, SQLAlchemy and pytest (sync sqlite)

fromcontextlibimportcontextmanagerfromrandomimportRandomfromtypingimportAnnotated,Any,Callable,Dict,IteratorimportpytestfromfastapiimportDepends,FastAPIfromsqlalchemyimportcreate_engine,textfromsqlalchemy.ormimportSession,sessionmakerfromstarlette.testclientimportTestClientfrominjectionimportDeclarativeContainer,Provide,inject,providers@contextmanagerdefdb_session_resource(session_factory:Callable[...,Session])->Iterator[Session]:session=session_factory()try:yieldsessionexceptException:session.rollback()finally:session.close()classSomeDAO:def__init__(self,db_session:Session)->None:self.db_session=db_sessiondefget_some_data(self,num:int)->int:stmt=text("SELECT :num").bindparams(num=num)data:int=self.db_session.execute(stmt).scalar_one()returndataclassDIContainer(DeclarativeContainer):db_engine=providers.Singleton(create_engine,url="sqlite:///db.db",pool_size=20,max_overflow=0,pool_pre_ping=False,    )session_factory=providers.Singleton(sessionmaker,db_engine.cast,autoflush=False,autocommit=False,    )db_session=providers.Resource(db_session_resource,session_factory=session_factory.cast,function_scope=True,    )some_dao=providers.Factory(SomeDAO,db_session=db_session.cast)SomeDAODependency=Annotated[SomeDAO,Depends(Provide[DIContainer.some_dao])]app=FastAPI()@app.get("/values/{value}")@injectasyncdefsqla_resource_handler_async(value:int,some_dao:SomeDAODependency,)->Dict[str,Any]:value=some_dao.get_some_data(num=value)return {"detail":value}@pytest.fixture(scope="session")deftest_client()->TestClient:client=TestClient(app)returnclientdeftest_sqla_resource(test_client:TestClient)->None:rnd=Random()random_int=rnd.randint(-(10**6),10**6)response=test_client.get(f"/values/{random_int}")assertresponse.status_code==200assertnotDIContainer.db_session.initializedbody=response.json()assertbody["detail"]==random_int

Quickstart with FastAPI, SQLAlchemy and pytest (async sqlite)

fromcontextlibimportasynccontextmanagerfromrandomimportRandomfromtypingimportAnnotated,Any,Callable,Dict,AsyncIteratorimportpytestfromfastapiimportDepends,FastAPIfromsqlalchemyimporttextfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromstarlette.testclientimportTestClientfrominjectionimportDeclarativeContainer,Provide,inject,providers@asynccontextmanagerasyncdefdb_session_resource(session_factory:Callable[...,AsyncSession])->AsyncIterator[AsyncSession]:session=session_factory()try:yieldsessionexceptException:awaitsession.rollback()finally:awaitsession.close()classSomeDAO:def__init__(self,db_session:AsyncSession)->None:self.db_session=db_sessionasyncdefget_some_data(self,num:int)->int:stmt=text("SELECT :num").bindparams(num=num)result=awaitself.db_session.execute(stmt)data:int=result.scalar_one()returndataclassDIContainer(DeclarativeContainer):# need to install aiosqlite and greenletdb_engine=providers.Singleton(create_async_engine,url="sqlite+aiosqlite:///db.db",pool_pre_ping=False,    )session_factory=providers.Singleton(async_sessionmaker,db_engine.cast,autoflush=False,autocommit=False,    )db_session=providers.Resource(db_session_resource,session_factory=session_factory.cast,function_scope=True,    )some_dao=providers.Factory(SomeDAO,db_session=db_session.cast)SomeDAODependency=Annotated[SomeDAO,Depends(Provide[DIContainer.some_dao])]app=FastAPI()@app.get("/values/{value}")@injectasyncdefsqla_resource_handler_async(value:int,some_dao:SomeDAODependency,)->Dict[str,Any]:value=awaitsome_dao.get_some_data(num=value)return {"detail":value}@pytest.fixture(scope="session")deftest_client()->TestClient:client=TestClient(app)returnclientdeftest_async_sqla_resource(test_client:TestClient)->None:rnd=Random()random_int=rnd.randint(-(10**6),10**6)response=test_client.get(f"/values/{random_int}")assertresponse.status_code==200assertnotDIContainer.db_session.initializedbody=response.json()assertbody["detail"]==random_int