Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def main():
log = path.join(sys.path[0], 'email_log.log')
with open(log, 'a') as log:
try:
with EmailServer(smtp_server, smtp_username, smtp_password, use_tls) as email_server:
target = gis.GIS(orgURL, username, password)
for email_service in email_services:
try:
if email_service['service url'] == '':
continue
for message in email_service['messages']:
# build sql
sql = "{} != '{}'".format(message['status field'], message['completed value'])
sql += " AND {}".format(message['query'])
feature_layer = FeatureLayer(email_service['service url'], target)
features = _get_features(feature_layer, sql)
for feature in features:
if new_item is not None:
form.update_form(target, new_item, item_mapping)
_add_message("Updated Form {0}".format(new_item['title']))
existing_item = _search_org_for_existing_item(target, item)
_add_message('Successfully added {0}'.format(item['title']))
if existing_item:
_add_message("New item id: {0}".format(existing_item['id']))
_add_message('------------------------')
return created_items
except Exception as ex:
if isinstance(ex, _ItemCreateException):
_add_message(ex.args[0], 'Error')
if isinstance(ex.args[1], (gis.Item, gis.Group)):
created_items.append(ex.args[1])
elif isinstance(ex, _CustomCancelException):
if len(ex.args) > 0 and isinstance(ex.args[0], (gis.Item, gis.Group)):
created_items.append(ex.args[0])
_add_message("{0} canceled".format(item['title']), 'Error')
else:
_add_message(str(ex), 'Error')
for created_item in reversed(created_items):
try:
if created_item is not None:
if created_item.delete():
if isinstance(created_item, gis.Group):
created_item['type'] = 'Group'
_add_message("Deleted {0} {1}".format(created_item['type'], created_item['title']))
except Exception:
def _get_existing_item(existing_items, item, item_type="Item"):
"""Find the cloned item from a list of items.
Keyword arguments:
existing_items - The list of items to search through to find if the item has already been cloned.
item - The original item to be cloned.
type - The type of item being search, options are 'Item' or 'Group'"""
if item_type == "Item":
for existing_item in [i for i in existing_items if isinstance(i, gis.Item)]:
if existing_item['type'] == item['type']:
for keyword in existing_item['typeKeywords']:
if keyword == "source-{0}".format(item['id']):
return existing_item
elif item_type == "Group":
for existing_group in [i for i in existing_items if isinstance(i, gis.Group)]:
for tag in existing_group['tags']:
if tag == "source-{0}".format(item['id']):
return existing_group
return None
def _get_existing_item(existing_items, item, item_type="Item"):
"""Find the cloned item from a list of items.
Keyword arguments:
existing_items - The list of items to search through to find if the item has already been cloned.
item - The original item to be cloned.
type - The type of item being search, options are 'Item' or 'Group'"""
if item_type == "Item":
for existing_item in [i for i in existing_items if isinstance(i, gis.Item)]:
if existing_item['type'] == item['type']:
for keyword in existing_item['typeKeywords']:
if keyword == "source-{0}".format(item['id']):
return existing_item
elif item_type == "Group":
for existing_group in [i for i in existing_items if isinstance(i, gis.Group)]:
for tag in existing_group['tags']:
if tag == "source-{0}".format(item['id']):
return existing_group
return None