I am using Python 3 and pandas 0.24.2 to do some data processing and ETL flows. I have followed this pattern a couple of times and it is really bothering me but I am not sure which direction to go to majorly improve it. Any advice for what to do differently is greatly appreciated. This is a toy example but the logic is the same. The main method is below. How can the structure of the class be improved?

import pandas as pd
from collections import deque
from collections import defaultdict
from collections import namedtuple
from src.query_helper import postgres_query_helper as pgqh

class ComputeNetProfitSummary():
    def __init__(self, date=None, factory_id=None):
        self.date = date
        self.factory_id = factory_id
        self.pg_query_helper = pgqh.PostgresQueryHelper()

    def build_sql_query(self):
        sql_query = f"""Select * FROM factory_table
            WHERE date='{self.date}' 
            AND factory_id={self.factory_id}
        return sql_query

    def create_orders_df(self):
        """Create orders Data Frame"""
        if self.date and self.factory_id:  
            sql_query = self.build_sql_query()       
            df = self.pg_query_helper.execute_sql_return_df(sql_query)
            return df

    def compute_complex_metrics_on_df(self):
        """Some complicated logic using FIFO attribution"""
        ComplexData = namedtuple('ComplexData', ['ColumnNamesA', ..., 'ColumnNamesZ'])      
        all_data = []  

        for row in self.orders_df.iterrows():
            """Do complicated stuff"""
            transform_row = # some computations on row
            complex_output = ComplexData(transform_row)

        complex_df = pd.DataFrame(all_data)
        return complex_df 

    def compute_summary_statistics(self):
        """Compute summary data"""
        summary_data = defaultdict(dict)
        # Create summary data, I am fine with this
        return summary_data 

    def compute_summary_data(self):
        self.orders_df = self.create_orders_df()
        self.complex_df = compute_complex_metrics_on_df()
        self.summary_statistics = compute_summary_statistics()

if __name__ == '__main__':
    cnp = ComputeNetProfitSummary(date='2019-01-01', factory_id='A')

    summary_data = cnp.compute_summary_data()

    for key, value in summary_data.items():

NOTE: I am one of the only Python developers on my team so I do not get a lot of feedback on this type of code very often. Any feedback or constructive criticism is welcome, from design patterns to PEP8.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.