transformer.py

import functools

import pandas as pd


class Transformer:

    __mappings__ = {}

    @classmethod
    def transform(cls, df, *, groupby=None, filter=None):
        transformer = cls(df, groupby=groupby, filter=filter)
        return transformer.apply_mappings()

    @classmethod
    def mapping(cls, target_key):
        if not isinstance(target_key, str):
            raise RuntimeError("`mapping` decorator requires pos arg `target key`")

        def _mapping(fn):

            cls.__mappings__[target_key] = fn

            @functools.wraps(fn)
            def wrapper(*args, **kwargs):
                return fn(*args, **kwargs)

            return wrapper

        return _mapping

    def __init__(self, df):
        self._df = df

    def _call(self, name):
        return self.__mappings__[name](self._df)

    def apply_mappings(self):
        return NotImplementedError


class PandasTransformer(Transformer):

    def __init__(self, df, *, groupby=None, filter=None):
        if filter:
            df = df[df[filter]]
        if groupby:
            df = df.groupby(groupby)
        self._output_df = pd.DataFrame(columns=self.__mappings__.keys())
        super().__init__(df)

    def apply_mappings(self):
        for key, value in self.__mappings__.items():
            self._output_df.loc[:, key] = value(self._df)
        return self._output_df



@Transformer.mapping('target')
def add(a, b):
    return a + b