This is an example of some processes that I have set to work in multiprocessing. The script checks to see if the datasets needs to be exported (source date > destination date) that returns a dictionary of basic messages. In a separate script is the worker:
# IMPORTS
from arcpy import CopyFeatures_management
from arcpy import env
def export_to_shp(fcname, fc, shp):
"""
Function to copy features to shapefile
"""
try:
env.overwriteOutput = True
CopyFeatures_management(fc, shp)
return {'fcname': f'{fcname}', 'status': True, 'exmessage': 'was exported!'}
except Exception as ex:
return {'fcname': f'{fcname}', 'status': False, 'exmessage': f'{ex}'}
The main script looks like this. I removed a lot of the variables that are in house to us, so this wont be a copy/run but just to get an idea of structure and how you can return useful information from the multiprocessing.
# IMPORTS
import datetime as dt
import os
import sys
import time as t
from multiprocessing import Pool
import arcpy
from multiprocesses import ExportToShp
# ---------------------------------------------------------------------------
def weekly_export(sde_view):
"""
Function to export featureclasses to shapefiles if they were changed in the past 7 days
:param sde_view: connection class to connect to sde and output directories
"""
arcpy.env.overwriteOutput = True
startTime = t.time()
try:
# Featureclasses variables:
themesDir = r'path to output folder'
fcDict = {
'zoning': [sde_view.fc_connection("admin_CLC", "zoning"),
os.path.join(themesDir, 'zoning.shp')],
'row': [sde_view.fc_connection("property", "row"),
os.path.join(themesDir, 'row.shp')],
'physftr': [sde_view.fc_connection("property", "physftr"),
os.path.join(themesDir, 'physftr.shp')],
'streets': [sde_view.fc_connection("transportation", "streets"),
os.path.join(themesDir, 'streets.shp')]
}
fcList = []
for fc, fcPaths in fcDict.items():
# mt is an in house maintenance package, its just checking dates and feature counts here.
if any([mt.last_edited_date_check(fcPaths[0], "last_edited_date", 7),
mt.get_number_of_features(fcPaths[1]) > mt.get_number_of_features(fcPaths[0])]):
fcList.append((fc, fcPaths[0], fcPaths[1]))
# If more than one dataset needs to be exported, do it in parallel
if len(fcList) > 1:
with Pool(processes=4) as pool:
result = pool.starmap(ExportToShp.export_to_shp, fcList)
for res in result:
print(f'{res["fcname"]} {res["exmessage"]}')
if not res['status']:
print(f'{res["fcname"]} {res["exmessage"]}')
elif len(fcList) == 1:
result = ExportToShp.export_to_shp(fcList[0][0], fcList[0][1], fcList[0][2])
print(f'{result["fcname"]} {result["exmessage"]}')
if not result['status']:
print(f'{result["fcname"]} {result["exmessage"]}')
else:
print('No updates were made and nothing was exported')
except Exception as err:
print(
f'{sys._getframe().f_code.co_name} - FAILED in {dt.timedelta(seconds=t.time() - startTime)}! See error log...')
print(f'{sys._getframe().f_code.co_name} FAIL! - {err}')
if __name__ == '__main__':
mTime = t.time()
try:
weekly_export(con)
print(f'Export Weekly - COMPLETED in {dt.timedelta(seconds=t.time() - mTime)} ...')
except Exception as err:
print(
f'{sys._getframe().f_code.co_name} - FAILED in {dt.timedelta(seconds=t.time() - mTime)}! See error log...')
print(f'{sys._getframe().f_code.co_name} FAIL! - {err}')
