Dependency Injection Module

class carthage.dependency_injection.AsyncInjectable(**kwargs)

Bases: Injectable

An Injectable that supports asyncronous operations as part of making a dependency available. This happens in several phases:

Prior to construction, all the dependencies of the Injectable are prepared.
async_resolve() is called. This asynchronous method can return a different object, which entirely replaces this object as the provider of the dependency. The async_resolve protocol is intended for cases where figuring out which object will provide a dependency requires asynchronous operations. In many cases async_resolve() returns self.
Call async_ready() to prepare this object. This may include doing things like running setup_task().

async async_become_ready(cycle_set=None, dependency_key=None): Interface point to request that a dependency be fully ready (async_ready() is called exactly once). This function manages tracking to make sure that async_ready is called once, and waits for that call if needed. This method also makes sure that dependencies registered with inject() are made ready before async_ready() is called. Subclasses should not override this method but instead should override async_ready().

async async_resolve(): Returns self or an object that should replace self in providing dependencies.

class carthage.dependency_injection.AsyncInjector(injector, loop)

Bases: Injectable

An asynchronous injector. AsyncInjector is not a subclass of Injector because AsyncInjector’s call function is a coroutine and so it has an incompatible interface. In other ways the classes should behave the same.

This class overrides Injectable’s behavior of claiming the injector. Instead, if you construct an AsyncInjector you get exactly what you asked for: an AsyncInjector that maps directly onto the injector you construct. Note however that when an AsyncInjector is constructed by AsyncInjectable, the injector is claimed properly.

async __call__(cls, *args, **kwargs): Coroutine to Construct an instance of cls using the providers in this injector. Instantiate providers as needed. In general a sub-injector is not constructed. However if any keyword arguments pased in specify a dependency, then construct an injector for that. Keyword arguments and arguments are passed to the class to construct the object. If keyword arguments do specify a dependency, they must satisfy the InjectionKey involved.

async filter_instantiate_async(target, predicate, *, stop_at=None, ready=None): Like filter_instantiate() except in an async context. With filter_instance objects are not-ready by default. With filter_instance_async, the ready state is default (true unless in the middle of instantiating something with ready false).

exception carthage.dependency_injection.AsyncRequired(msg, context): Bases: RuntimeError

class carthage.dependency_injection.DeferredInjection(injector: 'Injector' = None, key: 'InjectionKey' = None, value_: 'object' = NotPresent)

Bases: object

async instantiate_async(): Instantiates the deferred injection. After the first successful call, *self.value returns the result of self.injector.get_instance(self.key) Handles the case where an explicit value is passed in rather than an injector and injection key.

value_to_provide()

Return the value such that:

injector.add_provider(key, self.value_to_provide())

Will be the same as:

await self.instantiate()
injector.add_provider(key, self.value)

while still keeping the value deferred.

If a specific value is passed into the constructor, use that value directly.
Otherwise construct an injector_xref

exception carthage.dependency_injection.ExistingProvider(k, old_p, new_p): Bases: RuntimeError

class carthage.dependency_injection.Injectable(*args, **kwargs)

Bases: object

Represents a class that has dependencies injected into it. By default, the __init__() will:

Store any keyword arguments corresponding to injection dependencies into instance variables of the same name as the keyword argument
Remove these keyword arguments prior to calling the superclass init.

So for example:

@inject(router = SiteRouter)
class Receiver(Injectable): pass

When Receiver is instantiated, its instances will have the router attribute set.

It is recommended but not required that classes with injected dependencies inherit from Injectable. The satisfies_injection_key() and supplementary_injection_keys() protocols are only available to classes that do inherit from Injectable.

Subclasses that may be mixins and that wish injected dependency handling different than the keyword assignment provided by Injectable must inherit from Injectable.

This class does not have Injector as an injected dependency. It is possible to have injected dependencies without doing so. However, in a dependency is Injector, then that injector will be claimed.

classmethod default_class_injection_key(): Called by Injector.add_provider() in the single argument form to get the injection key to use when this class is added to provide a dependency.

default_instance_injection_key(): Called when an instance of an injectable is used to add a dependency provider bby the single argument form of Injectable.add_provider()

classmethod supplementary_injection_keys(k: InjectionKey): Returns an iteration of InjectionKeys that should be added to an injector when this class is added. The current injection key is taken as an argument so that constraints applied can modify what keys are added.

exception carthage.dependency_injection.InjectionFailed(context): Bases: RuntimeError

class carthage.dependency_injection.InjectionKey(target_, *, require_type=False, **constraints)

Bases: object

Represents information about what is requested to satisfy a dependency.

Parameters:

target –
A type or other object representing what is desired.
- A type indicating an object of that type is desired
- An object such as a s tring that is a unique identifier for what is desired
_optional – If True, then if no provider for the dependency is registered, None will be passed rather than raising; if some other true value, that value will be returned by default. If NotPresent, then no kwarg will be specified if the dependency is not provided. Because of the historical API, there is no current way to have an optional key default to True.
_ready – If None (the default), then use the same readyness as the object into which this is being injected (or full readyness if this is a base operation). If True, then to satisfy this dependency, the provided object must be fully ready. If False, then a not ready object is preferred.
_defer – If True, then this dependency is a Deferred Dependency and will be passed into the object as a DeferredInjection.
_globally_unique – If true, then this provided dependency (typically in a call to carthage.modeling.provides()) is globally unique and need not be modified during container propagation.

class carthage.dependency_injection.Injector(*providers, parent_injector=None)

Bases: Injectable, EventListener

__call__(cls, *args, **kwargs): Construct an instance of cls using the providers in this injector. Instantiate providers as needed. In general a sub-injector is not constructed. However if any keyword arguments pased in specify a dependency, then construct an injector for that. Keyword arguments and arguments are passed to the class to construct the object. If keyword arguments do specify a dep.dependency, they must satisfy the InjectionKey involved.

add_provider(k, p=None, *, allow_multiple=False, close=True, replace=False)

Add a provider for a dependency

Either called as add_provider(provider) or add_provider(injection_key, provider). In the first form, a key is automatically constructed. For Injectable objects, see Injectable.default_instance_injection_key(). For Injectable types, see Injectable.default_class_injection_key(). For other objects, the unconstrained injection key for types (or for the type of the object) is used.

Parameters:

allow_multiple – If true, then this provider may be instantiated multiple times in sub-injectors. If false (the default) then the provider will be instantiated on the injector where it is added and used by all sub-injectors.
close – If true (the default), then closing the injector will close or cancel this provider. If false, then the provider will not be deallocated. As an example, if the asyncio.AbstractEventLoop is added as a provider, but closing this injector should not close the loop and end all async operations, then close can be set to false.
replace – If True, an existing provider is being updated. replace_provider() is a convenience function for calling add_provider() with replace set to True. Replacing providers may lead to inconsistent results if the provider has already been injected to fill a dependency in a constructed object.

claim(claimed_by=True)

Take ownership of the injector.

Parameters:: claimed_by – Either True or an object that this injector is marked as belonging to.

Returns either self or a new subinjector.

close(canceled_futures=None)

Close all subinjectors or providers

For every provider registered with this injector, call close() if it is exists. Then clear out all providers. Note that this will also close sub-injectors of providers.

If using AsyncInjector, it is better to call shutdown_injector() to cancel any running asynchronous tasks.

If the provider’s close() method takes an argument called canceled_futures then the canceled_futures argument will be passed down.

filter(target: type, predicate: list | Callable, stop_at: Injector = None)

Returns:: list of InjectionKey with target type of target and satisfying predicate in the current injector and its parents.
Parameters:: predicate – A list of constraints that must all be

present in the key, or a callable that returns true if the key should be included. The predicate is a mandatory argument; while it is possible to pass in something like lambda k: True to get all keys, this is only likely to be desirable when writing inspection code to examine the injection system. Supplementary injection keys and third party plugins will add InjectionKeys using targets in manners unexpected by the target class. When filtering for a target it is best to have a constraint with a well-defined meaning in the context of that class; as an example Machine uses the host constraint for a FQDN.

Parameters:: target – A target type to filter against. This can be

None to filter against all targets. Again, that is likely to be useful only in inspection logic.

Parameters:: stop_at – An injector which must be a parent of this injector. Do not progress past that injector in finding keys. So if stop_at is self, only locally registered keys are returned.

Example usage would be to find all registered plugins similar to the following:

plugin_keys =injector.filter(CarthagePlugin, ['name'])

filter_instantiate(target, predicate, *, stop_at=None, ready=False): Like filter() but an iterator returning tuples of keys instance.

get_instance(k, placement=None, loop=None, futures=None, defer_dependencies: bool = False)

Get an instance satisfying a given InjectionKey.

Parameters:

loop – An asyncio loop. If provided, then asynchronous activities can take place.
placement – A function taking one argument. Once the dependency is resolved, this function will be called with the result. More convenient for asyncronous operations.
futures – If the result cannot be determined immediately, then a future will be added to this list.
defer_dependencies – If key.defer is True and

defer_dependencies is True, return (and place) a DeferredInjection. Unless key.optional is True, it is an error if the injector hierarchy provides no dependency for key at the time get_instance is called. If defer_dependencies is false, then key.defer is ignored. defer_dependencies is True when get_instance is called while instantiating a class and handling its dependencies.

Note that If any of loop or futures, are provided, both must be provided. If loop is provided, then the return may be a future.

injector_containing(k): Return the first injector in our parent chain containing k or None if there is no such injector.

If k has not yet been instantiated, this injector would be the one against which the instantiation is recorded unless the provider was added with the allow_multiple argument to add_provider().

inspect(*, key_filter=None, include_parent: bool = False)

Inspect the contents of this injector. This is a generator yielding key, InjectedDependencyInspector pairs for each dependency provided by the injector.

Parameters:

key_filter – A function that takes a key and returns true if the inspector should inspect the dependency for this key.
include_parent – By default only this injector is covered; if True, then the inspection recurses into the parent.

class carthage.dependency_injection.InstantiationContext(injector: injector, triggering_injector: Injector, key: InjectionKey, provider: DependencyProvider, ready: bool)

Bases: BaseInstantiationContext, InjectedDependencyInspector

Represents the instantiation of an InjectionKey in the scope of a Injector.

done(): Indicate that the instantiation has completed

carthage.dependency_injection.aspect_for(cls: Type[Injectable], property: str)

A decorator for a class indicating that the class provides an optional dependency to another class. Once decorated, newly created instances of cls may have a property property if cls’s Injector is able to instantiate the decorated class. Usage might be something like:

@aspect_for(Machine, 'model')
class MachineModel:
. . .

Then in code that is given a machine you can do things like:

try:
    os = machine.model.os
except AttributeError: os = "unknown"

In a dependency injection framework, it is common for a middle layer not to care about domain-specific knowledge that is needed both in an interface layer and in domain-specific implementation code. As an example, implementations of Container don’t care much what operating system the container is running. However user-interface code to configure the container might need to set what operating system is desired, and various domain-specific BaseCustomization classes may need to change behavior based on the operating system. Aspects provide a convenient syntax for the injector hierarchy to be used to access this information.

class carthage.dependency_injection.dependency_quote(value)

Bases: object

Indicate that a value (or dependency provider) should not be subject to injcoter resolution. Used like:

    injector(func, override = dependency_quote(self))

For example if *self* is an object that is not yet :meth:`~AsyncInjectable.async_ready` as part of bootstrapping.  Can be used both in a keyword argument to an injector call or in calls to :meth:`Injector.add_provider`.

carthage.dependency_injection.inject(**dependencies)

A decorator to indicate that a function requires dependencies:

Sample Usage:

@inject(injector = Injector,
    router = InjectionKey(SiteRouter, site ='cambridge'))
def testfn(injector, router): pass

Can be applied to classes or functions. Note that when an injector is used, dependencies will be resolved through the injector even if they are supplied directly. For example in:

 @inject(dependency = InjectedClass)
def func(dependency):
      assert isinstance(dependency, InjectedClass)


 injector(func, dependency=InjectedClass)

The assertion will be true if InjectedClass is an Injectable because the injector will instantiate the class. Resolution provided by the injector includes:

Instantiating subclasses of Injectable and providing their dependencies.
Calling async_ready() on AsyncInjectable.

carthage.dependency_injection.inject_autokwargs(**dependencies)

Like inject() but explicitly marks that the keywords are expected to fall through to Injectable.__init__() Applies to all dependencies at the current level so can be used either like:

@inject_autokwargs(foo = bar)
class baz(Injectable):

or like:

@inject_autokwargs()
@inject(foo = bar)
class baz(Injectable):

:class:`InjectionKey`s with _optional set to NotPresent are not required.

carthage.dependency_injection.injection_failed_unlogged(): Typically InjectionFailed is logged at severity error. Within this context, InjectionFailed errors are not logged.

carthage.dependency_injection.injector_xref(injectable_key: InjectionKey, target_key: InjectionKey)

Request that one injector look up a target in another injector. Typical usage:

base_injector.add_provider(target_key,
    injector_xref(sub_injector_key, target_key))

Parameters:

injectable_key – The InjectionKey of an Injectable or an Injector in which the target is actually looked up. If None, then look up in the current injector.
target_key – An InjectionKey registered with the

injector belonging to injectable_key. It is important that target_key actually be provided by that injector. In the common case where the parent of injectable_key eventually chains back to the injector in which this injector_xref is provided, a loop can happen otherwise.

carthage.dependency_injection.instantiation_not_ready(ready=False)

In this context, calling an Injector, or calling methods like Injector.get_instance() will not bring objects to async_ready state unless the ready parameter is True.

This can be overridden in the key of a dependency. For example:

@inject_autokwargs(bar=InjectionKey(Bar, _ready=True))
class Foo(AsyncInjectable):

    async def async_ready(self): pass # do stuff in this method

And then later:

with instantiation_not_ready():
    foo = await ainjector(Foo)

The bar dependency of foo will be brought to async_ready state. However, foo itself will not be ready: foo.async_ready() will not be called.

Do not confuse this function with :var:`instantitate_to_ready`. This is a context manager to manipulate the desired ready state. instantiate_to_ready is an internal context variable that tracks desired ready state. Calling instantiation_not_ready(ready=True) may be a bit counter-intuitive; it will reverse the effect of calling instantiation_not_ready() and turn on default ready state.

carthage.dependency_injection.partial_with_dependencies(func, *args, **kwargs)

Partially aply arguments and keep injected dependencies

Like functools.partial except also preserves dependencies. Used typically when passing the result of partial to Injector.add_provider()

This implementation assumes that no dependencies are removed by passing arguments into partial that specify one of the injected dependencies.

async carthage.dependency_injection.resolve_deferred(ainjector, item, args: dict)

Often it is desirable to have a data structure that has some elements dynamically calculated. For example looking up something in the environment of an injector, or calling some function that takes dependencies from an injector. Examples of where this is used include NetworkConfig.

This function traverses a data structure starting at an atomic element, list, or dict. It returns a copy where:

InjectionKey are replaced with AsyncInjector.get_instance_async() in the context of the current injector
A callable is called in the context of the current injector. If the callable has a signature, and if items in args match keys in the signature, the value from args is plugged into the call. An injected dependency could be used instead, but in some cases the performance of setting up an injector just for one parameter value doesn’t make sense.
dicts, lists, and tuples are recursed, although tuples become lists.
Other items are not modified.

This function does not deal with recursive data structures.

async carthage.dependency_injection.shutdown_injector(injector, timeout=5)

Close an injector and cancel running tasks

This closes an injector, canceling any running tasks. It waits up to timeout seconds for any canceled tasks to terminate.

class carthage.dependency_injection.introspection.InjectedDependencyInspector(injector: Optional[Injector], provider: DependencyProvider, key: InjectionKey)

Using inject(), a dependency can be injected into a function or class. Various APIs such as get_dependencies_for() or get_dependencies_for() will allow introspection of these dependencies. Those methods iterate over *INjectedDependencyInspector*s to give information on each dependency.

all_keys(): Iterate over all keys that provide this dependency in the given injector.

get_value(ready=None): Returns the value provided for the dependency. May raise KeyError if not provided.

async get_value_async(ready=None): Like get_value() but asynchronous.

get_value_no_instantiate(): Return the current value without instantiating; I.E. possibly whatever is passed into add_provider.

injector: Optional[Injector]: The Injector that provides the dependency. None if the dependency is unresolved.

property instantiation_contexts: Return any ongoing instantiations referencing this dependency. Typically contexts are very short lived if is_final() is true. However if a dependency is itself waiting for dependencies, the contexts can be used to chain those dependencies.

property is_final: True if the dependency has been fully instantiated. If true, get_value() will never raise AsyncRequired.

property is_provided: True if the dependency is provided; False if there is no injector in the chain providing this dependency.

property provider_id: If two inspectors have the same provider_id, they are guaranteed to refer to the same value. It is possible that inspectors with different provider_id may refer to the same value. Once is_final returns true, value_id will be stable.

class carthage.dependency_injection.introspection.InstantiationContext(injector: injector, triggering_injector: Injector, key: InjectionKey, provider: DependencyProvider, ready: bool)

Represents the instantiation of an InjectionKey in the scope of a Injector.

done(): Indicate that the instantiation has completed

carthage.dependency_injection.introspection.instantiation_leaves(): Start from instantiation_roots and chase down all the dependencies. Return all contexts that do not have any dependencies waiting that have not already been seen.

Events

The dependency injection system emits several events.

add_provider
Emitted when Injector.add_provider is called. Dispatched to all the keys that the dependency will satisfy. The target of the event is the object providing the dependency, typically an uninstantiated class. Also dispatched to InjectionKey(Injector) as a wildcard. Contains the add_provider parameters as well as other_keys, indicating other keys by which this dependency will be provided.

dependency_progress
Emitted whenever an instantiation makes progress (for example resolving a AsyncInjectable or calling a coroutine. The target is a InstantiationContext. The value can be obtained with the get_value method. This event is dispatched to all the keys that the add_provider event would be dispatched to.

dependency_final
Emitted whenever an instantiation finalizes (async object is ready for example). Same target and keys as dependency_progress.