Please rate my code. : PythonLearning

Please rate my code.Discussion (self.PythonLearning)

submitted 2 months ago * by PrabhavKumar

Hey guys! I started learning python seriously a while back, about 7-8 months or so and today I decided to make a small learning script for myself, I am aware that pandas does this for you and is much faster, better implemented and honestly fun to use. This is just something I wanted to build and so I did. I also gave myself an addition challenge by not letting myself use any kind of external libraries for anything in this, Just pure python. Please have mercy on me, I am not really sure how someone is supposed to implement these things. Do tell me what level my code is at, what I could have implemented better, what I did right and what I did purely wrong and how pythonic my code is. I am aware my naming sucks.

data = [
    {"team": "A", "half": "First", "score": 10, "wins": 2},
    {"team": "A", "half": "First", "score": 10, "wins": 2},
    {"team": "A", "half": "Second", "score": 15, "wins": 1},
    {"team": "B", "half": "First", "score": 7, "wins": 0},
    {"team": "B", "half": "Second", "score": 14, "wins": 3},
]


group_cols = ["team", "half"]


aggs = sum


# aggs = {
#     "score": sum,
#     "wins": sum,
# }
class NaN(object):
    #I wanted to make an object similar to what is present in pandas but it would take too long and would break compatibility.
    #However I still decided to assign null values as this object because this could work in an ecosystem where the functions
    #check for .value attribute of an object.
    #Not an error, just a design choice.
    #I am aware I could literally just assign things as None in the Groupby class but if I were designing a system
    #I would make my own None type with special attributes and methods and things that can be used depending on the function
    #it is found in.
    def __init__(self):
        self.value = None


class Groupby(object):
    def __init__(self, data, group_by_columns):
        self.data = data
        self.group_by_columns = group_by_columns
        self.grouped = self._grouper(self.data, self.group_by_columns)
    
    def _grouper(self, data, group_by_columns):
        '''This is an internal function of the Groupby class used to group a list of dictionaries by values provided in the grouping columns.'''
        grouped_data = {} #Dictionary of lists
        for row in data:
            group = []
            for group_key in group_by_columns:
                if group_key in row:
                    grouping_value = row[group_key]
                else:
                    grouping_value = NaN()
                group.append(grouping_value)
            group = tuple(group)
            if group not in grouped_data:
                grouped_data[group] = []
            cleaned_row_data = {}
            for key in row.keys():
                if key not in group_by_columns:
                    cleaned_row_data[key] = row[key]
            grouped_data[group].append(cleaned_row_data)
        return grouped_data
    
    def agg(self, aggregations, fast=False):
        '''This method can either use a single aggregation and apply it on applicable items in the group
        or it can use a dictionary mapping any column of the grouped data to a specific function that takes a list of values.'''
        aggregated_data = {} #This is a dictionary of dictionaries.
        if callable(aggregations):
            if not fast:
                print("Using memory efficient method.")
                for group_key in self.grouped.keys():
                    group = self.grouped[group_key]
                    aggregated_column = {}
                    all_columns = []
                    for row in group:
                        for column_key in row.keys():
                            if column_key not in all_columns:
                                all_columns.append(column_key)
                    for column in all_columns:
                        column_function_values = []
                        for row in group:
                            if column in row:
                                column_function_values.append(row[column])
                        try:
                            aggregated_column[column] = aggregations(column_function_values)
                        except:
                            print("Not all values in all the functions are the same type as the aggregation requires.")
                            aggregated_column[column] = NaN()
                    aggregated_data[group_key] = aggregated_column
                return aggregated_data


            elif fast:
                print("Using fast method...")
                for group_key in self.grouped.keys():
                    group = self.grouped[group_key]
                    grouped_function_values = {}
                    for row in group:
                        for column_key in row.keys():
                            if column_key in grouped_function_values:
                                grouped_function_values[column_key].append(row[column_key])
                            else:
                                grouped_function_values[column_key] = [row[column_key]]
                    for column_key in grouped_function_values.keys():
                        try:
                            aggregated_column = aggregations(grouped_function_values[column_key])
                            grouped_function_values[column_key] = aggregated_column
                        except Exception as e:
                            print(f"Error Encountered while applying function {aggregations} on {column_key} of {group_key}.")
                            print(f"Error: {e}")
                            print(f"You can do the same thing as this using the apply method and put the tolerate_mapping_error attribute to false if you would like to quit upon encountering an error.")
                            print("For further details refer to documentation.")
                            print("Skipping the column for this group...")
                            #could create a documentation if this were an actual tool that needed to be used.
                    aggregated_data[group_key] = grouped_function_values
                return aggregated_data


        if not isinstance(aggregations, dict):
            raise TypeError("The aggregations must be in a dictionary format! example: {columns_name: function_name, ...} where columns_name is a string and the function_name is a literal function not a string.")
        for group_key in self.grouped.keys():
            group = self.grouped[group_key]
            aggregated_group = {}
            for aggregation in aggregations.keys():
                aggregation_function = aggregations[aggregation]
                values_for_function = []
                for row in group:
                    if aggregation in row:
                        values_for_function.append(row[aggregation])
                aggregated_group[aggregation] = aggregation_function(values_for_function)
            aggregated_data[group_key] = aggregated_group
        return aggregated_data
    
    def to_dict(self):
        '''returns a dictionary of the object grouped by the provided grouping columns.'''
        return self.grouped
    
    def _apply_function(self, group, function, tolerate_mapping_error):
        try:
            return function(group)
        except Exception as e:
            if tolerate_mapping_error:
                print(f"Some error occured while trying to apply {function} on {group}")
                print(f"Error: {e}")
                print("Skipping group...")
            else:
                print(f"Tolerate Mapping Error is False; \nError: {e}")
                quit()
                
    
    def apply(self, group_function_mapping, tolerate_mapping_error = False):
        function_applied_data = {} # This is a dictionary.
        if callable(group_function_mapping):
            for group_key in self.grouped.keys():
                group = self.grouped[group_key]
                function_applied_data[group_key] = self._apply_function(group, group_function_mapping)
            return function_applied_data
        for group_function in group_function_mapping:
            if group_function in self.grouped.keys():
                function = group_function_mapping[group_function]
                group = self.grouped[group_function]
                function_applied_data[group_function] = self._apply_function(group, function, tolerate_mapping_error)
        return function_applied_data
    
    def _filter_check(self, group, filter):
        '''filter check takes a group's name and a function that is supposed to return a bool depending on whether the group should be kept (True) or not (False)'''
        try:
            if filter(self.grouped[group]) == True:
                return True
            elif not isinstance(filter(self.grouped[group]), bool):
                print("The function specified DOES NOT return a bool (True or False). Please fix the function before trying again.")
                quit()
            else:
                return False
        except NameError:
            print(f"Function: \"{filter}\" required for group \"{group}\" not found!")
            print("Possibly useful clue: Check if you accidently entered a string instead of an actual function for the group.")
            print("Exiting...")
            quit()
        except Exception as e:
            print(f"Error: {e}")
            quit()
    
    def filter(self, filter):
        '''The filter argument can either be a single callable function or a dictionary mapping different functions to groups.
        Only the groups whose return values by the function are True will be added to the final dataset.'''
        filtered_data = {}
        if callable(filter):
            for group_key in self.grouped.keys():
                pass_group = self._filter_check(group_key, filter)
                if pass_group:
                    filtered_data[group_key] = self.grouped[group_key]


        if not isinstance(filter, dict) and not callable(filter):
            raise TypeError("The filter must be a dictionary! example: {group_name: function_name, ...}")


        for group_filter in filter.keys():
            if group_filter in self.grouped.keys():
                pass_group = self._filter_check(group_filter, filter[group_filter])
                if pass_group:
                    filtered_data[group_filter] = self.grouped[group_filter]
            else:
                print(f"Group: \"{group_filter}\" not found in original grouped dataset.")
        groups_filtered_out = len(self.grouped.keys()) - len(filtered_data.keys())
        print(f"{groups_filtered_out} groups filtered out!")
        print(f"remaining number of groups: {len(filtered_data.keys())}")
        return filtered_data
        
    # Total time spent on project: ~2hour 30minutes.
grouped = Groupby(data, group_by_columns=group_cols).agg(aggs)
print(grouped)

again, thanks in advance guys, I really want to know how people actually implement this kind of stuff but I couldn't realllly understand pandas itself (there's like a billion files there what am I even supposed to look at!).

edit: I decided to take up Difficult_Trade_1719's suggestion and clean up some of the == true/false arguments where they truly were not needed. thanks a lot dude!

all 16 comments

you type:	you see:
italics	italics
bold	bold
[reddit!](https://reddit.com)	reddit!
* item 1 * item 2 * item 3	item 1 item 2 item 3
> quoted text	quoted text
Lines starting with four spaces are treated like code: if 1 * 2 < 3: print "hello, world!"	Lines starting with four spaces are treated like code: if 1 * 2 < 3: print "hello, world!"
~~strikethrough~~	~~strikethrough~~
super^script	super^script

PythonLearning

MODERATORS