Implementing data sync from multiple source using OOP concepts

Question

I am writing code to implement data sync from multiple sources. I want code to be maintainable to add more data sources if required.

Here is the pattern I wish to follow:

class Source1Client:

    def get_domain(self):

        pass

    def get_endpoint(self):

        pass

    def get_headers(self):

        pass

    def make_post_request(self, url, data, headers):
        response = requests.post(url, data, headers=headers)
        return response

    def make_get_request(self, url, data):
        response = requests.get(url, data)
        return response

class Source2Client:

    def get_domain(self):

        pass

    def get_endpoint(self):

        pass

    def get_headers(self):

        pass

    def make_post_request(self, url, data, headers):
        response = requests.post(url, data, headers=headers)
        return response

    def make_get_request(self, url, data):
        response = requests.get(url, data)
        return response

    def call_stored_procedure(self, query):

        pass

    def run_query(self, query):

        pass

class BaseSync:

    def convert_data_to_serialized_objects(self, data, serializer, many=True):
        db_object, db_object_list = {}, []
        for i in data:
            for column, value in i.items():
                db_object = {**db_object, **{column: value}}
            db_object_list.append(db_object)

        schema = serializer(many=many)
        m = schema.dump(db_object_list)
        return m.data

    def get_serialized_data(
        self,source=None, query=None, proc=None, serializer=None, many=True, proc_kwargs={}, 
        endpoint=None, method=None, data=None, headers=None
    ):  
        if source == "source1":
            client = Source1Client()
        if source == "source2":
            client = Source2Client()
        if proc:
            formatted_query = proc.format(**proc_kwargs)
            raw_data = cllient.call_stored_procedure(formatted_query)
        if query:
            raw_data = client.run_query(query)
        else:
            if method == "get":
                raw_data == client.make_get_request(endpoint, data)
            if method == "post":
                raw_data == client.make_post_request(endpoint, data, headers)

        serialized_data = self.convert_data_to_serialized_objects(
            raw_data, serializer, many=many
        )
        return serialized_data


class DataSync(BaseSync):

    def sync_source1_data():

        source1_data = self.get_serialized_data(source="source1", endpoint="source1url/path")

        # other operations

    def sync_source2_data():

        source1_data = self.get_serialized_data(source="source2", endpoint="source1url/path")

        # other operations

In my code get_serialized_data is using multiple if-else cases, which will make code dirty and hard to read if more sources are added. Also, it has too many parameters, which is against clean code principles.

What can be a better implementation to make code cleaner, maintainable and readable?

Parvathirajan Natarajan · Accepted Answer · 2019-12-23 16:03:17Z

1

Use __init__ to handle bunch of arguments, then your self will take of passing arguments to your respective definition - that's the good way to use OOP in python.

class Foo:
    def __init__(self, value_a, value_b):
        self.a = value_a
        self.b = value_b

    def func(self):
        pass

To avoid multiple if else you can have list or dict and from there you can access. Hope this helps !!

answered Dec 23, 2019 at 16:03

Parvathirajan Natarajan

1193 bronze badges

Add a comment |

Stack Exchange Network

Implementing data sync from multiple source using OOP concepts

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Implementing data sync from multiple source using OOP concepts

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions