Simple Dependency Injection framework
Project description
"That Depends"
This package is dependency injection framework for Python, mostly inspired by python-dependency-injector
.
Quickstart
Install
pip install that-depends
Usage
DI-container with dependencies:
import dataclasses
import logging
import typing
from that_depends import BaseContainer, providers
logger = logging.getLogger(__name__)
def create_sync_resource() -> typing.Iterator[str]:
logger.debug("Resource initiated")
yield "sync resource"
logger.debug("Resource destructed")
async def create_async_resource() -> typing.AsyncIterator[str]:
logger.debug("Async resource initiated")
yield "async resource"
logger.debug("Async resource destructed")
@dataclasses.dataclass(kw_only=True, slots=True)
class IndependentFactory:
dep1: str
dep2: int
@dataclasses.dataclass(kw_only=True, slots=True)
class SyncDependentFactory:
independent_factory: IndependentFactory
sync_resource: str
@dataclasses.dataclass(kw_only=True, slots=True)
class AsyncDependentFactory:
independent_factory: IndependentFactory
async_resource: str
class DIContainer(BaseContainer):
sync_resource = providers.Resource[str](create_sync_resource)
async_resource = providers.AsyncResource[str](create_async_resource)
independent_factory = providers.Factory(IndependentFactory, dep1="text", dep2=123)
sync_dependent_factory = providers.Factory(
SyncDependentFactory,
independent_factory=independent_factory,
sync_resource=sync_resource,
)
async_dependent_factory = providers.Factory(
AsyncDependentFactory,
independent_factory=independent_factory,
async_resource=async_resource,
)
Usage with Fastapi
:
import contextlib
import typing
import fastapi
from starlette import status
from starlette.testclient import TestClient
from tests import container
@contextlib.asynccontextmanager
async def lifespan_manager(_: fastapi.FastAPI) -> typing.AsyncIterator[None]:
yield
await container.DIContainer.tear_down()
app = fastapi.FastAPI(lifespan=lifespan_manager)
@app.get("/")
async def read_root(
sync_dependency: typing.Annotated[
container.AsyncDependentFactory,
fastapi.Depends(container.DIContainer.async_dependent_factory),
],
) -> str:
return sync_dependency.async_resource
client = TestClient(app)
response = client.get("/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == "async resource"
Usage with Litestar
:
import typing
import fastapi
import contextlib
from litestar import Litestar, get
from litestar.di import Provide
from litestar.status_codes import HTTP_200_OK
from litestar.testing import TestClient
from tests import container
@get("/")
async def index(injected: str) -> str:
return injected
@contextlib.asynccontextmanager
async def lifespan_manager(_: fastapi.FastAPI) -> typing.AsyncIterator[None]:
yield
await container.DIContainer.tear_down()
app = Litestar(
route_handlers=[index],
dependencies={"injected": Provide(container.DIContainer.async_resource)},
lifespan=[lifespan_manager],
)
def test_litestar_di() -> None:
with (TestClient(app=app) as client):
response = client.get("/")
assert response.status_code == HTTP_200_OK, response.text
assert response.text == "async resource"
Docs
Providers
Resource
- Resource initialized only once and have teardown logic.
- Generator function is required.
import typing
from that_depends import BaseContainer, providers
def create_sync_resource() -> typing.Iterator[str]:
# resource initialization
yield "sync resource"
# resource teardown
class DIContainer(BaseContainer):
sync_resource = providers.Resource(create_sync_resource)
AsyncResource
- Same as but async generator function is required.
import typing
from that_depends import BaseContainer, providers
async def create_async_resource() -> typing.AsyncIterator[str]:
# resource initialization
yield "async resource"
# resource teardown
class DIContainer(BaseContainer):
async_resource = providers.AsyncResource(create_async_resource)
Singleton
- Initialized only once, but without teardown logic.
- Class or simple function is allowed.
import dataclasses
from that_depends import BaseContainer, providers
@dataclasses.dataclass(kw_only=True, slots=True)
class SingletonFactory:
dep1: bool
class DIContainer(BaseContainer):
singleton = providers.Singleton(SingletonFactory, dep1=True)
Factory
- Initialized on every call.
- Class or simple function is allowed.
import dataclasses
from that_depends import BaseContainer, providers
@dataclasses.dataclass(kw_only=True, slots=True)
class IndependentFactory:
dep1: str
dep2: int
class DIContainer(BaseContainer):
independent_factory = providers.Factory(IndependentFactory, dep1="text", dep2=123)
AsyncFactory
- Initialized on every call, as Factory.
- Async function is required.
import datetime
from that_depends import BaseContainer, providers
async def async_factory() -> datetime.datetime:
return datetime.datetime.now(tz=datetime.UTC)
class DIContainer(BaseContainer):
async_factory = providers.Factory(async_factory)
List
- List provider contains other providers.
- Resolves into list of dependencies.
import random
from that_depends import BaseContainer, providers
class DIContainer(BaseContainer):
random_number = providers.Factory(random.random)
numbers_sequence = providers.List(random_number, random_number)
Main decisions:
- Every dependency resolving is async, so you should construct with
await
keyword:
from tests.container import DIContainer
async def main():
some_dependency = await DIContainer.independent_factory()
- No containers initialization to avoid wiring -> only one global instance of container is supported
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
that_depends-1.5.0.tar.gz
(4.5 kB
view hashes)
Built Distribution
Close
Hashes for that_depends-1.5.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ef2817226c4b19ef373f09f82a13cbc5855e352194cfa659215af3b5fe22b83f |
|
MD5 | 66375861019762676c6755707775d2fb |
|
BLAKE2b-256 | 05c31dfc11d9742562fb2b9a269169fd1739e9d6437ed73cd44fcc1b4afa4c63 |