1
\$\begingroup\$

Each day I receive a CSV that contains data concerning a patient record and a related appointment record, each on the same row of the CSV. My code is designed to:

  1. Parse the entered CSV for two separate but related records per row
  2. Validate some data (email)
  3. Generate some data (duration)
  4. Insert the patient record - this returns the ID of the newly created record
  5. Add that ID to the appointment record
  6. Insert the appointment record

I've left out the parts where I am actually inserting the records because I've been testing a lot and don't feel like going in and deleting thousands of records each day, but ultimately I know the code works. I feel confident in some parts of the code but mostly I do not feel like this is the most sensible way to approach it - primarily the size of PatientHandler.run_csv() and the heavy use of kwargs.pop() cause me to hesitate.

main.py

from patient_handler import PatientHandler
from simple_salesforce import Salesforce
from secret import * # salesforce credentials


def connectToSF():
    try:
        sf = Salesforce(username=sandbox_username, domain='test',
                        password=sandbox_password, security_token=sandbox_token,
                        client_id='PythonUploader')
        if sf:
            print('Connection to Salesforce successful\n')
            return sf
    except:
        raise Exception('Connection to Salesforce failed\n')


def main():
    csv_path = input('Enter full CSV Path:\n')
    sf = connectToSF()
    p = PatientHandler(sf)
    appointments = p.run_CSV(open(csv_path, newline='\n'))
    appointments = p.dedupe(appointments)


if __name__ == '__main__':
    main()

patient_handler.py

fieldMap.py has just a single dict (fieldMap) with the names of the Salesforce fields mapped to the csv column headers I am looking for. The CSV does not change much but the goal was to make this agnostic of the number and order of columns.

from validate_email import validate_email
from fieldMap import fieldMap
from patient import Patient
from appointment import Appointment
import csv


class PatientHandler(object):
    def __init__(self, sf_instance):
        self.__record_count = 0
        self.__unique_appointments = 0
        self.__valid_email_count = 0
        self.sf = sf_instance

    @staticmethod
    def dedupe(appointments):
        unique_ids = []
        final_list = []
        for appt in appointments:
            if appt['ApptID'] not in unique_ids:
                unique_ids.append(appt['ApptID'])
                final_list.append(appt)

        print('{0} unique appointments found'.format(str(len(final_list))))
        return final_list

    def run_CSV(self, ed):
        with ed as ed_file:
            patient_appts = []
            header_row = True  # prepare to skip to data rows
            ed_reader = csv.reader(ed_file, delimiter=',', quotechar='"')

            for row in ed_reader:
                if not row[0] and not row[1]:  # blank or end of document
                    print('\n%i records found' % self.__record_count
                          + '\n%i valid emails found' % self.__valid_email_count)
                    break
                elif header_row:
                    headers = list(row)
                    i_fname = headers.index(fieldMap['FirstName'])
                    i_mname = headers.index(fieldMap['MiddleName'])
                    i_lname = headers.index(fieldMap['LastName'])
                    i_email = headers.index(fieldMap['Email'])
                    i_start = headers.index(fieldMap['StartTime'])
                    i_end = headers.index(fieldMap['EndTime'])
                    i_location = headers.index(fieldMap['Location'])
                    i_type = headers.index(fieldMap['Type'])
                    i_appt_id = headers.index(fieldMap['ApptID'])
                    i_provider = headers.index(fieldMap['Provider'])
                    header_row = False
                else:
                    valid_email = validate_email(row[i_email])

                    a = Appointment(Type=row[i_type],
                                    Location=row[i_location],
                                    StartTime=row[i_start],
                                    EndTime=row[i_end],
                                    Provider=row[i_provider],
                                    sf_instance=self.sf)

                    p = Patient(FirstName=row[i_fname],
                                MiddleName=row[i_mname],
                                LastName=row[i_lname],
                                Email=row[i_email],
                                ValidEmail=valid_email,
                                LeadSource='Call Center',
                                Appointment=a.__dict__,
                                sf_instance=self.sf)

                    if valid_email:
                        self.__valid_email_count += 1
                    self.__record_count += 1

                    patient_id = p.insert()
                    a.setSFID(patient_id)
                    a.insert()

                    entry = {'ApptID': row[i_appt_id], 'Patient': p.__dict__}
                    patient_appts.append(entry)

            return patient_appts

patient.py

import json


class Patient(object):
    def __init__(self, **kwargs):
        self.firstName = str(kwargs.pop('FirstName')).title()
        self.middleName = str(kwargs.pop('MiddleName')).title()
        self.lastName = str(kwargs.pop('LastName')).title()
        self.email = kwargs.pop('Email')
        self.validEmail = kwargs.pop('ValidEmail')
        self.appointment = kwargs.pop('Appointment')
        self.leadsource = kwargs.pop('LeadSource')
        self.sf = kwargs.pop('sf_instance')

    def insert(self):
        payload = self.__dict__
        payload.pop('appointment')
        if not payload['validEmail']:
            payload.pop('email')  # remove field with no data
        payload.pop('validEmail')
        # perform insert operation, return id to be used in appointment record
        # self.setSFID(self.sf.Lead.create(payload['id']))
        # for now, simulate record insert and assignment of record ID
        self.setSFID('00Q123456789123')
        print(json.dumps(payload, indent=4))

    def setSFID(self, the_id):
        self.sfid = the_id

appointment.py

from datetime import datetime, timedelta
import json


class Appointment(object):
    def __init__(self, sf_instance, **kwargs):
        self.provider = kwargs.pop('Provider')
        self.type = kwargs.pop('Type')
        self.location = kwargs.pop('Location')
        self.start = kwargs.pop('StartTime')
        self.end = kwargs.pop('EndTime')
        self.duration = self.getDuration(self.start, self.end)
        self.sf = sf_instance

    def getDuration(self, start_time, end_time):
        fmt = '%I:%M %p'
        tdelta = datetime.strptime(
            end_time, fmt) - datetime.strptime(start_time, fmt)
        # time will always be under 1 hour
        duration = str(tdelta).split(':')[1]
        return duration

    def insert(self):
        # perform insert operation, return id to be used in appointment record
        # self.setSFID(self.sf.Inquiry__c.create(payload['appointment']))
        # for now, simulate record insert
        payload = self.__dict__
        print(json.dumps(payload, indent=4))

    def setSFID(self, the_id):
        self.sfid = the_id
\$\endgroup\$
  • 1
    \$\begingroup\$ I'm pretty sure I'm inserting the records before deduping, so there's that :( \$\endgroup\$ – JaredT Jun 22 '18 at 14:22
  • \$\begingroup\$ Yes you are.... \$\endgroup\$ – Graipher Jun 22 '18 at 15:18

Your Answer

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

Browse other questions tagged or ask your own question.