StealthyCoder

So in the previous post I showed how to use the pattern of Suppliers and Consumers to structure any kind of input and output of data and structuring the processing. Well for bonus points we are going to make something fancy, a Chain. A Chain is nothing more than the series of steps that starts with one action and the result of that action will be the input for the next action and so on all the way till the end. Let us make one.

Chain

First off let us create a directory to house everything in chains/planet. Inside the chains directory place a __init__.py file with the following contents:

import abc
from typing import List, Dict, Callable

class SimpleChain(object):
    class Step(object):
        def __init__(self, func: Callable, pos_args: List, kw_args: Dict = None):
            self.func = func
            self.pos_args = pos_args
            self.kw_args = kw_args
        
        def execute(self):
            if len(self.pos_args) > 0 and self.kw_args is not None:
                return self.func(*self.pos_args, **self.kw_args)
            elif len(self.pos_args) > 0:
                return self.func(*self.pos_args)
            elif self.kw_args is not None:
                return self.func(**self.kw_args)
            else:
                return self.func()


    def __init__(self):
        self.steps: List[self.Step] = []

    def then(self, func: Callable, pos_args: List = [], kw_args: Dict = None):
        self.steps.append(self.Step(func, pos_args, kw_args))

    def execute(self):
        counter = 0
        for step in self.steps:
            if counter == 0:
                r = step.execute()
            else:
                step.pos_args = [r] + step.pos_args
                r = step.execute()
            counter += 1
            
        return r

The SimpleChain class has a method to add steps called then and an execute method that will execute them all. You can return the counter as well to verify all steps have been run. The inner class of Step is there to better manage everything and keep things nice and easy to comprehend. So the logic is that for each step beyond the first one we will add the results of the previous step as the first positional argument. I chose to prepend it rather than append it because we now can use unnamed functions called lambdas where it will always be the first positional argument.

Next up our PlanetWeightChain to see it in action. Create the PlanetWeightChain.py file inside the chains/planet directory with the following contents:

from chains import SimpleChain
from config import Config
from suppliers import AgnosticSupplier
from transformers.planet.WeightTransformer import WeightTransformer

class PlanetWeightChain(object):
    def __init__(self, config: Config, weight: float):
        self.chain = SimpleChain()
        self.chain.then(AgnosticSupplier, [config])
        self.chain.then(lambda x: x.get_supplier())
        self.chain.then(lambda x, y: WeightTransformer(x, y), [weight])
        self.chain.then(lambda x: x.transform())
        self.chain.then(lambda x: PlanetWeightChain.print(x))        

    def execute(self):
        self.chain.execute()

    @staticmethod
    def print(results):
        for result in results:
            print(result.kilograms, result.planet.name)

This Chain will first instantiate a SimpleChain and add the steps to it that seem awfully familiar to the main.py of the previous post. We want to pass in functions, but we don't necessarily want to make named functions for all these steps, therefore we make lambdas. The first positional argument will always be the result of the previous one. So the first step is to create an instance of AgnosticSupplier and that will be given to the second lambda and therefore x is the AgnosticSupplier instance and get_supplier will give us the supplier we need for the next step which is to create a WeightTransformer.

Remember the print at the end does not return anything so a None will be passed along if there were more steps.

Now to make our main.py slightly different:

from config import Config
from chains.planet.PlanetWeightChain import PlanetWeightChain

chain = PlanetWeightChain(Config(), 75)
chain.execute()

That is it. We have the full chain. It will automatically print the results at the end, naturally if you don't want it to print the results you could stop at the transform function and get the results back.

Closing thoughts

I really like this extra system. It seems like an overkill but this makes everything so flexible. You can insert extra steps, take an existing Chain and add or prepend steps to it wherever necessary. This system can make it so we can join the CommunicationTiming and WeightTransformer of previous post as examples into one chain and get the total data back.

#code #python