Minimalist dependency injection in Python

Overview

Dependency injection and Service Locator primer

Dependency injection (DI) is a well known and broadly used pattern to provide dependencies to a given object, following inversion of control principles. It provides loosely coupling between software artifacts by automatically providing any required dependencies upon object creation.

Let's say we have 2 classes, Client and Connection, where Client depends on Connection. A DI system will assemble the Connection object that Client requires, and instantiate a Client object with the assembled connection, as shown in the following example:

 1class Connection:
 2  def __init__(self):
 3    pass
 4
 5  def action():
 6    print("hello from connection")
 7
 8class Client:
 9
10  def __init__(self, conn: Connection):
11    self._conn = conn
12
13  def do_something():
14    self._conn.action()
15
16# assemble a Client instance by automatically instantiating and injecting 
17# Connection into the object constructor
18# di is our hypothetical dependency injection system
19
20obj = di.build(Client)
21obj.do_something()

In more complex scenarios, the DI system will resolve and assemble all dependencies on the dependency graph automatically, and inject them on the appropriate objects. This injection is usually done via constructor - in our case, python's __init__() - but other techniques may be used as well.

While at first it may seem that is a variant of the Factory pattern, in reality most DI systems provide the equivalent of a configurable factory.Depending on the implementation, the object dependency graph may be expressed via configuration, implicitly via reflection and/or by using annotations or explicit interfaces in the constructor (often complemented by configuration), by runtime routines that may act as a factory, or all of the previous options combined.

The pure approach to dependency injection will require that classes that are assembled do not interact at all with the dependency injection mechanism, akin to the Factory technique. All external dependencies are injected upon the object creation. This is necessary to ensure the Inversion of Control (IoC) principle, and many existing implementations in several languages follow this approach.

However, some implementations use another pattern called Service Locator. This technique is comprised of a central repository, the Registry, that keeps track of instantiated dependencies. Instead of providing external dependencies upfront, the class will interact with the registry to build and retrieve necessary dependencies dependencies, as shown in this example:

 1
 2class Connection:
 3  def __init__(self):
 4    pass
 5
 6  def action():
 7    print("hello from connection")
 8
 9
10class Client:
11
12  def __init__(self, sl: ServiceLocator):
13    # explicit dependency on ServiceLocator and implicit dependency on Connection
14    self._conn = sl.get(Connection) 
15  
16  def do_something():
17    self._conn.action()
18
19# sl is our hypothetical service locator system
20# sl will instantiate a copy of Client and pass itself as a dependency
21obj = sl.get(Client)
22obj.do_something()

While both approaches aren't mutually exclusive (the service locator can be dependency-injected onto a given object by using the traditional DI approach), it obscures external dependencies in the code, which its not desirable from a IoC perspective.

A more formal DI approach requires that all dependencies are instantiated at object creation; this has the advantage of generating errors for missing dependencies on object creation (or even at compile time), instead of runtime, as it usually happens with service location; The test process is also simplified, as mocked dependencies can also be provided easily.

However, in scripted languages - specially the case where a given application may only be run to serve a specific request (such as PHP) - this comes at a cost, as all dependencies for a given object are instantiated, regardless if they are necessary to fulfill the current execution cycle. Lazy loading is usually possible, but often requires proxy logic that may contribute to an over engineered implementation.

Implementations in compiled languages such as C# or Java often offer both a complete DI and Service Locator implementation, with different usage scenarios - dependency injection is used for assembling objects, and service location is used to manage dynamic references. In contrast, scripted languages may offer DI implementations ranging from minimal registries to full-featured, "by-the-book" DI pattern implementations. There are many different approaches that cater to specific needs or features, and with a different vision on how DI is implemented.

What about Python?

Oddly enough, dependency injection isn't that popular in Python, at least as a standalone notion. The dynamic module loading system and mechanisms to override behavior (such as decorators) are often used to achieve dependency injection, without being depicted as such.

As an example, the Django framework relies heavily on DI and IoC principles, but aren't exposed as a full-featured library. A quick glance at the settings.py configuration file will confirm this:

1# cache configuration example from settings.py
2CACHES = {
3  'default': {
4    'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache', # cache backend dependency
5    'LOCATION': '/var/tmp/django_cache', # configuration parameters
6  }
7}

There are also some available libraries that provide full-featured dependency injection to use in Python projects, such as dependency-injector, pinject or python-inject. However, it is quite easy to build simple a low-overhead Service Locator DI system, and this is what this article is all about.

Building a a simple DI system

As stated previously, it is quite simple to build a simple, yet powerful DI system, somewhat inspired on the Registry pattern. The specific pattern was chosen based not only on personal preference, but also due to the simplicity of implementation and overall performance. This kind of approach can be integrated with existing code bases with minimal effort, and - if desired - extended to leverage configuration-based injection and dependency resolution.

However, we will keep the current implementation to a minimum - our DI will use a code-first approach, and leverage lazy loading of dependencies - dependencies are registered in runtime, but actual instantiation is done upon actual usage.

Also, our implementation will provide limited scoping, a mechanism to provide different di containers to specific parts of the code, such as modules. Every registered object will have a single instance throughout the application, and our DI component will be made available using a global variable - di.

The Registry

The first step is to build the service registry, that will maintain objects (the dependencies) for a given key - in this case a string . These items are kept on an internal dictionary. To keep it simple, we are assuming cPython implementation of dict, that is generically thread-safe due to GIL. Removal methods are also omitted, as they are seldom necessary. We will cover override methods later in step 7, that can be used for testing. Our initial skeleton for di.py will look like this:

 1class Di:
 2 
 3  def __init__(self):
 4    """
 5    Initialize internal variables
 6    """
 7    self._registry = {} # internal dependency registry
 8
 9  def add(self, name: str, item):
10    """
11    Adds a new item to the registry
12    :param name:
13    :param item:
14    :return:
15    """
16    if self.has(name):
17      raise RuntimeError("Dependency name '{}' already in use".format(name))
18    self._registry[name] = item
19  
20  def get(self, name: str):
21    if not self.has(name):
22      raise RuntimeError("Dependency '{}' not found in the registry".format(name))
23    return self._registry[name]
24  
25  def has(self, name: str) -> bool:
26    """
27    Verifies if a given name exists in the registry
28    :param name:
29    :return:
30    """
31    return name in self._registry.keys()
32  
33  def keys(self) -> list:
34    """
35    Retrieve a list of all registered names
36    :return:
37    """
38    return list(self._registry.keys())

We can now use our over engineered, glorified dict to manually register objects:

 1class A:
 2
 3  def __init__(self, di: Di):
 4    pass
 5
 6  def action(self):
 7    print("Hi from A")
 8
 9class B:
10
11  def __init__(self, di: Di):
12    self._a = di.get('A') # ask Di instance for the required B dependency
13
14  def run(self):
15    self._a.action()
16
17
18di = Di() # instantiate our DI object
19di.add('A', A(di)) # register an instance of A, inject Di instance
20di.add('B', B(di)) # register an instance of B, inject Di instance
21
22di.get('B').run() # print message "Hi from A"

Automatic registration of classes

One major limitation in the previous step is that objects (dependencies) must be registered explicitly via code. However, by using a decorator, we can easily extend it to allow automatic registering of classes.As a side-effect of this approach, we can also provide custom factories for dependency registration and lazy loading by default. So, we add a register() method and extend a bit both the get() and add() method to our existing di.py:

 1import types
 2from inspect import isclass
 3
 4
 5class Di:
 6    
 7    (...)
 8    
 9    def register(self, name: str):
10        """
11        Decorator to register classes
12        :param name:
13        :return: wrapper function for registered item
14        """
15
16        def wrap(fn):  # wrapper function to automatically register the object
17            self.add(name, fn)
18        return wrap
19
20    def add(self, name: str, item):
21        """
22        Adds a new item to the registry
23        :param name:
24        :param item:
25        :return:
26        """
27        if self.has(name):
28            raise RuntimeError("Dependency name '{}' already in use".format(name))
29        if isclass(item):
30            # if it is a class, we'll create a wrap function to instantiate the object
31            def cls_wrap(_di):
32                return item(_di)
33
34            self._registry[name] = cls_wrap  # store the wrapper class
35        else:
36            # or if it is an object or callable, just store it
37            self._registry[name] = item
38
39    def get(self, name: str):
40        """
41        Retrieve a dependency from the registry by name
42        - If dependency exists as a function or lambda, the result of the function replaces the registry contents
43        :param name: dependency name
44        :return: dependency object
45        """
46        if not self.has(name):
47            raise RuntimeError("Key '{}' not found in the registry".format(name))
48
49        item = self._registry[name]
50
51        # if callable, lets execute and use the result instead
52        # and replace the stored item with the result of the callable;
53        #
54        # if class, just instantiate the class
55        if type(item) in [types.LambdaType, types.FunctionType]:
56            item = item(self)
57            self._registry.pop(name)  # remove 'factory' or wrapper reference
58            self._registry[name] = item  # replace it with the actual object
59        return item
60
61    (...)

Now, lets use the new changes in the previous example and see if it works:

 1di = Di()
 2
 3@di.register('A')
 4class A:
 5    
 6    def __init__(self, di: Di):
 7        pass
 8    
 9    def action(self):
10        print("Hi from A")
11
12@di.register('B')
13class B:
14 
15    def __init__(self, di: Di):
16        self._a = di.get('A')
17        
18    def run(self):
19        self._a.action()
20        
21        
22di.get('B').run() # print message "Hi from A"

Prevent circular references

Now that we have a working implementation with automatic registration of classes, lets addreess a common problem - circular references. It may happen that class B requires class A, class A requires class C, but class C requires class A.

Lazy loading can prevent some of these problems, but it is dependent on the way our DI class is used. Ideally, we should prevent it from happening, and is quite easy to achieve. We just add a local name stack, and perform a couple of changes to our get() method:

 1import types
 2from inspect import isclass
 3
 4class Di:
 5
 6    def __init__(self):
 7        """
 8        Initialize internal variables
 9        """
10        self._registry = {} # internal dependency registry
11        self._stack = [] # internal context stack
12
13    def get(self, name: str):
14        """
15        Retrieve a dependency from the registry by name
16        - If dependency exists as a function or lambda, the result of the function replaces the registry
17        contents
18        :param name: dependency name
19        :return: dependency object
20        """
21        if not self.has(name):
22            raise RuntimeError("Key '{}' not found in the registry".format(name))
23
24        # check if current name is in the context stack
25        if name in self._stack: 
26            raise RuntimeError("Circular dependency detected on dependency '{}'".format(name))
27
28        # add the name to the stack before calling any potential wrappers
29        self._stack.append(name)
30        item = self._registry[name]
31
32        # if callable, lets execute and use the result instead
33        # and replace the stored item with the result of the callable;
34        #
35        # if class, just instantiate the class
36        if type(item) in [types.LambdaType, types.FunctionType]:
37            item = item(self)
38            self._registry.pop(name) # remove 'factory' or wrapper reference
39            self._registry[name] = item # replace it with the actual object
40
41        # play is over, remove name from the stack
42        self._stack.remove(name) 
43        return item
44    
45    (...)

We can test the circular detection system by adapting the example:

 1di = Di()
 2
 3@di.register('A')
 4class A:
 5
 6    def __init__(self, di: Di):
 7        self._c = di.get('C')  # class A requires C class
 8
 9    def action(self):
10        print("Hi from A")
11
12
13@di.register('B')
14class B:
15
16    def __init__(self, di: Di):
17        self._a = di.get('A')  # try to get A object, but will fail miserably
18
19    def run(self):
20        self._a.action()
21
22
23@di.register('C')
24class C:
25
26    def __init__(self, di: Di):
27        self._a = di.get('A')  # class C requires back A class - circular dependency
28
29    def run(self):
30        self._a.action()
31
32# will generate RuntimeError exception with message "Circular dependency detected on dependency 'A'"
33di.get('B').run()
34

Optimization

Our get() method encapsulates some logic and type verification, that can be optimized. Starting with v3.2, Python provides high-order functions via functools that can help, specifically functools.lru_cache() decorator. Lets add it to the get() method. Adding a single line can improve get() performance up to 50% when heavily reusing the same dependencies. If using Python >=v3.9, you can use functools.cache instead:

 1import functools
 2import types
 3from inspect import isclass
 4
 5class Di:
 6    (...)
 7    
 8    @functools.lru_cache(maxsize=None) # caches results from the get()
 9    def get(self, name: str):    
10    (...)

Taking advantage of lazy loading

Some applications will benefit from lazy loading, a technique that will defer instantiation of a dependency until it is actually used. Our simple implementation provides lazy loading out of the box, due to the usage of wrapper functions, but also requires moving the dependency loading closer to the point where the dependency is used.

The following example demonstrates the implementation of a custom factory for a class C object. The factory's code will only be run when di.get('C') is executed. The factory is only executed once, as our Di object will automatically perform the replacement of the stored factory with its own result:

 1di = Di()
 2@di.register('A')
 3class A:
 4    
 5    def __init__(self, di: Di):
 6        pass
 7    
 8    def action(self):
 9        print("Hi from A")
10
11@di.register('B')
12class B:
13    
14    def __init__(self, di: Di):
15        self._di = di # store di for local usage
16        self._a = None # variable will be filled when calling get_a() for the first time
17    
18    def run(self):
19        self.get_a().action() # retrieve actual dependency when used via method
20    
21    def get_a(self):
22        if self._a is None:
23            # assemble dependency when it is required, and store it on a local property
24            self._a = self._di.get('A') 
25        return self._a
26
27# some class that requires a factory
28class C:
29    
30    def __init__(self, b: B):
31        self._b = b
32    
33    def run(self):
34        self._b.run()
35
36@di.register('C') # register C as a factory
37def init_c(_di: Di): # factory for class C, receives di as a parameter
38    return C(_di.get('B')) # instantiate C and inject dependencies from di
39
40# print message "Hi from A"
41di.get('C').run()

Scoping

Many DI frameworks provide scoping, a mechanism that allows the creation of a local instance of the DI component. This instance should allow the registration and resolving of locally-scoped dependencies (such as in a module), but still provide access to the globally-scoped dependencies.

To achieve this behaviour, we first change the __init__() signature to receive an optional copy of the parent Di object. We then add a new method, scope() to build new scoped instances of our Di object. Finally, we change the get() method implementation to perform cascaded lookups to the parent instance, to maintain resolutiuon of global dependencies:

 1import functools
 2import types
 3from inspect import isclass
 4
 5class Di:
 6
 7    def __init__(self, di=None):
 8        """
 9        Initialize internal variables
10        """
11        self._parent = di # parent DI object
12        self._registry = {} # internal dependency registry
13
14    (...)
15
16    def scope(self, name: str):
17        """
18        Create a scoped DI instance
19        :param name:
20        :return:
21        """
22        result = Di(self)
23        self.add(name, result)
24        return result
25
26    @functools.lru_cache(maxsize=None)
27    def get(self, name: str):
28        """
29        Retrieve a dependency from the registry by name
30        - If dependency exists as a function or lambda, the result of the function replaces the registry contents
31        :param name: dependency name
32        :return: dependency object
33        """
34        if not self.has(name):
35            if self._parent is not None:
36                # if a parent exists, try to solve non-existing local name there
37                return self._parent(name)
38
39            raise RuntimeError("Key '{}' not found in the registry".format(name))
40        
41        item = self._registry[name]
42        
43        # if callable, lets execute and use the result instead
44        # and replace the stored item with the result of the callable;
45        #
46        # if class, just instantiate the class
47        if type(item) in [types.LambdaType, types.FunctionType]:
48            item = item(self)
49            self._registry.pop(name) # remove 'factory' or wrapper reference
50            self._registry[name] = item # replace it with the actual object
51        
52        return item

Please note, our approach to scoping requires unique names, as we are adding the child Di instance to the parent's registry. As a side-effect of this, removal/destruction of these scoped (child) instances must be done explicitly - the objects and their dependencies will still be referenced by the parent registry.

Scoping example:

 1di = Di()
 2local_di = di.scope('local') # our local scope di
 3
 4@di.register('A') # register class A on global DI
 5class A:
 6    
 7    def __init__(self, di: Di):
 8        pass
 9
10    def action(self):
11        print("Hi from A")
12
13@di.register('B') # register class B on global DI
14class B:
15
16    def __init__(self, di: Di):
17        self._a = di.get('A')
18    
19    def run(self):
20        self._a.action()
21
22@local_di.register('A') # register class C on local DI, but named A
23class C:
24    
25    def __init__(self, di: Di):
26        pass
27    
28    def action(self):
29        print("Hi from C")
30
31@local_di.register('new_B') # register class B on local DI, but named new_b
32class B:
33    
34    def __init__(self, di: Di):
35        self._a = di.get('A') # this will retrieve class C, because uses local registry
36
37    def run(self):
38        self._a.action()
39
40# print message "Hi from A", because B receives A dependency from the di, not local_di
41local_di.get('B').run() 
42# print message "Hi from C" because class C is injected with local_di onto 'new_B'
43local_di.get('new_B').run() 

Replacing and removing registry entries

One of the touted advantages of DI is the capability to easily glue mock items to the components being tested; With some additional changes, our Di is also able to replace existing registered dependencies. To achieve this, we add a new decorator called override() and change the add() method to allow optional replacement of existing registry entries and clear the lru_cache decorator cache when necessary:

 1import functools
 2import types
 3from inspect import isclass
 4
 5class Di:
 6
 7(...)
 8
 9    def override(self, name: str):
10        """
11        Decorator to override a dependency definition
12        :param name:
13        :return: wrapper function for registered item
14        """
15        def wrap(fn): # wrapper function to be detected as callable for the registered class
16            self.add(name, fn, True)
17        return wrap
18        
19    def add(self, name: str, item, replace=False): # added new parameter, replace
20        """
21        Adds a new item to the registry
22        :param name:
23        :param item:
24        :param replace:
25        :return:
26        """
27        if self.has(name) and not replace:
28            raise RuntimeError("Dependency name '{}' already in use".format(name))
29        
30        if isclass(item):
31            def cls_wrap(_di): # if it is a class, we'll create a wrap function to instantiate the object
32                return item(_di)
33            self._registry[name] = cls_wrap # store the wrapper class
34        else:
35            self._registry[name] = item # or if it is an object or callable, just store it
36        if replace: # if replacing existing item, clear lru_cache
37            self.get.cache_clear()
38(...)            

The removal of entries can easily be achieved by implementing a remove() method, whose function is to remove an entry from the internal registry and purge the get() cache:

 1import functools
 2import types
 3from inspect import isclass
 4class Di:
 5
 6(...)
 7
 8    def remove(self, name: str):
 9        """
10        Removes a registered dependency
11        :param name:
12        :return: 
13        """
14        if self.has(name):
15          del self._registry[name]
16          # clear lru_cache cache
17          self.get.cache_clear()
18        else:
19          raise RuntimeError("Dependency name '{}' not found in the registry".format(name))          
20(...)
21

Usage example:

 1di = Di()
 2
 3@di.register('A')
 4class A:
 5
 6  def __init__(self, di: Di):
 7    pass
 8
 9  def action(self):
10    print("Hi from A")
11
12@di.register('B')
13class B:
14
15  def __init__(self, di: Di):
16    self._di = di
17    self._a = None
18
19  def run(self):
20    self.get_a().action()
21
22  def get_a(self):
23    if self._a is None:
24      self._a = self._di.get('A')
25    return self._a
26
27@di.override('A') # replace class A definition with a new one
28class C:
29
30  def __init__(self, di: Di):
31    pass
32
33  def action(self):
34    print("Hi from C")
35
36di.get('B').run() # prints "Hi from C"
37di.remove('B') # removes the B dependency from the registry

Wrapping up

Our simple, yet quite useful DI implementation totals a little more than 100 lines. Here is how our final di.py looks like:

  1import functools
  2import types
  3from inspect import isclass
  4
  5
  6class Di:
  7
  8    def __init__(self, di=None):
  9        """
 10        Initialize internal variables
 11        """
 12        self._parent = di
 13        self._registry = {}  # internal dependency registry
 14
 15    def register(self, name: str):
 16        """
 17        Decorator to register classes
 18        :param name:
 19        :return: wrapper function for registered item
 20        """
 21
 22        def wrap(fn):  # wrapper function to be detected as callable for the registered class
 23            self.add(name, fn)
 24
 25        return wrap
 26
 27    def override(self, name: str):
 28        """
 29        Override a dependency definition
 30        :param name:
 31        :return: wrapper function for registered item
 32        """
 33
 34        def wrap(fn):  # wrapper function to be detected as callable for the registered class
 35            self.add(name, fn, True)
 36
 37        return wrap
 38
 39    def add(self, name: str, item, replace=False):
 40        """
 41        Adds a new item to the registry
 42        :param name:
 43        :param item:
 44        :param replace:
 45        :return:
 46        """
 47        if self.has(name) and not replace:
 48            raise RuntimeError("Dependency name '{}' already in use".format(name))
 49
 50        if isclass(item):
 51            def cls_wrap(_di):  # if it is a class, we'll create a wrap function to instantiate the object
 52                return item(_di)
 53
 54            self._registry[name] = cls_wrap  # store the wrapper class
 55        else:
 56            self._registry[name] = item  # or if it is an object or callable, just store it
 57        if replace:  # if replacing existing item, clear lru_cache
 58            self.get.cache_clear()
 59
 60    @functools.lru_cache(maxsize=None)
 61    def get(self, name: str):
 62        """
 63        Retrieve a dependency from the registry by name
 64        - If dependency exists as a function or lambda, the result of the function replaces the registry contents
 65        :param name: dependency name
 66        :return: dependency object
 67        """
 68        if not self.has(name):
 69            if self._parent is not None:
 70                return self._parent.get(name)
 71            raise RuntimeError("Key '{}' not found in the registry".format(name))
 72
 73        item = self._registry[name]
 74
 75        # if callable, lets execute and use the result instead
 76        # and replace the stored item with the result of the callable
 77        # if class, just instantiate the class
 78        if type(item) in [types.LambdaType, types.FunctionType]:
 79            item = item(self)
 80            self._registry.pop(name)  # remove 'factory' or wrapper reference
 81            self._registry[name] = item  # replace it with the actual object
 82
 83        return item
 84
 85    def scope(self, name: str):
 86        """
 87        Create a scoped DI instance
 88        :param name:
 89        :return:
 90        """
 91        result = Di(self)
 92        self.add(name, result)
 93        return result
 94
 95    def has(self, name: str) -> bool:
 96        """
 97        Verifies if a given name exists in the registry
 98        :param name:
 99        :return:
100        """
101        return name in self._registry.keys()
102
103    def keys(self) -> list:
104        """
105        Retrieve a list of all registered names
106        :return:
107        """
108        return list(self._registry.keys())
109
110    def remove(self, name: str):
111        """
112        Removes a registered dependency
113        :param name:
114        :return: 
115        """
116        if self.has(name):
117            del self._registry[name]
118            # clear lru_cache cache
119            self.get.cache_clear()
120        else:
121            raise RuntimeError("Dependency name '{}' not found in the registry".format(name))

A versioned implementation of our simple di, as well as a couple of examples and unit tests is available in the oddbit/python-di repository.