diff --git a/ServiceBased/Jupyter Notebook Examples/Workflow Manager - Automated Creations.html b/ServiceBased/Jupyter Notebook Examples/Workflow Manager - Automated Creations.html new file mode 100644 index 0000000..7c45344 --- /dev/null +++ b/ServiceBased/Jupyter Notebook Examples/Workflow Manager - Automated Creations.html @@ -0,0 +1,14802 @@ + + +
+ + +import arcgis
+import re
+import datetime
+from arcgis.gis.workflowmanager import WorkflowManager
+
+gis = arcgis.gis.GIS(profile="ps0008227", verify_cert=False)
+item = gis.content.search('title:"Python Sample"')[0]
+workflowManager = WorkflowManager(item)
+print("Created connection to workflow manager item")
+Created connection to workflow manager item ++
# Get Default diagram
+diagrams = workflowManager.diagrams
+diagram = {}
+for gram in diagrams:
+ if gram.diagram_name == 'Introduction to Workflow Manager':
+ diagram = gram
+ break
+
+# Create the job template
+uniqueness = re.sub("[^0-9a-z]+", "_", str(datetime.datetime.now()))
+
+#unique names so there is no duplicate naming errors
+template_name = 'Example Template ' + uniqueness
+table_name = 'example_table_' + uniqueness
+
+template_id = workflowManager.create_job_template(name=template_name, diagram_id=diagram.diagram_id, diagram_name=diagram.diagram_name,
+ priority='High', category='Functional Tests', job_duration=5,assigned_to="admin",
+ default_due_date='2020-04-02T13:25:50Z', default_start_date='2020-04-02T13:25:50Z',
+ start_date_type='CreationDate',assigned_type='Unassigned', description='Test Test test',
+ default_description='Test Test123', state='Active', last_updated_by='Abbie Admin',
+ last_updated_date='2020-04-02T13:25:50Z'
+ )
+if template_id:
+ print('Successfully created template')
+
+template = workflowManager.job_template(template_id)
+
+# Create Automated Creation for job template
+adds = [{
+ "automationName": "auto_mation",
+ "automationType": "Scheduled",
+ "enabled": True,
+ "details": "{\"timeType\":\"NumberOfDays\",\"dayOfMonth\":1,\"hour\":8,\"minutes\":0}"},
+ {
+ "automationName": "auto_mation",
+ "automationType": "Scheduled",
+ "enabled": True,
+ "details": "{\"timeType\":\"NumberOfDays\",\"dayOfMonth\":1,\"hour\":8,\"minutes\":0}"}
+ ]
+
+success = template.update_automated_creations(adds)
+
+if success:
+ print('Successfully created automation')
+Successfully created template +Successfully created automation ++
# Update and Delete
+creations = workflowManager.job_template_automated_creations(template_id)
+
+
+id_one = creations[0]["automationId"]
+id_two = creations[1]["automationId"]
+
+adds = [{
+ "automationName": "test_three",
+ "automationType": "Scheduled",
+ "enabled": True,
+ "details": "{\"timeType\":\"NumberOfDays\",\"dayOfMonth\":1,\"hour\":8,\"minutes\":0}"
+ }]
+updates = [{
+ "automationId": id_two,
+ "automationName": "test_two_updated"
+ }]
+deletes = [ id_one ]
+
+template.update_automated_creations(adds, updates, deletes)
+creations_two = workflowManager.job_template_automated_creations(template_id)
+
+print(creations_two)
+[{'automationId': 'wb5zpNjJS-iOX6baOq5ueQ', 'automationName': 'test_two_updated', 'automationType': 'Scheduled', 'createdBy': 'admin'}, {'automationId': 'B-klmp70S7a7pZtGvSvIhA', 'automationName': 'test_three', 'automationType': 'Scheduled', 'createdBy': 'admin'}]
+
+
+This sample will create a new Workflow item and populate it with configuration data from an existing Workflow item. This data includes the system settings, diagram, job templates and roles.
+ +import arcgis
+from arcgis.gis.workflowmanager import WorkflowManager, WorkflowManagerAdmin
+gis = arcgis.gis.GIS(profile="ps0008227", verify_cert=False)
+dest_gis = arcgis.gis.GIS(profile="ps0010644", verify_cert=False)
+source_item = gis.content.search('title:"Python Sample"')[0]
+new_item_id = WorkflowManagerAdmin(dest_gis).create_item('Duplicated Workflow Sample')
+print(f'Created item {new_item_id}')
+dest_item = dest_gis.content.get(new_item_id)
+source = WorkflowManager(source_item)
+dest = WorkflowManager(dest_item)
+
+# Could also update sharing properties here
+Created item 94ba49aa317449afa6e167d57057a6ef ++
# Copy Settings
+# Note that the password won't be copied
+dest.update_settings(source.settings)
+True+
# Copy Diagrams
+existing_diagrams = [x.diagram_id for x in dest.diagrams]
+diagram_ids = [x.diagram_id for x in source.diagrams if not(x.diagram_id in existing_diagrams)]
+diagram_id_map = {}
+for id in diagram_ids:
+ d = source.diagram(id)
+ new_id = dest.create_diagram(d.diagram_name, d.steps, d.display_grid, d.description, d.diagram_version >= 0, d.annotations,
+ d.data_sources) # Remove datasources here if you don't want to copy
+ diagram_id_map.update({id: new_id})
+ print(f'Created diagram {d.diagram_name}. ID = {new_id}')
+diagram_id_map
+Created diagram Fixing Infrastructure. ID = BuknnYDDSSefLQ0bkZkorw ++
{'9WDJNN1XQmy9RqZghxZfcA': 'BuknnYDDSSefLQ0bkZkorw'}
+# Copy Job Templates
+existing_job_templates = [x.job_template_id for x in dest.job_templates]
+job_template_ids = [x.job_template_id for x in source.job_templates if not(x.job_template_id in existing_job_templates)]
+for id in job_template_ids:
+ t = source.job_template(id)
+ new_diagram_id = diagram_id_map[t.diagram_id]
+ result = dest.create_job_template(t.job_template_name, t.default_priority_name, t.job_template_id, t.category,
+ t.default_job_duration, t.default_assigned_to, t.default_due_date,
+ t.default_start_date, t.job_start_date_type, new_diagram_id, t.diagram_name,
+ t.default_assigned_type, t.description, t.default_description,
+ t.state, t.last_updated_by, t.last_updated_date, t.extended_property_table_definitions)
+ print(f'Created job template {t.job_template_name}. ID = {result}')
+
+Created job template Fixing Infrastructure. ID = 0Y9yoV7MTaGGd8z3wHL6fQ ++
# Roles
+dest_roles = [x.role_name for x in dest.wm_roles]
+new_roles = [x.role_name for x in source.wm_roles if not(x.role_name in dest_roles)]
+print(f'Creating roles {new_roles}')
+for name in new_roles:
+ r = source.wm_role(name)
+ dest.create_wm_role(name, r.description, r.privileges)
+Creating roles ['Test Role'] ++
existing_searches = [x['searchId'] for x in dest.searches()]
+searches = [x for x in source.searches() if not(x['searchId'] in existing_searches)]
+for s in searches:
+ new_search = dest.create_saved_search(name=s['name'], definition=s['definition'], search_type=s['searchType'],
+ sort_index=s['sortIndex'], search_id=s['searchId'])
+ print('Created search '+ s['name'] + f'. ID = {new_search}')
+Created search Example Search. ID = kkeLH0f8TsSHLjgjeZGMvw ++
+This sample creates a job in ArcGIS Workflow Manager for each response to a Survey 123 survey. It will query the survey data, and use that information to create a new job with a priority based on data in the survey, copy any attachments from the survey to the job, and link the job to the survey via extended properties
+ +import arcgis
+import tempfile
+from arcgis.gis.workflowmanager import WorkflowManager
+online = arcgis.gis.GIS(profile='agol')
+survey = online.content.search('title:"Damaged Infrastructure" type:"Form"')[0]
+survey
+survey_data = survey.related_items('Survey2Service')[0]
+survey_layer = survey_data.layers[0]
+survey_features = survey_layer.query()
+survey_features.sdf
+| + | objectid | +globalid | +CreationDate | +Creator | +EditDate | +Editor | +type_of_item | +SHAPE | +
|---|---|---|---|---|---|---|---|---|
| 0 | +6 | +d0b9da18-c6eb-4e3a-8ac1-db39dfaaf127 | +2020-07-27 21:38:20.219000101 | +adelgadillo_solutions | +2020-07-27 21:38:20.219000101 | +adelgadillo_solutions | +sign | +{"x": -117.19751319999804, "y": 34.05609419999... | +
| 1 | +7 | +142e0ac6-d74b-4685-84c3-16afbb40c3f1 | +2020-07-27 21:40:06.358000040 | +adelgadillo_solutions | +2020-07-27 21:40:06.358000040 | +adelgadillo_solutions | +tree | +{"x": -117.20251749527404, "y": 34.05563510623... | +
| 2 | +8 | +284ce6a4-1e26-4628-8d41-5a766d570afc | +2020-07-27 21:40:30.857000113 | +adelgadillo_solutions | +2020-07-27 21:40:30.857000113 | +adelgadillo_solutions | +streetlight | +{"x": -117.2007358661602, "y": 34.055539896910... | +
gis = arcgis.gis.GIS(profile='ps0008227', verify_cert=False)
+item = gis.content.search('title:"Python Sample"')[0]
+workflowManager = WorkflowManager(item)
+job_template_ids = [x.job_template_id for x in workflowManager.job_templates if x.job_template_name == 'Fix Damaged infrastructure']
+job_template_id = job_template_ids[0]
+job_template = workflowManager.job_template(job_template_id)
+job_template.extended_property_table_definitions
+WrkdpVMVTdWNSeSgE2w1rw ++
[{'tableName': 'request_info',
+ 'tableAlias': 'Request_info',
+ 'tableOrder': 0,
+ 'relationshipType': 'OneToOne',
+ 'extendedPropertyDefinitions': [{'propertyName': 'survey_id',
+ 'propertyAlias': 'Survey ID',
+ 'propertyOrder': 0,
+ 'dataType': 'String',
+ 'defaultValue': '',
+ 'fieldLength': 50,
+ 'required': True,
+ 'editable': True,
+ 'visible': True}]}]
+def get_priority_from_type(type_of_item):
+ if (type_of_item == 'sign'):
+ return 'High'
+ elif (type_of_item == 'sidewalk'):
+ return 'Low'
+ else:
+ return 'Medium'
+
+
+def copy_attachments(oid, attachments, job_id):
+ with tempfile.TemporaryDirectory() as temp_dir:
+ for attachment in attachments:
+ files = survey_layer.attachments.download(oid, attachment['id'], save_path=temp_dir)
+ workflowManager.jobs.get(job_id).add_attachment(files[0])
+
+
+for f in survey_features.features:
+ priority = get_priority_from_type(f.get_value('type_of_item'))
+
+ # Check for existing jobs with the ext property field
+ survey_id = f.get_value('globalid')
+ num = workflowManager.jobs.search(query=f"request_info.survey_id = '{survey_id}'", fields=['job_id'], num=1)['num']
+ if num > 0:
+ continue
+
+ new_jobs = workflowManager.jobs.create(job_template_id, priority=priority)
+ if new_jobs != None and len(new_jobs) == 1:
+ job_id = new_jobs[0]
+ print(f'Created job {job_id}')
+ else:
+ print(f'Failed to create job: {new_jobs}')
+ continue
+
+ # Copy the attachments if there are any
+ survey_oid = f.get_value('objectid')
+ attachments = survey_layer.attachments.get_list(survey_oid)
+ if (attachments):
+ print('Survey has attachments, copying them to job')
+ copy_attachments(survey_oid, attachments, job_id)
+
+ # Set the extended property value
+ workflowManager.jobs.update(job_id, {
+ 'extendedProperties': [{'identifier': 'request_info.survey_id', 'value': survey_id}]
+ })
+
+Created job tNyssd9IQ7C9wA9FwlqWjg +Survey has attachments, copying them to job +Created job dQ6OaHvfScmk_91Q1R2tZw +Survey has attachments, copying them to job +Created job P9E-LlLHT2CQbwQKACxSGw +Survey has attachments, copying them to job ++
# Search for all the jobs related to surveys
+workflowManager.jobs.search(query=f"request_info.survey_id is not null and request_info.survey_id <> ''", fields=['job_id', 'request_info.survey_id'])
+{'q': "exmhczkswtzig_survey_id is not null and exmhczkswtzig_survey_id <> ''",
+ 'fields': [{'name': 'job_id', 'fieldType': 'String'},
+ {'name': 'request_info.survey_id', 'fieldType': 'String'}],
+ 'results': [['tNyssd9IQ7C9wA9FwlqWjg',
+ 'd0b9da18-c6eb-4e3a-8ac1-db39dfaaf127'],
+ ['dQ6OaHvfScmk_91Q1R2tZw', '142e0ac6-d74b-4685-84c3-16afbb40c3f1'],
+ ['P9E-LlLHT2CQbwQKACxSGw', '284ce6a4-1e26-4628-8d41-5a766d570afc']],
+ 'start': 0,
+ 'next_start': -1,
+ 'num': 3}
+
+
+This sample will create a job with job properties, extended properties. Then it will show examples of how to update each.
+ +import arcgis
+import re
+import datetime
+from arcgis.gis.workflowmanager import WorkflowManager
+
+gis = arcgis.gis.GIS(profile="ps0008227", verify_cert=False)
+item = gis.content.search('title:"Python Sample"')[0]
+workflowManager = WorkflowManager(item)
+# Get Default diagram
+diagrams = workflowManager.diagrams
+diagram = {}
+for gram in diagrams:
+ if gram.diagram_name == 'Introduction to Workflow Manager':
+ diagram = gram
+ break
+
+# Create the job template with extended properties
+uniqueness = re.sub("[^0-9a-z]+", "_", str(datetime.datetime.now()))
+
+#unique names so there is no duplicate naming errors
+template_name = 'Example Template ' + uniqueness
+table_name = 'example_table_' + uniqueness
+
+success = workflowManager.create_job_template(name=template_name, diagram_id=diagram.diagram_id, diagram_name=diagram.diagram_name,
+ priority='High', category='Functional Tests', job_duration=5,assigned_to="admin",
+ default_due_date='2020-04-02T13:25:50Z', default_start_date='2020-04-02T13:25:50Z',
+ start_date_type='CreationDate',assigned_type='Unassigned', description='Test Test test',
+ default_description='Test Test123', state='Active', last_updated_by='Abbie Admin',
+ last_updated_date='2020-04-02T13:25:50Z',
+ extended_property_table_definitions=[
+ {
+ "tableName": table_name,
+ "tableAlias": table_name,
+ "tableOrder": 0,
+ "relationshipType": "OneToOne",
+ "extendedPropertyDefinitions": [
+ # String Property
+ {"propertyOrder": 0,
+ "visible": True,
+ "propertyName": "prop1",
+ "editable": True,
+ "dataType": "String",
+ "propertyAlias": "prop1",
+ "required": True,
+ "fieldLength": 50
+ },
+ # String Property
+ {"propertyOrder": 1,
+ "visible": True,
+ "propertyName": "prop2",
+ "editable": True,
+ "dataType": "String",
+ "propertyAlias": "prop2",
+ "required": True,
+ "fieldLength": 50
+ },
+ # Domain Property
+ {"propertyOrder": 2,
+ "visible": True,
+ "propertyName": "prop4",
+ "editable": True,
+ "domain": {
+ "type": "codedValue",
+ "codedValues": [
+ {
+ "code": "1",
+ "name": "Uno"
+ },
+ {
+ "code": "2",
+ "name": "Dos"
+ }
+ ],
+ "range": [
+ "string"
+ ]
+ },
+ "dataType": "Integer",
+ "propertyAlias": "prop4",
+ "required": True,
+ "fieldLength": 50
+ }
+ ],
+ }
+ ]
+ )
+if success:
+ print('Successfully created template')
+Successfully created template ++
# Find newly create template
+job_templates = workflowManager.job_templates
+job_template = {}
+for x in job_templates:
+ if x.job_template_name == template_name:
+ job_template = x
+
+print('Template \"' + job_template.job_template_name + '\" Id: ' + job_template.job_template_id)
+
+# Create Job setting some of the jobs extended properties
+job_ids = workflowManager.jobs.create( template=job_template.job_template_id, count=1, name='Test New Job123',
+ start='2020-04-02T13:25:50Z', end='2020-04-02T13:25:50Z', priority='High',
+ description='Perfect description of work', owner="admin", assigned="admin",
+ complete=42, notes='testing notes', parent='',
+ extended_properties=[
+ # String Property
+ {
+ "identifier": table_name + ".prop1",
+ "value": "newly_created123"
+ },
+ # String Property
+ {
+ "identifier": table_name + ".prop2",
+ "value": "newly_created456"
+ },
+ # Domain Property
+ {
+ "identifier": table_name + ".prop4",
+ "value": "1"
+ }
+
+ ])
+print('Created job: ' + job_ids[0])
+Template "Example Template 2021_05_11_13_21_17_783922" Id: t98IsJlPRq6fqAEDuq3Hkg +Created: 5facuDmqRW-Tk3Cvcl05bQ ++
# Get job we previously created. Set True to return the extended properties with the get call.
+job = workflowManager.jobs.get(job_ids[0], True)
+
+#update job properties
+job.priority = 'Updated'
+job.job_name = 'updated Job 123'
+job.description = "Updated description"
+
+# update extended properties
+table_name = job.extended_properties[0]["tableName"]
+job.extended_properties = [
+ {
+ "identifier": table_name + ".prop1",
+ "value": "updated_123"
+ },
+ {
+ "identifier": table_name + ".prop2",
+ "value": "updated_456"
+ },
+ {
+ "identifier": table_name + ".prop4",
+ "value": "2"
+ },
+]
+
+# Actual update call
+success = workflowManager.jobs.update(job_ids[0], vars(job))
+
+if success:
+ print('Successfully updated template\n')
+
+updated_job = workflowManager.jobs.get(job_ids[0], True)
+
+print('Updated Job\'s Extended Properties')
+print(updated_job.extended_properties)
+Successfully updated template
+
+Updated Job's Extended Properties
+[{'tableName': 'example_table_2021_05_11_13_21_17_783922', 'properties': [{'propertyName': 'prop1', 'value': 'updated_123'}, {'propertyName': 'prop2', 'value': 'updated_456'}, {'propertyName': 'prop4', 'value': 2}]}]
+
+
+import arcgis
+import re
+import datetime
+from arcgis.gis.workflowmanager import WorkflowManager
+
+gis = arcgis.gis.GIS(url='https://ps0010886.esri.com/portal', username='admin', password='esri.agp', verify_cert=False)
+item = gis.content.search('title:"Python Sample"')[0]
+workflowManager = WorkflowManager(item)
+print("Created connection to workflow manager item")
+# Create Job
+job_templates = workflowManager.job_templates
+job_template = {}
+for x in job_templates:
+ if x.job_template_name == 'Introduction to Workflow Manager':
+ job_template = x
+
+jobs = workflowManager.jobs.create(template=job_template.job_template_id,
+ count=count,
+ name='Test New Job123',
+ start='2020-04-02T13:25:50Z',
+ end='2020-04-02T13:25:50Z',
+ priority='High',
+ description='hopefully this works...',
+ owner=self.connection.portal_username,
+ assigned=self.connection.portal_username,
+ complete=42,
+ notes='testing notes'
+ )
+job_id = jobs[0]
+
+attachments = [{'url': 'linked text', 'folder': 'General', 'alias': 'linkedText'},
+ {'url': 'https://www.esri.com', 'folder': 'General', 'alias': 'webpath'},
+ {'url': 'tests//integration//README.md', 'folder': 'General', 'alias': 'filepath'}
+ ]
+
+a_list = workflowManager.jobs.get(job_id).add_linked_attachment(attachments)
+print(a_list)
+group_name = 'Workflow Manager Admin Python Sample'
+for group in groups:
+ if group['title'] == default_group_name:
+ break
+
+updates = { "adds": {"roles": ["Manage Jobs - Advanced", "Workflow Designer"]},
+ "deletes": {"roles": ["Workflow Administrator"]}}
+
+workflowManager.update_group(group['id'], updates)
+
+# check for updates
+specific_group = self.connection.workflow_manager.group(group['id'])
+
+has_role = 'Workflow Designer' in specific_group.roles
+print(has_role)
+does_not_have_role = 'Workflow Designer' not in specific_group.roles
+print(has_role)
+