Python pattern for a pipeline of tasks

Viewed 1694

I'm attempting to create a simple pipeline on a car object which is

  1. build wheels of the car
  2. add brakes to the car
  3. store the car

The code I've developed for this task is:

def build_brakes(car) :
    car.brakes = 'new high quality brakes'
    return car
        
class Car :
    def __init__(self , wheels):
        self.wheels = wheels
        self.jsonTransform = {}
        self.jsonTransform['wheels'] = wheels
        
    def toJson(self):
        return self.jsonTransform

def store(car):
    return 'car' , car , 'stored'
    
c = Car("wheels")
print(c.toJson())

car_with_brakes = build_brakes(c)
print(car_with_brakes.toJson())

stored = store(c)
print(stored)

Which prints:

{'wheels': 'wheels'}
{'wheels': 'wheels'}
('car', <__main__.Car object at 0x7f40a86bdcf8>, 'stored')

The line print(car_with_brakes.toJson()) prints {'wheels': 'wheels'}. Is there a pattern I can use which will generate the JSON for the car and brakes attribute and encapsulate the pipeline described above? I could create a new class called CarWithBrakes such as :

class CarWithBrakes(Car):
    
    def __init__(self , car, brakes):
        self.car = car
        self.brakes = brakes

    def toJson(self):
        self.jsonTransform = {}
        self.jsonTransform['wheels'] = self.car.wheels
        self.jsonTransform['brakes'] = self.car.brakes
        return self.jsonTransform
    
car_with_brakes = CarWithBrakes(c , 'new high qualiy brakes')

car_with_brakes.toJson()

which prints :

{'wheels': 'wheels', 'brakes': 'new high qualiy brakes'}

But this seems like too much for a seemingly simple problem?

1 Answers

You have several options.

One is to adjust build_brakes function:

# note brakes x breaks
def build_brakes(car) :
    car.brakes = 'new high quality brakes'
    car.jsonTransform['breaks'] = car.brakes
    return car

Another (IMO a cleaner one) is to modify toJson method:

# note brakes x breaks
class Car :
    def __init__(self , wheels):
        self.wheels = wheels
    def toJson(self):
        ret = {'wheels':self.wheels}
        if hasattr(self,"brakes"):
            ret['breaks'] = self.brakes
        return ret
Related