Last active
October 2, 2019 15:25
-
-
Save andres-fr/05e6c67fb29bb210c711cc8060ca1b67 to your computer and use it in GitHub Desktop.
Proper interfacing in Python
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- coding:utf-8 -*- | |
""" | |
Python's duck typing system allows for very pleasant prototyping. But at some | |
point, it is convenient to enforce some interfacing. This can be achieved | |
Through abstract base classes and annotations to a great extent. | |
Another approach is to use dependency injection and inversion of control | |
patterns. This gist exemplifies how to do that using ``dependency_injector``. | |
In this example, we have two types of dummy 'image processing' operations: | |
1. pick one image and a range, and apply some nondeterministic transform | |
2. pick two images and two ranges, and fuse them | |
And we want to achieve the following goals | |
1. Automated way of ensuring that the op developers follow that interface | |
2. Flexible and efficient way of chaining and running the operations | |
3. Interface-agnostic interaction between operations and the file system | |
For that, we structure the code in 4 main sections: | |
1. Actual code which should fulfill given interfaces and have no side-effects | |
2. Definitions for the graph nodes, which enforce the interfaces | |
3. Definition of the computational graph | |
4. Main routine to run the graph | |
From these, only the first one provides functionality. | |
* Abstracting point 2 allows to define different ways of interacting with | |
the filesystem (or any other service) without impacting functionality. | |
* Abstracting point 3 allows to define different chains of computations, | |
while being parallelizable, thread-safe and efficient in terms of memory | |
and runtime. | |
* Abstracting point 4 allows to specify deployment conditions independently | |
of the topology and semantics of the computation. Note how the graph container | |
is responsible for creating, running and orchestrating the functional | |
components. Also note how very dynamic configurations are still possible, | |
despite the level of abstraction. | |
""" | |
import os | |
import random | |
from dependency_injector import containers, providers | |
__author__ = "Andres FR" | |
################################################################################ | |
### ORIGINAL OPERATIONS | |
################################################################################ | |
def load_img_from_disk(img_dir, img_name): | |
""" | |
A dummy image loader | |
""" | |
return os.path.join(img_dir, img_name) | |
def img_operation1(img, a, b): | |
""" | |
A dummy operation | |
""" | |
num = random.randint(a, b) | |
print("[img_operation1] called!", img, num) | |
return img + " -> img_operation1_{}".format(num) | |
def img_operation2(img, a, b): | |
""" | |
A dummy operation | |
""" | |
num = random.randint(a, b) | |
print("[img_operation2] called!", img, num) | |
return img + " -> img_operation2_{}".format(num) | |
def img_operation3(img1, img2, a1, b1, a2, b2): | |
""" | |
A dummy operation | |
""" | |
bracket = [min(a1, a2), max(b1, b2)] | |
print("[img_operation3] called!", img1, img2, bracket) | |
return "img_operation3->{}".format(bracket) | |
def main_routine(operation, | |
img_names=("img_{}.png".format(x) for x in range(5))): | |
""" | |
A loop of dummy operations | |
""" | |
for i in img_names: | |
result = operation(i) | |
print(">>>>", result) | |
################################################################################ | |
### "INTERFACE" DEFINITIONS | |
################################################################################ | |
class ImageServer: | |
""" | |
This mock object provides images and stores them in a cache. | |
Its interface is the ``img_dir`` constructor parameter and the | |
``_call__(img_name) -> img`` method. | |
""" | |
def __init__(self, img_dir): | |
""" | |
""" | |
print("[ImageServer] constructor called!") | |
self.img_dir = img_dir | |
self.images = {} | |
def __call__(self, name): | |
""" | |
""" | |
try: | |
img = self.images[name] | |
print("[ImageServer] retrieving", name, "from cache") | |
return img | |
except KeyError: | |
print("[ImageServer] loading", name, "from", self.img_dir) | |
img = load_img_from_disk(self.img_dir, name) | |
self.images[name] = img | |
return img | |
class BasicImgProcessingNode: | |
""" | |
This mock object gets injected an image_server, and applies an operation | |
to an image provided by it. | |
The operations must fulfill the interface ``op(img, a, b)->img_result`` | |
""" | |
def __init__(self, img_server, operation, a, b): | |
""" | |
""" | |
print("[BasicImgProcessingNode] constructor called with args:", | |
operation.__name__, a, b) | |
self.fn = operation | |
self.a = a | |
self.b = b | |
self.img_server = img_server | |
def __call__(self, img_name): | |
""" | |
""" | |
print("[BasicImgProcessingNode] called for image", img_name) | |
img = self.img_server(img_name) | |
output = self.fn(img, self.a, self.b) | |
return output | |
class FusionNode: | |
""" | |
This mock object gets injected two image processing nodes. When called, | |
applies them to an image and then passes the results to a given fusion | |
operation, which fulfills the interface | |
``op(img1, img2 a1, b1, a2, b2)->img_result`` | |
""" | |
def __init__(self, node1, node2, operation): | |
""" | |
""" | |
print("[FusionNode] constructor called") | |
self.n1 = node1 | |
self.n2 = node2 | |
self.fn = operation | |
def __call__(self, img_name): | |
""" | |
""" | |
print("[FusionNode] called for image", img_name) | |
im1 = self.n1(img_name) | |
im2 = self.n2(img_name) | |
output = self.fn(im1, im2, self.n1.a, self.n1.b, self.n2.a, self.n2.b) | |
return output | |
################################################################################ | |
### DEPENDENCY GRAPH | |
################################################################################ | |
class ComputationContainer(containers.DeclarativeContainer): | |
""" | |
Once we took the job of defining the interfaces, putting the pieces together | |
is straightforward. Note that this is highly customizable: | |
* The class can be extended, modified... to generate different comp. graphs | |
* Config can be set at graph construction and overriden at any time after | |
""" | |
my_config = providers.Configuration("conf") | |
img_server = providers.Singleton(ImageServer, my_config.img_dir) | |
# | |
process1 = providers.Factory(BasicImgProcessingNode, img_server, | |
img_operation1, my_config.p1.a, my_config.p1.b) | |
process2 = providers.Factory(BasicImgProcessingNode, img_server, | |
img_operation2, my_config.p2.a, my_config.p2.b) | |
fusion = providers.Factory(FusionNode, process1, process2, img_operation3) | |
main = providers.Callable(main_routine, operation=fusion) | |
################################################################################ | |
### MAIN ROUTINE | |
################################################################################ | |
if __name__ == "__main__": | |
# construct the graph | |
CONTAINER = ComputationContainer( | |
my_config={ | |
"img_dir": "/test/images", | |
"p1": {"a": 0, "b": 1000}, | |
"p2": {"a": 1001, "b": 2000} | |
} | |
) | |
print("\n\nTEST CONFIG OVERRIDE AND RETRIEVING IMAGE FROM CACHE:\n\n") | |
p1_low = CONTAINER.process1() | |
CONTAINER.my_config.override({"p1": {"a": 100000, "b": 2000000}}) | |
p1_high = CONTAINER.process1() | |
print("\n") | |
p1_low("hello1") | |
print("\n") | |
p1_low("hello2") | |
print("\n") | |
p1_high("hello1") | |
print("\n\nTEST FUSION NODE AND RETRIEVING IMAGE FROM CACHE:\n\n") | |
p_fusion = CONTAINER.fusion() | |
print("\n") | |
p_fusion("hello_fusion") | |
print("\n") | |
p_fusion("hello_fusion") | |
print("\n\nRUN MAIN ROUTINE:\n\n") | |
CONTAINER.main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment