Wiring¶
Wiring feature provides a way to inject container providers into the functions and methods.
To use wiring you need:
Place @inject decorator. Decorator
@inject
injects the dependencies.Place markers. Wiring marker specifies what dependency to inject, e.g.
Provide[Container.bar]
. This helps container to find the injections.Wire the container with the markers in the code. Call
container.wire()
specifying modules and packages you would like to wire it with.Use functions and classes as you normally do. Framework will provide specified injections.
import sys
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide
class Service:
...
class Container(containers.DeclarativeContainer):
service = providers.Factory(Service)
@inject
def main(service: Service = Provide[Container.service]) -> None:
...
if __name__ == '__main__':
container = Container()
container.wire(modules=[sys.modules[__name__]])
main()
Markers¶
Wiring feature uses markers to make injections. Injection marker is specified as a default value of a function or method argument:
from dependency_injector.wiring import inject, Provide
@inject
def foo(bar: Bar = Provide[Container.bar]):
...
Specifying an annotation is optional.
There are two types of markers:
Provide[foo]
- call the providerfoo
and injects the resultProvider[foo]
- injects the providerfoo
itself
from dependency_injector.wiring import inject, Provider
@inject
def foo(bar_provider: Callable[..., Bar] = Provider[Container.bar]):
bar = bar_provider()
...
You can use configuration, provided instance and sub-container providers as you normally do.
@inject
def foo(token: str = Provide[Container.config.api_token]):
...
@inject
def foo(timeout: int = Provide[Container.config.timeout.as_(int)]):
...
@inject
def foo(baz: Baz = Provide[Container.bar.provided.baz]):
...
@inject
def foo(bar: Bar = Provide[Container.subcontainer.bar]):
...
You can compound wiring and Resource
provider to implement per-function execution scope.
See Resources, wiring and per-function execution scope for details.
Also you can use Provide
marker to inject a container.
import sys
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide
class Service:
...
class Container(containers.DeclarativeContainer):
service = providers.Factory(Service)
@inject
def main(container: Container = Provide[Container]):
service = container.service()
...
if __name__ == '__main__':
container = Container()
container.wire(modules=[sys.modules[__name__]])
main()
Wiring with modules and packages¶
To wire a container with a module you need to call container.wire(modules=[...])
method. Argument
modules
is an iterable of the module objects.
from yourapp import module1, module2
container = Container()
container.wire(modules=[module1, module2])
You can wire container with a package. Container walks recursively over package modules.
from yourapp import package1, package2
container = Container()
container.wire(packages=[package1, package2])
Arguments modules
and packages
can be used together.
When wiring is done functions and methods with the markers are patched to provide injections when called.
@inject
def foo(bar: Bar = Provide[Container.bar]):
...
container = Container()
container.wire(modules=[sys.modules[__name__]])
foo() # <--- Argument "bar" is injected
Injections are done as keyword arguments.
foo() # Equivalent to:
foo(bar=container.bar())
Context keyword arguments have a priority over injections.
foo(bar=Bar()) # Bar() is injected
To unpatch previously patched functions and methods call container.unwire()
method.
container.unwire()
You can use that in testing to re-create and re-wire a container before each test.
import unittest
class SomeTest(unittest.TestCase):
def setUp(self):
self.container = Container()
self.container.wire(modules=[module1, module2])
self.addCleanup(self.container.unwire)
import pytest
@pytest.fixture
def container():
container = Container()
container.wire(modules=[module1, module2])
yield container
container.unwire()
Note
Wiring can take time if you have a large codebase. Consider to persist a container instance and avoid re-wiring between tests.
Note
Python has a limitation on patching individually imported functions. To protect from errors prefer importing modules to importing individual functions or make sure imports happen after the wiring:
# Potential error:
from .module import fn
fn()
Instead use next:
# Always works:
from . import module
module.fn()
Asynchronous injections¶
Wiring feature supports asynchronous injections:
class Container(containers.DeclarativeContainer):
db = providers.Resource(init_async_db_client)
cache = providers.Resource(init_async_cache_client)
@inject
async def main(
db: Database = Provide[Container.db],
cache: Cache = Provide[Container.cache],
):
...
When you call asynchronous function wiring prepares injections asynchronously. Here is what it does for previous example:
db, cache = await asyncio.gather(
container.db(),
container.cache(),
)
await main(db=db, cache=cache)
You can also use Closing
marker with the asynchronous Resource
providers:
@inject
async def main(
db: Database = Closing[Provide[Container.db]],
cache: Cache = Closing[Provide[Container.cache]],
):
...
Wiring does closing asynchronously:
db, cache = await asyncio.gather(
container.db(),
container.cache(),
)
await main(db=db, cache=cache)
await asyncio.gather(
container.db.shutdown(),
container.cache.shutdown(),
)
See Resources, wiring and per-function execution scope for
details on Closing
marker.
Note
Wiring does not not convert asynchronous injections to synchronous.
It handles asynchronous injections only for async def
functions. Asynchronous injections into
synchronous def
function still work, but you need to take care of awaitables by your own.
See also:
Provider Asynchronous injections
Resource provider Asynchronous initializers
Integration with other frameworks¶
Wiring feature helps to integrate with other frameworks like Django, Flask, etc.
With wiring you do not need to change the traditional application structure of your framework.
Create a container and put framework-independent components as providers.
Place wiring markers in the functions and methods where you want the providers to be injected (Flask or Django views, Aiohttp or Sanic handlers, etc).
Wire the container with the application modules.
Run the application.
import sys
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide
from flask import Flask, json
class Service:
...
class Container(containers.DeclarativeContainer):
service = providers.Factory(Service)
@inject
def index_view(service: Service = Provide[Container.service]) -> str:
return json.dumps({'service_id': id(service)})
if __name__ == '__main__':
container = Container()
container.wire(modules=[sys.modules[__name__]])
app = Flask(__name__)
app.add_url_rule('/', 'index', index_view)
app.run()
Take a look at other application examples: