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.