python-dependency-injector
python-dependency-injector copied to clipboard
Lazy/Proxy provide injection
My app wires up some dependencies on intialization and I want to override those during my unit-tests but since they are already injected I'm unable to override them.
This can current be achieved by injecting the provider instead of providing the dependency, i.e.
@inject
def my_func(my_var: int, my_dependency_provider: Provider["my_container_attr"])
my_dependency = my_depenency_provider()
...
my_dependency.do_work()
...
my_dependency.do_work2()
I'd like a ProxyProvide and LazyProvide wiring adding, such that it won't resolve the provider until it is called, or, it resolves on every use of the object.
For example, the equivilant of the above example would be,
@inject
def my_func(my_var: int, my_dependency: LazyProvide["my_container_attr"]):
...
my_dependency.do_work()
...
my_dependency.do_more_work()
The difference between the two new markers is,
LazyProvide = Calls the provider only on the first use of the proxy object. ProxyProvide = Resolves the provider on every use of the proxy object.
Below is the patch I've been using for now,
"""
Additional functionality added to dependency-injector.
ProvideProxy marker will call the provider on every invocation of the object.
ProvideLazy marker will call the provider on the first invocation of the object.
Provide marker will call provider and inject that value immediately.
"""
from collections.abc import Callable
from functools import wraps
from typing import Any, cast
from dependency_injector.wiring import (
F,
Provider,
_fetch_reference_injections,
_get_patched,
)
from werkzeug.local import LocalProxy
class ProvideLazy(Provider):
"""Marker for providing a Proxy object that lazily calls the Provider."""
class ProvideProxy(Provider):
"""Marker for providing a Proxy that calls the Provider on every invocation."""
def inject_ext(fn: F) -> F:
"""Extended version of the dependency-injector inject.
Provides support for the ProvideLazy and ProvideProxy markers.
"""
reference_injections, reference_closing = _fetch_reference_injections(fn)
@wraps(fn)
def _lazyinject(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401
for k, v in reference_injections.items():
if isinstance(v, ProvideProxy):
kwargs[k] = LocalProxy(kwargs[k])
if isinstance(v, ProvideLazy):
kwargs[k] = _lazy_proxy(kwargs[k])
return fn(*args, **kwargs)
patched = _get_patched(_lazyinject, reference_injections, reference_closing)
return cast(F, patched)
def _lazy_proxy(provider: Callable[[], Any]) -> Any: # noqa: ANN401
"""Create a new lazy proxy object."""
provided = False
provided_value = None
def _provide() -> Any: # noqa: ANN401
nonlocal provided, provided_value, provider
if provided is True:
return provided_value
provided_value = provider()
provided = True
return None
return LocalProxy(_provide)