Introduction to the Swimlane REST API and Python Driver

Background

The Swimlane platform provides a REST API with numerous endpoints to allow solution developers to harness and extend the platform's automation and orchestration power. Developers can use any programming tool with an HTTP library/capability. Or, they can use the Python library known as the Swimlane Driver, a client library that can reduce the effort required to implement custom solutions.

REST API

Here's a sample script for connecting to Swimlane and creating a new record in a chosen application:

import requests from urlparse import urljoin s = requests.Session() # Authenticate host = 'HOST' base_url = 'https://{0}'.format(host) creds = {'userName': 'USERNAME', 'password': 'PASSWORD'} resp = s.post(urljoin(base_url, 'api/user/login'), json=creds) json_content = resp.json() token = json_content.pop('token', None) headers = { 'Authorization': 'Bearer {}'.format(token) } s.headers.update(headers) # Retrieve the chosen application resp = s.get(urljoin(base_url, 'api/app')) apps = resp.json() app_obj = None for app in apps: if app['name'] == 'Event': app_obj = app break # Create a new record in the chosen application. First find pertinent field ID values in order to make correct field value assignments. name_field_id = None msg_field_id = None for field in app_obj['fields']: if field['name'] == 'Event Name': name_field_id = field['id'] if field['name'] == 'Event Message': msg_field_id = field['id'] record = { 'applicationId': app_obj['id'], 'values': { '$type': 'System.Collections.Generic.Dictionary`2[[System.String, mscorlib],[System.Object, mscorlib]], mscorlib', name_field_id: 'demo security event 1', msg_field_id: 'this event is for demonstration purposes' } } result = s.post(urljoin(base_url, 'api/app/' + app_obj['id'] + '/record'), json=record) print result

Here is a sample script for fetching an existing record, modifying it, and persisting the changes:

Here is the same script made more optimal by using HTTP PATCH rather than HTTP PUT to avoid REST API-centric race conditions that can arise when two scripts are attempting to operate on the same record within the same span of time. The use of PATCH reduces the chances of harmful, colliding alterations to the record.

import requests from urlparse import urljoin import json s = requests.Session() # Authenticate host = 'HOST' base_url = 'https://{0}'.format(host) creds = {'userName': 'USER', 'password': 'PASSWORD'} resp = s.post(urljoin(base_url, 'api/user/login'), json=creds) json_content = resp.json() token = json_content.pop('token', None) headers = { 'Authorization': 'Bearer {}'.format(token) } s.headers.update(headers) # Retrieve the chosen application resp = s.get(urljoin(base_url, 'api/app')) apps = resp.json() app_obj = None for app in apps: if app['name'] == 'Event': app_obj = app break # Find chosen field IDs name_field_id = None msg_field_id = None tracking_field_id = None for field in app_obj['fields']: if field['name'] == 'Event Name': name_field_id = field['id'] if field['name'] == 'Event Message': msg_field_id = field['id'] if field['name'] == 'Tracking Id': tracking_field_id = field['id'] # Retrieve an existing record result = s.get(urljoin(base_url, 'api/app/' + app_obj['id'] + '/record/tracking/EV1-1')) print result rec = result.json() #print json.dumps(rec, sort_keys=True, indent=4, separators=(",", ": ")) # Update the record rec['values'][msg_field_id] = 'I\'ve been updated!!!' rec['values'].pop(tracking_field_id, None) # The PATCH endpoint will throw error with a 400 response if the Tracking Id is present as a field value. rec['values'].pop(name_field_id, None) #print json.dumps(rec, sort_keys=True, indent=4, separators=(",", ": ")) # Persist the alteration to the record result = s.patch(urljoin(base_url, 'api/app/' + app_obj['id'] + '/record/' + rec['id']), json=rec) print result # Print result.text if you want to see the error message that accompanies an HTTP 400 response.

Swimlane Driver

The above examples help to convey the power and appeal of the Swimlane Driver. Compare the above examples with their Driver equivalents below:

from swimlane import Swimlane # Connect Swimlane client host = 'HOST' base_url = 'https://{0}'.format(host) swimlane = Swimlane(base_url, 'USER', 'PASSWORD', verify_ssl=False) # Retrieve App by name app = swimlane.apps.get(name='Event') # Create new Record new_record = app.records.create(**{ 'Event Name': 'demo security event 1', 'Event Message': 'this event is for demonstration purposes' })

and

from swimlane import Swimlane # Connect Swimlane client host = 'HOST' base_url = 'https://{0}'.format(host) swimlane = Swimlane(base_url, 'USERNAME', 'PASSWORD', verify_ssl=False) # Retrieve App by name app = swimlane.apps.get(name='Event') # Fetch existing record record = app.records.get(tracking_id='EV1-4') # Modify record record['Event Message'] = 'I\'ve been updated!!!' # Persist the modification record.patch()

Conclusion

When authoring Python scripts for the Swimlane Platform follow these guidelines:

When authoring scripts that will run outside of the Platform, use the Swimlane Driver whenever possible and fall back on the unwrapped REST API when necessary.

When authoring scripts that will run inside of the Platform, use the Task API first, and then fall back on the Driver and/or REST API when necessary.