REST API and the Swimlane Python Driver Library

The Swimlane platform provides a REST API with numerous endpoints that allow developers to harness and extend the platform's automation and orchestration power. As a developer, you can use any programming tool with an HTTP library or capability. You can also use the Swimlane Python Driver documentation, which is a client library that can streamline your efforts to implement solutions.

In general, when authoring Python scripts for Swimlane, 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.
Tip: See Custom Direct Requests from the Swimlane Python Driver documentation to review the preferred method for blending these two approaches.
When authoring scripts that will run within Swimlane, use the Task API first, and then fall back on the driver and/or REST API when necessary.

Swimlane Python Driver

This is an example of a Swimlane Python Driver script for connecting to Swimlane and creating a new record in a specified application:

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' })

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()

REST API

The scripts above highlight the efficiency of scripting with Swimlane Python Driver. The examples that follow are example scripts in REST API.

Here's a sample REST API script for connecting to Swimlane and creating a new record in a specified 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:

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 ID msg_field_id = None for field in app_obj['fields']: if field['name'] == 'Event Message': msg_field_id = field['id'] break # 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!!!' # Persist the alteration to the record result = s.put(urljoin(base_url, 'api/app/' + app_obj['id'] + '/record/' + rec['id']), json=rec) print result

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) # PATCH fails if Tracking Id present 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.