5
\$\begingroup\$

TLDR: I don't develop for a living. If someone could point me in the right direction to make my script more readable/pythonic I would appreciate the assistance.

In particular I'm struggling with the charts found in makeCharts() and code-flow throughout.

I have tried writing makeCharts() section as a for loop for workbook.add_chart({"type": "pie"}) as piechart, but received a _exit error which I guess makes sense as I'm not writing till the end of the function.

Background: We use a Security Incident Response Platform which has a available API but somewhat awful reporting. To offset this I have been writing a personal project script while learning pandas which pulls the last 30/60/90 day case metrics and creates some charts. The below is my current complete script sans identifying information.

Objectives: I really have been struggling to make this script:

A) More Pythonic (specifically makeCharts() and main()).

B) Faster I timed it at 42seconds across 350+ entries, which leads me to believe my main() loop could be improved.

C) Easier to read code-flow in the process of addressing the above.

# imports
# defined globals

start = time.time()
def main(api):
    # Finds all cases on SIRP endpoint
    # Creates all30, all60, all90 dictionaries from epoch
    all30, all60, all90 = {}, {}, {}
    response = api.find_cases(range="all", sort=[])

    def addToDict(arg, response, i):
        arg[(i)] = {
            "Name":
            response.json()[i]["title"],
            "ID":
            response.json()[i]["id"],
            "Owner":
            response.json()[i]["owner"],
            "Severity":
            response.json()[i]["severity"],
            "Created": (time.strftime(
                "%m/%d/%Y %H:%M:%S",
                time.gmtime(response.json()[i]["createdAt"] / 1000.0)))
        }
        if 'endDate' in response.json()[i]:
            arg[(i)].update({
                "Closed": (time.strftime(
                    "%m/%d/%Y %H:%M:%S",
                    time.gmtime(response.json()[i]["endDate"] / 1000.0))),
                "Resolution":
                response.json()[i]["resolutionStatus"]
            })
        return

    if response.status_code == 200:
        i = 0
        while i < len(response.json()):
            if (response.json()[i]["createdAt"] / 1000) > time.mktime(
                (datetime.date.today() -
                 datetime.timedelta(days=30)).timetuple()):
                addToDict(all30, response, i)
            if (response.json()[i]["createdAt"] / 1000) > time.mktime(
                (datetime.date.today() -
                 datetime.timedelta(days=60)).timetuple()):
                addToDict(all60, response, i)
            else:
                addToDict(all90, response, i)
            i += 1
    makeSS(all30, all60, all90)


def makeSS(all30, all60, all90):
    # creates (4) dataframes in pandas
    # all30/df1 - 30day dataframe from all30 dict
    # all60/df2 - 60day dataframe from all60 dict
    # all90/df3 - 90 day dataframe from all90 dict
    # df4 - separate sheet for chart data
    df1 = pd.DataFrame(
        all30,
        index=['Created', 'Severity', 'Owner', 'Name', 'Closed',
               'Resolution']).transpose()
    df2 = pd.DataFrame(
        all60,
        index=['Created', 'Severity', 'Owner', 'Name', 'Closed',
               'Resolution']).transpose()
    df3 = pd.DataFrame(
        all90,
        index=['Created', 'Severity', 'Owner', 'Name', 'Closed',
               'Resolution']).transpose()
    df4 = pd.DataFrame({
        'Created': (df1.count()['Created']),
        'Closed': (df1.count()['Closed']),
        'Owner': (df1['Owner'].value_counts().to_dict()),
        'Resolution': (df1['Resolution'].value_counts().to_dict()),
        'Severity': (df1['Severity'].value_counts().to_dict())
    })
    makeCharts(df1, df2, df3, df4)


def makeCharts(df1, df2, df3, df4):
    # create pie charts, xlsx and save locally
    with pd.ExcelWriter("foo.xlsx",
                        engine="xlsxwriter",
                        options={"strings_to_urls": False}) as writer:
        workbook = writer.book
        worksheet = workbook.add_worksheet("Summary Charts")
        worksheet.hide_gridlines(2)
        piechart = workbook.add_chart({"type": "pie"})
        piechart.set_title({'name': 'New vs. Closed Cases'})
        piechart.set_style(10)
        piechart.add_series({
            'name': 'Open vs. Closed Cases Last 30',
            'categories': '=Tracking!$B$1:$C$1',
            'values': '=Tracking!$B$2:$C$2',
        })
        worksheet.insert_chart("D2", piechart, {
            'x_offset': 25,
            'y_offset': 10
        })
        piechart1 = workbook.add_chart({"type": "pie"})
        piechart1.set_title({'name': 'Severities'})
        piechart1.set_style(10)
        piechart1.add_series({
            'name': 'Severity Last 30',
            'categories': '=Tracking!$A$2:$A$4',
            'values': '=Tracking!$F$2:$F$4',
        })
        worksheet.insert_chart("M2", piechart1, {
            'x_offset': 25,
            'y_offset': 10
        })
        piechart2 = workbook.add_chart({"type": "pie"})
        piechart2.set_title({'name': 'Resolution Last 30'})
        piechart2.set_style(10)
        piechart2.add_series({
            'name': 'Resolution Last 30',
            'categories': '=Tracking!$A$5:$A$6',
            'values': '=Tracking!$E$5:$E$6',
        })
        worksheet.insert_chart("D19", piechart2, {
            'x_offset': 25,
            'y_offset': 10
        })
        piechart3 = workbook.add_chart({"type": "pie"})
        piechart3.set_title({'name': 'Case Ownership Last 30'})
        piechart3.set_style(10)
        piechart3.add_series({
            'name': 'Case Ownership Last 30',
            'categories': '=Tracking!$A$7:$A$10',
            'values': '=Tracking!$D$7:$D$10',
        })
        worksheet.insert_chart("M19", piechart3, {
            'x_offset': 25,
            'y_offset': 10
        })
        df1.to_excel(writer,
                     sheet_name="Cases newer than 30 Days",
                     startrow=1,
                     header=False)
        formatSS((writer.sheets["Cases newer than 30 Days"]), workbook, df1)
        df2.to_excel(writer,
                     sheet_name="Cases older than 60 days",
                     startrow=1,
                     header=False)
        formatSS((writer.sheets["Cases older than 60 days"]), workbook, df2)
        df3.to_excel(writer,
                     sheet_name="Cases older than 90 days",
                     startrow=1,
                     header=False)
        formatSS((writer.sheets["Cases older than 90 days"]), workbook, df3)
        df4.to_excel(writer, sheet_name="Tracking")
    writer.save()
    send()


def formatSS(worksheet, workbook, arg):
    # add width to columns, filter, freeze
    worksheet.set_column("A:A", 3.5, workbook.add_format())
    worksheet.set_column("B:B", 17.25, workbook.add_format())
    worksheet.set_column("C:C", 10, workbook.add_format())
    worksheet.set_column("D:D", 10, workbook.add_format())
    worksheet.set_column("E:E", 100, workbook.add_format())
    worksheet.set_column("F:F", 17.25, workbook.add_format())
    worksheet.set_column("G:G", 11.25, workbook.add_format())
    worksheet.freeze_panes(1, 0)
    worksheet.autofilter("A1:G100")
    header_format = workbook.add_format({
        "bold": True,
        "text_wrap": True,
        "valign": "top",
        "fg_color": "#CCCCCC",
        "border": 1,
    })
    for col_num, value in enumerate(arg.columns.values):
        worksheet.write(0, col_num + 1, value, header_format)
    return worksheet, workbook


def send():
    # send the created xlsx
    msg = MIMEMultipart()
    msg["From"] = "Address@Domain.com"
    msg["To"] = sendTo
    msg["Subject"] = "Metrics"
    msg.attach(
        MIMEText("Attached are the requested case metrics in .XLSX format."))
    part = MIMEBase("application", "octet-stream")
    part.set_payload(open("Foo.xlsx", "rb").read())
    encoders.encode_base64(part)
    part.add_header("Content-Disposition",
                    'attachment; filename="Foo.xlsx"')
    msg.attach(part)
    smtp = smtplib.SMTP(smtp_server)
    smtp.starttls()
    smtp.sendmail(msg["From"], [msg["To"]], msg.as_string())
    smtp.quit()


main(api)
print('It took', time.time() - start, 'seconds.')
exit()
\$\endgroup\$
0
2
\$\begingroup\$

Complex refactoring/improvements:

Involved Refactoring techniques: Rename variable, Rename function, Extract variable, Extract function, Substitute Algorithm, Slide statements, Split phase (well-known classics https://refactoring.com/catalog/) + eliminating duplication and rearranging responsibility

Completely different OOP approach to your script, representing SIRPPipeline (SIRP Pipeline) which is initialized with passed api component/client, composed and running as a set of consecutive operations/phases:
loading data | making dataframes | making/building charts | send email.

In more details:

  • loading data. Covered by instance method _load_data which initiates API call to fetch the crucial source data and, if successful, populates target days dictionaries with records data (method _fill_day_dicts)

  • making dataframes. Covered by method make_dataframes which has a concrete single responsibility: construct 4 crucial dataframes (30 days, 60 days, 90 days, counts)

  • making/building charts. Covered by complex method make_charts. It initiates ExcelWriter object, declares inner functions for internal usage: _insert_pie_chart (builds and inserts pie chart to specified worksheet) and _df_days_to_excel (writes passed dataframe to writer object). _set_workbook_layout method establishes workbook/worksheet layout/format and saves it to worksheet.

  • send email. Static method send_email builds and sends an email related to "case metrics" pipeline event. The old function send was too generalized while send_email method reflects concrete purpose in scope of pipeline.


import datetime
import smtplib
import time
from email import encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import pandas as pd


class SIRPPipeline:
    TIME_FMT = "%m/%d/%Y %H:%M:%S"
    DF_INDEX = ['Created', 'Severity', 'Owner', 'Name', 'Closed', 'Resolution']

    def __init__(self, api):
        """Security Incident Response Platform prosessing pipeline.
           Accepts API object on initialization phase.
        """
        self._api = api
        self._all30_dict = {}
        self._all60_dict = {}
        self._all90_dict = {}

        self._df_30days = None
        self._df_60days = None
        self._df_90days = None
        self._df_counts = None
        self._dataset = None

    def _load_data(self):
        # Finds all cases on SIRP endpoint
        self._api_response = self._api.find_cases(range="all", sort=[])

        if self._api_response.status_code == 200:
            self._dataset = self._api_response.json()
            self._fill_day_dicts()

    @staticmethod
    def _add_record(days_dict, record, key):
        days_dict[key] = {
            "Name": record["title"],
            "ID": record["id"],
            "Owner": record["owner"],
            "Severity": record["severity"],
            "Created": (time.strftime(
                SIRPPipeline.TIME_FMT,
                time.gmtime(record["createdAt"] / 1000.0)))
        }
        if 'endDate' in record:
            days_dict.update({
                "Closed": (time.strftime(
                    SIRPPipeline.TIME_FMT,
                    time.gmtime(record["endDate"] / 1000.0))),
                "Resolution": record["resolutionStatus"]
            })

    def _fill_day_dicts(self):
        today = datetime.date.today()

        for i, record in enumerate(self._dataset):
            if (record["createdAt"] / 1000) > time.mktime(
                    (today - datetime.timedelta(days=30)).timetuple()):
                self._add_record(self._all30_dict, record, key=i)

            elif (record["createdAt"] / 1000) > time.mktime(
                    (today - datetime.timedelta(days=60)).timetuple()):
                self._add_record(self._all60_dict, record, key=i)

            else:
                self._add_record(self._all90_dict, record, key=i)

    def make_dataframes(self):
        """Creates (4) pandas dataframes:
        - df_30days dataframe from all30 dict
        - df_60days dataframe from all60 dict
        - df_90days dataframe from all90 dict
        - df_counts - separate sheet for chart data
        """
        self._df_30days = pd.DataFrame(self._all30_dict, index=SIRPPipeline.DF_INDEX).transpose()
        self._df_60days = pd.DataFrame(self._all60_dict, index=SIRPPipeline.DF_INDEX).transpose()
        self._df_90days = pd.DataFrame(self._all90_dict, index=SIRPPipeline.DF_INDEX).transpose()
        self._df_counts = pd.DataFrame({
            'Created': (self._df_30days.count()['Created']),
            'Closed': (self._df_30days.count()['Closed']),
            'Owner': (self._df_30days['Owner'].value_counts().to_dict()),
            'Resolution': (self._df_30days['Resolution'].value_counts().to_dict()),
            'Severity': (self._df_30days['Severity'].value_counts().to_dict())
        })

    @staticmethod
    def _set_workbook_layout(workbook, worksheet, df):
        # add width to columns, filter, freeze
        worksheet.set_column("A:A", 3.5, workbook.add_format())
        worksheet.set_column("B:B", 17.25, workbook.add_format())
        worksheet.set_column("C:C", 10, workbook.add_format())
        worksheet.set_column("D:D", 10, workbook.add_format())
        worksheet.set_column("E:E", 100, workbook.add_format())
        worksheet.set_column("F:F", 17.25, workbook.add_format())
        worksheet.set_column("G:G", 11.25, workbook.add_format())
        worksheet.freeze_panes(1, 0)
        worksheet.autofilter("A1:G100")

        header_format = workbook.add_format({
            "bold": True,
            "text_wrap": True,
            "valign": "top",
            "fg_color": "#CCCCCC",
            "border": 1,
        })

        for col_num, value in enumerate(df.columns.values, 1):
            worksheet.write(0, col_num, value, header_format)

    def make_charts(self):

        def _insert_pie_chart(wbook, wsheet, title, cell_pos, series):
            piechart = wbook.add_chart({"type": "pie"})
            piechart.set_title({'name': title})
            piechart.set_style(10)
            piechart.add_series(series)
            wsheet.insert_chart(cell_pos, piechart, {
                'x_offset': 25,
                'y_offset': 10
            })

        def _df_days_to_excel(writer, sheet_name, df_days):
            df_days.to_excel(writer, sheet_name=sheet_name, startrow=1, header=False)
            self._set_workbook_layout(writer.book, (writer.sheets[sheet_name]), df_days)

        # create pie charts, xlsx and save locally
        with pd.ExcelWriter("foo.xlsx",
                            engine="xlsxwriter",
                            options={"strings_to_urls": False}) as writer:
            workbook = writer.book
            worksheet = workbook.add_worksheet("Summary Charts")
            worksheet.hide_gridlines(2)

            _insert_pie_chart(workbook, worksheet, title='New vs. Closed Cases', cell_pos='D2', series={
                'name': 'Open vs. Closed Cases Last 30',
                'categories': '=Tracking!$B$1:$C$1',
                'values': '=Tracking!$B$2:$C$2',
            })
            _insert_pie_chart(workbook, worksheet, title='Severities', cell_pos='M2', series={
                'name': 'Severity Last 30',
                'categories': '=Tracking!$A$2:$A$4',
                'values': '=Tracking!$F$2:$F$4',
            })
            _insert_pie_chart(workbook, worksheet, title='Resolution Last 30', cell_pos='D19', series={
                'name': 'Resolution Last 30',
                'categories': '=Tracking!$A$5:$A$6',
                'values': '=Tracking!$E$5:$E$6',
            })
            _insert_pie_chart(workbook, worksheet, title='Case Ownership Last 30', cell_pos='M19', series={
                'name': 'Case Ownership Last 30',
                'categories': '=Tracking!$A$7:$A$10',
                'values': '=Tracking!$D$7:$D$10',
            })

            _df_days_to_excel(writer, sheet_name="Cases newer than 30 Days", df_days=self._df_30days)
            _df_days_to_excel(writer, sheet_name="Cases older than 60 days", df_days=self._df_60days)
            _df_days_to_excel(writer, sheet_name="Cases newer than 90 Days", df_days=self._df_90days)

            self._df_counts.to_excel(writer, sheet_name="Tracking")
            writer.save()

    @staticmethod
    def send_mail():
        # send_mail the created xlsx
        msg = MIMEMultipart()
        msg["From"] = "Address@Domain.com"
        msg["To"] = send_to  # consider `send_to` declaration
        msg["Subject"] = "Metrics"
        msg.attach(
            MIMEText("Attached are the requested case metrics in .XLSX format."))
        part = MIMEBase("application", "octet-stream")
        part.set_payload(open("Foo.xlsx", "rb").read())
        encoders.encode_base64(part)
        part.add_header("Content-Disposition",
                        'attachment; filename="Foo.xlsx"')
        msg.attach(part)
        smtp = smtplib.SMTP(smtp_server)  # consider `smtp_server` declaration
        smtp.starttls()
        smtp.sendmail(msg["From"], [msg["To"]], msg.as_string())
        smtp.quit()

    def run(self):
        self._load_data()
        self.make_dataframes()  # may be protected
        self.make_charts()      # may be protected
        self.send_mail()


def main(api):
    pipe = SIRPPipeline(api)
    pipe.run()


# api initialization
# ...
start = time.time()
main(api)
print('It took', time.time() - start, 'seconds.')
exit()

As for time performance, it requires an appropriate testable sample data for realistic measurements.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ Thank you very much for this well thought out answer, I really appreciate it! \$\endgroup\$ – ImNotLeet Oct 21 '19 at 13:22
  • 1
    \$\begingroup\$ @ImNotLeet, you're welcome. It's great if it's helped you to find the "way". \$\endgroup\$ – RomanPerekhrest Oct 21 '19 at 14:02

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.