trjhtr

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 successfuln')
 return sf
 except:
 raise Exception('Connection to Salesforce failedn')


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