Search code examples
pythonoopdecoupling

Pythonic Decoupling of Input and Resulting Object


I have various input streams that give information I want packaged into an object that I can then pass around to separate functions:

class FruitBasket:
    def __init__(self, a, b, c, **kwargs):
        self.a = a
        self.b = b
        self.c = c

However, each data stream names the data I am looking for differently. StreamA's input looks like this:

{"apple" : 3, "banana" : 2, "citrus" : 0}

and StreamB's input looks like this:

{"fruit1" : 1, "fruit2" : 7, "fruit3" : 4, "dog" : 21}

Currently my solution to this is to have a separate class that maps the information between the input data and the object output:

class StreamAMap(Enum):
    a = "apple"
    b = "banana"
    c = "citrus"

class StreamBMap(Enum):
    a = "fruit2"
    b = "fruit1"
    c = "fruit3"

I then pass both this "map" class, the target object, and input data into a function:

def translate(map_enum, target, data):
    translation = {}

    for item in map_enum:
        translation[item.name] = data[item.value]

    return(target(**translation))

data_a = {"apple" : 3, "banana" : 2, "citrus" : 0}
data_b = {"fruit1" : 1, "fruit2" : 7, "fruit3" : 4, "dog" : 21}

basket_1 = translate(StreamAMap, FruitBasket, data_a)
basket_2 = translate(StreamBMap, FruitBasket, data_b)

It works, but I am wondering if there is a more Pythonic way of achieving this.

edit: The goal here is to keep FruitBasket completely decoupled from whatever stream may produce data that end up in it, i.e. every stream knows what information needs to go into FruitBasket, but regardless of what streams are added or removed in the future, I don't have to actually change anything about the FruitBasket class, but rather only ensure that each stream is returning a valid FruitBasket to the main application. My current implementation requires only a "map" class be defined for each new stream I add, and I'm wondering if there is a "Pythonic" (built-in) way of doing this rather than rolling my own translate function to do it.

To clarify what I mean by "Pythonic", the "Pythonic" way of determining the mean of a list is sum(list) / len(list) rather than iterating through the list to increment a counter variable and add each item to a sum variable.


Solution

  • So I've figured out a way to do this very simply and concisely:

    class FruitBasket:
        def __init__(self, a, b, c, **kwargs):
            self.a = a
            self.b = b
            self.c = c
    
    class FruitBasketIntermediateA:
        def __init__(self, apple, banana, citrus, **kwargs):
            self.a = apple
            self.b = banana
            self.c = citrus
    
    class FruitBasketIntermediateB:
        def __init__(self, fruit1, fruit2, fruit3, **kwargs):
            self.a = fruit2
            self.b = fruit1
            self.c = fruit3
    
    data_a = {"apple" : 3, "banana" : 2, "citrus" : 0}
    data_b = {"fruit1" : 1, "fruit2" : 7, "fruit3" : 4, "dog" : 21}
    
    my_fb_a = FruitBasket(**vars(FruitBasketIntermediateA(**data)))
    my_fb_b = FruitBasket(**vars(FruitBasketIntermediateB(**data)))
    

    If I was consuming hundreds of thousands of messages per second I would likely have to use something more performant, like the method in the original question but with dicts instead of enums, or using jsbueno's method, however the hard limit is orders of magnitude less so this works well for my needs and keeps things decoupled and concise.