Modèles de conception pour chaîner les méthodes de transformation des données en utilisant pandas

Question

Je reçois mensuellement un fichier CSV qui contient certaines colonnes. Peu importe les colonnes que je reçois, je devrais produire un CSV avec les colonnes C1, C2, C3, ... C29, C30 si possible + un fichier journal des étapes que j'ai suivies.

Je sais que l'ordre de mes transformations de données devrait être t1, t2, t3, t4, t5.

t1 génère les colonnes C8, C9, C12, C22 en utilisant C1, C2, C3, C4
t2 génère les colonnes C10, C11, C17 en utilisant C3, C6, C7, C8
t3 génère les colonnes C13, C14, C15, C16 en utilisant C5, C8, C10, C11, C22
t4 génère les colonnes C18, C19, C20, C21, C23, C24, C25 en utilisant C13, C15
t5 génère les colonnes C26, C27, C28, C29, C30 en utilisant C5, C19, C20, C21

Je ne peux pas contrôler les colonnes que je reçois dans mes données d'entrée.

Si mes données d'entrée ont les colonnes C1, C2, C3, C4, C5, C6, C7, je peux générer toutes les colonnes C1 ... C30.

Si mes données d'entrée ont les colonnes C1, C2, C3, C4, C5, C6, C7, C8, C10, C11, C17, je peux générer toutes les colonnes C1 ... C30, mais je devrais sauter t2, car ce n'est pas nécessaire

Si mes données d'entrée ont les colonnes C1, C2, C3, C4, C6, C7, je ne peux faire que t1, t2, t3, t4. Je ne peux pas exécuter t5, donc je devrais créer les colonnes C26, C27, C28, C29, C30 avec seulement des valeurs NaN et je devrais ajouter dans le journal "Impossible d'effectuer la transformation t5 car C5 est manquant. Les colonnes C26, C27, C28, C29, C30 sont remplies de valeurs NaN"

Mes t1, t2, t3, t4, t5 sont déjà créés, mais je ne sais pas comment organiser le code de manière élégante de sorte que les répétitions de code soient minimales.

J'ai dû développer mon code en très peu de temps. Par conséquent, tous mes méthodes t1, t2, t3, t4, t5 ressemblent à

def ti(df):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return df, "{} sauté, les colonnes de sortie {} existent déjà".format(inspect.stack()[0][3], output_cols)
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols):
        if missing_cols == []:
            // faire quelque chose
            log = "Transformation {} effectuée. Colonnes créées : {}".format(inspect.stack()[0][3], input_cols)
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = "Impossible d'effectuer la transformation {}. Les colonnes suivantes manquent : {}. Les colonnes {} sont remplies de valeurs NaN".format(inspect.stack()[0][3], missing_cols, output_cols)

De plus, j'utilise les fonctions de la manière suivante :

text = ""
df = pd.read_csv(input_path)
df, log_text = t1(df)
text = text + log_text + "\n"
df, log_text = t2(df)
text = text + log_text + "\n"
df, log_text = t3(df)
text = text + log_text + "\n"
df, log_text = t4(df)
text = text + log_text + "\n"
df, log_text = t5(df)
text = text + log_text + "\n"
df.to_csv("output_data.csv", index = False)
logging.info(text)

Comme vous pouvez le constater, mon code est laid et répétitif. J'ai maintenant le temps de le refactoriser, mais je ne sais pas quelle serait la meilleure approche. Je veux aussi que mon code soit extensible, car je pense également ajouter une transformation t6. Pouvez-vous m'aider en me donnant quelques orientations / motifs de conception que je pourrais suivre ? (Je suis également ouvert à l'utilisation d'autres bibliothèques python au-delà de pandas)

Answer 1

Étant donné que, en Python, les fonctions sont des objets de première classe, vous pourriez refactoriser votre code afin de généraliser vos fonctions t[i] en extrayant ce qui semble les différencier (la partie faire quelque chose), en faire une fonction auxiliaire et la traiter comme un paramètre.

Vous pouvez également éviter les répétitions lors de l'appel des fonctions (soit les t1, t2, etc. ou les versions helpers refacturées ci-après) en itérant sur une liste.

Enfin, l'utilisation des f-strings aide à rendre votre code un peu plus lisible.

Quelque chose comme ça :

# La fonction t prend un dataframe et une fonction en tant que paramètres
def t(df, do_stuff_func):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return (
            df,
            (
                f"{inspect.stack()[0][3]} sauté, "
                f"les colonnes de sortie {output_cols} existent déjà",
            ),
        )
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols)
        if missing_cols == []:
            # Appeler la fonction d'aide
            do_stuff_func()
            log = (
                f"Transformation {inspect.stack()[0][3]} effectuée."
                f"Créées les colonnes {input_cols}"
            )
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = (
                f"Impossible de réaliser la transformation {inspect.stack()[0][3]}"
                f"car les colonnes {missing_cols} sont manquantes. "
                f"{output_cols} sont remplis avec des valeurs NaN"
            )

# Définir les cinq nouvelles fonctions 'faire quelque chose'
def do_stuff1():
    pass
...
def do_stuff5():
    pass

# Stocker les fonctions
do_stuff_funcs = [do_stuff1, do_stuff2, do_stuff3, do_stuff4, do_stuff5]

# Appeler la fonction t en combinaison avec les fonctions do_stuff_funcs et df helpers
for do_stuff_func in do_stuff_funcs:
    df, log_text = t(df, do_stuff_func)
    text = text + log_text + "\n"

# Enregistrer les résultats
df.to_csv("output_data.csv", index = False)
logging.info(text)