Skip to content

Instantly share code, notes, and snippets.

@andres-fr
Last active October 2, 2019 15:25
Show Gist options
  • Save andres-fr/05e6c67fb29bb210c711cc8060ca1b67 to your computer and use it in GitHub Desktop.
Save andres-fr/05e6c67fb29bb210c711cc8060ca1b67 to your computer and use it in GitHub Desktop.
Proper interfacing in Python
# -*- 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