#🔒 Code Review Sending api records to a rest server.
onyx lava
import time
from tqdm.autonotebook import tqdm
from Backfill import backfill_main
from Cache import caching_main
from Timing import get_time
from Util import data_objects
def main():
print('\n\nHello!')
backfill_amounts = {}
while True:
for object_name in tqdm(data_objects.keys(), desc='Iterate over salesforce objects'):
try:
backfill_amounts = process(backfill_amounts, object_name)
print('*********\n\n')
except Exception as e:
print(e)
print('backfills', backfill_amounts)
print('done and sleeping\n\n')
# Sleep for 30 min
time.sleep(3600 / 2)
print('I woke up!')
@get_time
def process(backfill_amounts, object_name):
print(f"\nLets sync salesforce {object_name}")
caching_main(object_name)
return backfill_main(object_name, backfill_amounts, False)
if __name__ == "__main__":
main()
#Util
from datetime import datetime
import mysql.connector
from dateutil import parser, tz
from simple_salesforce import Salesforce
import configparser
config = configparser.RawConfigParser()
config.read('config.properties')
def get_prop(prop):
return config['DEFAULT'][prop]
def get_sf():
return Salesforce(username=get_prop('sf_username'),
password=get_prop('sf_password'),
security_token=get_prop('sf_security_token'),
client_id=get_prop('sf_client_id'))
def get_mysql(prod=False):
return mysql.connector.connect(
allow_local_infile=True,
host=get_prop('db_host_prod') if prod else get_prop('db_host'),
user=get_prop('db_user'),
password=get_prop('db_password'),
database=get_prop('db_database')
)
def fix_date(sf_date) -> str:
time_zone = tz.gettz('US/Pacific')
converted_data = parser.parse(sf_date).astimezone(time_zone) \
if isinstance(sf_date, str) \
else datetime.fromtimestamp(sf_date / 1000, time_zone)
return converted_data.strftime("%Y-%m-%d %H:%M:%S")
def object_table(obj):
return obj if not data_objects[obj] else data_objects[obj]
data_objects = {
'lead': None,
'closer': 'closer__c',
'users': 'user',
'conversation': 'ringdna__Conversation__c',
'opportunity': None,
'deal': 'deal__c',
'subscription': 'subscription__c',
'account': None,
'contact': None,
'task': None
}
#Cache
import csv
import os
from contextlib import closing
from multiprocessing import Pool
from Timing import get_time
from Util import get_mysql, fix_date, object_table, get_sf
cache_table = 'sf_SfPySnc'
data_cache_file_name = 'cache_data.csv'
cache_directory = 'cache_data'
def delete_mysql_cache_records() -> None:
print(f'delete cache data from mysql cache table {cache_table}')
with closing(get_mysql()) as mydb:
mydb.cursor().execute(f'DELETE FROM {cache_table}')
mydb.commit()
def write_to_file(csv_read_file):
with open(f'{cache_directory}/{csv_read_file}', "r") as reader:
csv_reader = csv.reader(reader, delimiter=',', quoting=csv.QUOTE_ALL, quotechar='"')
with open(data_cache_file_name, 'w+', newline='\n') as csv_write_file:
writer = csv.writer(csv_write_file, delimiter=',', quoting=csv.QUOTE_ALL, quotechar='"', escapechar='\\', )
next(csv_reader)
for row in csv_reader:
writer.writerow([row[0], fix_date(row[1])])
@get_time
def query_cache_dates(object_name: str) -> None:
delete_import_cache_data()
delete_import_cache_file()
print(f'query cache data from salesforce {object_table(object_name)} records')
query = f'SELECT Id, LastModifiedDate FROM {object_table(object_name)}'
getattr(get_sf().bulk2, object_table(object_name)).download(query, path=cache_directory)
print(f'writing cache data to combined data file {data_cache_file_name}')
with Pool() as pool:
pool.map(write_to_file, os.listdir(cache_directory))
def delete_import_cache_data():
files = os.listdir(cache_directory)
for file in files:
os.remove(os.path.join(cache_directory, file))
print("cache data files deleted")
def delete_import_cache_file():
if os.path.exists(data_cache_file_name) and os.path.isfile(data_cache_file_name):
os.remove(data_cache_file_name)
print("cache combined data file deleted")
@get_time
def insert_cache() -> None:
print(f'insert cache data file into {cache_table}')
sql = f"""
LOAD DATA LOCAL INFILE '{data_cache_file_name}' INTO TABLE {cache_table}
FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n';
"""
with closing(get_mysql()) as mydb:
mycursor = mydb.cursor()
mycursor.execute('SET GLOBAL local_infile=1')
mycursor.execute(sql)
print('*** commit cache')
mydb.commit()
def create_cache_dir() -> None:
if not os.path.exists(cache_directory):
os.makedirs(cache_directory)
def caching_main(object_name: str):
create_cache_dir()
delete_mysql_cache_records()
query_cache_dates(object_name)
insert_cache()
tulip lindenBOT
@onyx lava
Python help channel closed