1
\$\begingroup\$

I just finished up this script that crawls hundreds of local git repos for csv files and then stores them into a database. I've tried to follow a "functional" paradigm for this script but am kind of confused with all the side-effects (printing, writing to db, shell subprocess). Definitely looking for a classic code-review with some comments regarding my logic, style, commenting, etc.

#!/usr/bin/env python
"""Script to stuff all historical data into Nightly.db."""
import sqlite3
import glob
import os
import subprocess
import re
import pandas as pd
from typing import List, Tuple, Callable


def generate_reader(repo_path: str) -> Tuple[Callable[[str], pd.DataFrame], Callable[[str], List]]:
    """
    Closure to maintain state of each repository.

    A replacement for a mini-class containing state information for each
    git-repo. This closure returns a tuple of functions.

    Args:
        repo_path (str) - absolute path to git-repo.

    Return:
       Tuple of functions
    """
    rep_hash = repo_hash(repo_path)
    rep_date = repo_date(repo_path)

    def read_and_annotate(file_path: str) -> pd.DataFrame:
        """Return a data-frame with identifying columns."""
        delim_data = (pd.read_csv(file_path, usecols=[i for i in range(0, 12)],
                                  error_bad_lines=False, warn_bad_lines=False,
                                  memory_map=True)
                      .assign(repo_root=repo_path,
                              repo_hash=rep_hash,
                              repo_date=rep_date,
                              full_path=file_path))

        # Let's only grab a few columns for now
        return delim_data[["repo_root", "repo_hash",
                           "repo_date", "full_path",
                           "simulation_alive_time"]]

    def repo_paths(pattern: str) -> List:
        """
        Return list of files matching glob pattern.

        Args:
            pattern (str) - glob pattern for files of interest.

        Return:
            List of absolute file-paths.
        """
        return glob.glob(f"{repo_path}/assessment/**/{pattern}", recursive=True)

    return (read_and_annotate, repo_paths)


def repo_hash(repo_path: str) -> str:
    """
    Return the current commmit hash of a repo.

    This function runs a shell subprocess to fetch the most-recent
    commit-hash from the git-repo provided.

    Args:
       repo_path (str): absolute path to git-repo

    Return:
       str - commit hash
    """
    # Use universal_newlines to get a string instead of bytes
    proc = subprocess.Popen(['git', 'ls-remote', repo_path, 'HEAD'],
                            shell=False, stdout=subprocess.PIPE,
                            universal_newlines=True)
    return re.match(r'(\S+)', proc.communicate()[0]).group(0)


def repo_date(repo_path: str) -> str:
    """
    Return the date-code of given file-path.

    This function uses a regexp to fetch the date-code (e.g. 20200305)
    from the provided repository path.

    Args:
        repo_path (str) - path to relevant git repository

    Return:
        str - unformatted date code
    """
    return re.search(r'[0-9]{8}', repo_path).group()


def crawl_repo(repo_path: str) -> None:
    """
    Wrapper function to write csv data into Nightly.db.

    This function will handle looping through a repo's respective csv
    files. It will also handle KeyErrors and OSErrors coming from the
    underlying pandas `read_csv()` function.

    Args:
        repo_path (str) - path to git repo containing csv files.

    Return:
        None - this function just launches the `write_to_db()` function.
    """
    reader, path_finder = generate_reader(repo_path)
    for data in path_finder("*_metrics.csv"):
        try:
            result = reader(data)
        except KeyError as e:
            reader_error(repo_path, data, e)
            continue
        except OSError as e:
            permission_error(repo_path, data, e)
            continue
        else:
            reader_success(result, repo_path, data)
            write_to_db(result)
    return None


def write_to_db(df):
    """
    Write a pandas dataframe to Nightly.db.

    Args:
        df (DataFrame) - pandas dataframe of csv file.

    Return:
        None
    """
    conn = sqlite3.connect("Nightly.db")
    df.to_sql('PERF', conn, if_exists='append', index=False)
    conn.commit()
    conn.close()
    return None


def stdout_printer(rp: str, fp: str, msg: str) -> None:
    """
    Generalized printer function.

    This function provides the base for all user consumed output in the
    script.

    Args:
        rp  (str) - absolute path to git repo
        fp  (str) - absolute path to current csv file
        msg (str) - custom message to output to the user

    Return:
        None
    """
    output = f"""
    {'-' * 72}
    repo_path:    {rp}
    file_path:    {os.path.basename(fp)}

    {msg}
    {'-' * 72}
    """
    print(output)
    return None


def permission_error(rp: str, fp: str, e: Exception) -> None:
    """
    Handle bad permissions on csv file.

    There are a few csv files that currently have permissions that
    prevent pandas from reading in the data. This function outputs
    the error and logs the offending file path.

    Args:
        rp  (str) - absolute path to git repo
        fp  (str) - absolute path to current csv file
        e   (Exception) - thrown by a try/catch block.

    Return:
        None
    """
    stdout_printer(rp, fp, f"Exception: {str(e)}")
    log_to_file(fp, 'bad_permissions.txt')
    return None


def reader_error(rp: str, fp: str, e: Exception) -> None:
    """
    Handle bad permissions on csv file.

    There are a few csv files that currently don't have the proper
    column names we need causing pandas to throw a KeyError.
    This function outputs the error and logs the offending file path.

    Args:
        rp  (str) - absolute path to git repo
        fp  (str) - absolute path to current csv file
        e   (Exception) - thrown by a try/catch block.

    Return:
        None
    """
    stdout_printer(rp, fp, f"Exception: {str(e)}")
    log_to_file(fp, 'key_error.txt')
    return None


def reader_success(df, rp: str, fp: str) -> None:
    """
    Output information pertaining to a successful data read-in.

    If pandas read-in is successful, we'll output the head of the
    dataframe.

    Args:
        df (DataFrame) - data-frame of csv file.
        rp (str) - absolute path to git repo.
        fp (str) - absolute path to csv file

    Return:
        None
    """
    data_preview = (df.head()
                    .to_string(col_space=3, justify='match-parent',
                               max_colwidth=10, index=False, line_width=82)
                    .replace('\n', '\n\t'))
    stdout_printer(rp, fp, f"Data:\n\t{data_preview}")
    return None


def log_to_file(fp: str, file_name: str) -> None:
    """
    Write file-path that caused exception to specified file.

    This impure function will log problematic file-paths that can be further
    examined.

    Args:
        fp        (str): problematic file-path to log.
        file_name (str): name of log file

    Return:
        None
    """
    with open(file_name, 'a') as log:
        log.write(f"{fp}\n")
    return None


def main():
    conn = sqlite3.connect("Nightly.db")
    c = conn.cursor()
    c.execute('CREATE TABLE IF NOT EXISTS PERF (repo_root text, \
    repo_hash text, repo_date text, \
    full_path text, simulation_alive_time numeric)')
    conn.commit()
    conn.close()

    bison_git_dirs = glob.glob("/projects/bison/git/bison_[0-9]*")
    for repo in bison_git_dirs:
        crawl_repo(repo)


if __name__ == '__main__':
    main()
\$\endgroup\$
0
\$\begingroup\$

I don't think you are doing functional programming in a good way here. If you have to jump through too many hoops to inject state into your functions, and have functions that have side-effects and explicitly return None, then this is probably not functional programming.

The easiest solution would probably be to write a Repo class, that consolidates all functions regarding to one repository:

class Repo:
    def __init__(self, path):
        self.path = path

    @property
    def hash(self):
        proc = subprocess.Popen(['git', 'ls-remote', self.path, 'HEAD'],
                                shell=False, stdout=subprocess.PIPE,
                                universal_newlines=True)
        return re.match(r'(\S+)', proc.communicate()[0]).group(0)

    @property
    def date(self):
        return re.search(r'[0-9]{8}', self.path).group()

    def files(self, pattern):
        return glob.glob(f"{self.path}/assessment/**/{pattern}", recursive=True)

    def read_csv_annotated(self, path) -> pd.DataFrame:
        """Read a CSV file and annotate it with information about the repo."""
        try:
            df = pd.read_csv(path, usecols=[i for i in range(0, 12)],
                             error_bad_lines=False, warn_bad_lines=False,
                             memory_map=True)
        except OSError as e:
            permission_error(repo_path, data, e)
            return
        df = df.assign(repo_root=self.path,
                       repo_hash=self.hash,
                       repo_date=self.date,
                       full_path=path)

        # Let's only grab a few columns for now
        try:
            return df[["repo_root", "repo_hash", "repo_date", "full_path",
                    "simulation_alive_time"]]
        except KeyError as e:
            reader_error(repo_path, data, e)

The actual writing to the DB should be left as the job of the consumer of this output:

def create_table(file_name):
    conn = sqlite3.connect(file_name)
    c = conn.cursor()
    c.execute('CREATE TABLE IF NOT EXISTS PERF (repo_root text, \
    repo_hash text, repo_date text, \
    full_path text, simulation_alive_time numeric)')
    conn.commit()
    conn.close()


if __name__ == "__main__":
    create_table("Nightly.db")
    bison_git_dirs = glob.glob("/projects/bison/git/bison_[0-9]*")
    for repo in map(Repo, bison_git_dirs):
        for csv_file in repo.files("*_metrics.csv"):
            write_to_db(repo.read_csv_annotated(csv_file))

Of course, if you really want to not use classes, that is also possible, but the latter part is still true. Only in functional programming you probably want an interface such that it works like this:

if __name__ == "__main__":
    create_table("Nightly.db")
    bison_git_dirs = glob.glob("/projects/bison/git/bison_[0-9]*")
    dfs = (annotate_df(read_file(csv_file), repo_info(repo_path))
           for repo_path in bison_git_dirs
           for csv_file in csv_files(repo_path))
    for df in dfs:
        write_to_db(df)
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thanks for the comments. For the code you provided, how would I then handle read_csv_annotated throwing an exception and then passing None into write_to_db() and breaking the code. I just want to skip and log that file if an exception is thrown. \$\endgroup\$ – dylanjm May 20 at 21:29
  • \$\begingroup\$ @dylanjm Just handle the exceptions in read_csv and the resulting None in annotate_df and in write_to_db, if you are talking about the second code. \$\endgroup\$ – Graipher May 20 at 21:41

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.