To help users with our API, we have put together the following Python examples.


see also: API Programming Guide


Example 1: Add_Nodes.py

import requests
import time

url = 'https://<...>'
auth = ('<user@domain>', '<password>')

newNodes = [
	{'address': '192.168.1.1', 'label': 'Node 1'},
	{'address': '192.168.1.2', 'label': 'Node 2'},
	{'address': '192.168.1.3', 'label': 'Node 3'}
]

def run_task(type, data):
	task = {
		'type': type
	}
	if data:
		task['data'] = data

	r = requests.post(url + '/api/tasks/', auth=auth, json=task)
	if (r.status_code != 201) :
		print('Could not start task: ' + r.text)
		exit(1)

	taskLocation = r.headers['Location']
	# TODO: Implement timeout.
	while True :
		time.sleep(1)
		r = requests.get(url + taskLocation, auth=auth)
		taskOutput = r.json()
		if (taskOutput['finished']) :
			return taskOutput


data = {
	'discoveryType': 'fullSNMP',
	'protoNodes': newNodes
}

print('Starting discovery for ' + str(len(newNodes)) + ' addresses.')
taskOutput = run_task('discovery', data)
print('Discovery completed: ' + taskOutput['stateLabel'])
if 'addedNodes' in taskOutput:
	print('New Node IDs: ' + str(taskOutput['addedNodes']))

### Alternative version that adds addresses sequentially ###
# for newNode in newNodes:
# 	data = {
# 		'discoveryType': 'fullSNMP',
# 		'addresses': [newNode['address']],
# 		'label': newNode['label']		
# 	}
#
# 	print('\nStarting discovery for ' + newNode['address'] + ' / ' + newNode['label'])
# 	taskOutput = run_task('discovery', data)
# 	print('Discovery completed: ' + taskOutput['stateLabel'])
# 	if 'addedNodes' in taskOutput:
# 		print('New Node ID: ' + taskOutput['addedNodes'][0])


Example 2: Get_node_status.py

import requests

url = 'https://<...>'
auth = ('<user@domain>', '<password>')

r = requests.get(url + '/api/nodes/', auth=auth,
	params={'outputFormat': 'extended', 'sort': 'status', 'sortDir': 'desc'}
)

if (r.status_code == 200):
	nodes = r.json()
	for node in nodes:
		print(node['label'] + '\t\t' + node['status']['status'], end='')
		if 'maintenance' in node['status']:
			print(' - In Maintenance', end='')
		if 'acknowledged' in node['status']:
			print(' - Acknowledged', end='')
		print()

else:
	print('Failed: ' + str(r.status_code))


Example 3: reparent_children.py

This call will cause the children of a node to be moved to a new parent by using a traceroute.

import requests
import time
import argparse

url = 'https://<...>'
auth = ('<user@domain>', '<password>')

def run_task(type, data):
	task = {
		'type': type
	}
	if data:
		task['data'] = data

	r = requests.post(url + '/api/tasks/', auth=auth, json=task)
	if (r.status_code != 201) :
		print('Could not start task: ' + r.text)
		exit(1)

	taskLocation = r.headers['Location']
	# TODO: Implement timeout.
	while True :
		time.sleep(1)
		r = requests.get(url + taskLocation, auth=auth)
		taskOutput = r.json()
		if (taskOutput['finished']) :
			return taskOutput


parser = argparse.ArgumentParser(description='', formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('parent_id', help='ID of parent Node')
args = parser.parse_args()
config = vars(args)
parentID = config['parent_id']

r = requests.get(url + '/api/nodes/' + parentID + '/children/', auth=auth)

if (r.status_code != 200) :
	print('Could not read children: ' + r.text)
	exit(1)

children = r.json()
if (len(children) == 0) :
	print('Node has no children')
	exit(2)

print('Starting TraceAndMoveTask for ' + str(len(children)) + ' children.')

data = {
	'nodes': children
}

taskOutput = run_task('traceAndMove', data)

print('TraceAndMoveTask completed: ' + taskOutput['stateLabel'])


Example 4: Get Events

import requests
import datetime

#####################################################################

# 
# IMPORTANT: Please edit the following variables to match your environment.
#
url = 'https://<...>'
auth = ('<user@domain>', '<password>')
verifyHTTPS = False

#####################################################################

# Calculate the previous midnight and midnight one week ago.
now = datetime.datetime.now()
midnight = datetime.datetime.combine(now, datetime.time.min, now.astimezone().tzinfo)
weekAgo = midnight - datetime.timedelta(days=7)

# Make a request to the API to get the events from the past week.
r = requests.get(url + '/api/events/',
	auth=auth,
	verify=verifyHTTPS,
	params={
		'startTime': weekAgo.isoformat(),
		'endTime': midnight.isoformat(),
		'outputFormat': 'csv',
		'limit': 500,
		'sort': 'OPEN_TIME_DESC'
	}
)

if (r.status_code == 401 or r.status_code == 403):
	print('Failed Login: Could not authenticate')
elif (r.status_code == 200):
	print(r.text)
else:
	print('Failed: ' + str(r.status_code))


Example 5: Get Open Events

import requests

#####################################################################

# 
# IMPORTANT: Please edit the following variables to match your environment.
#
url = 'https://<...>'
auth = ('<user@domain>', '<password>')
verifyHTTPS = False

#####################################################################

# Make a request to the API to get the events from the past week.
r = requests.get(url + '/api/events/',
  auth=auth,
  verify=verifyHTTPS,
  params={
    'state': 'OPEN',
    'outputFormat': 'csv',
    'limit': 500,
    'sort': 'OPEN_TIME_DESC'
  }
)

if (r.status_code == 401 or r.status_code == 403):
  print('Failed Login: Could not authenticate')
elif (r.status_code == 200):
  print(r.text)
else:
  print('Failed: ' + str(r.status_code))


Example 6: Get Events Test (this fetches the events from demo.mutiny.com)

import requests
import datetime

#####################################################################

# 
# IMPORTANT: Please edit the following variables to match your environment.
#
url = 'https://demo.mutiny.com'
auth = ('user@mutiny.com', 'password')
verifyHTTPS = False

#####################################################################

print('-- Starting...')

session = requests.Session()

# Make a request to the public API to setup the session.
r = session.get(url + '/api/pollingInfo/statusPoller',
  auth=auth,
  verify=verifyHTTPS
)

if (r.status_code == 401 or r.status_code == 403):
  print('Failed Login: Could not authenticate')
  exit()

if (r.status_code != 200):
  print('Failed Login: ' + str(r.status_code))
  exit()

print('-- Initial login succeeded...')
print('-- Cookies: ' + str(session.cookies))

# Calculate the previous midnight and midnight one week ago.
now = datetime.datetime.now()
midnight = datetime.datetime.combine(now, datetime.time.min, now.astimezone().tzinfo)
weekAgo = midnight - datetime.timedelta(days=7)

print('-- Requesting events...')

# Make a request to the internal API to get the events from the past week.
r = session.get(url + '/interface/GetGridCSV',
  verify=verifyHTTPS,
  params={
    'gObject': 'events',
    'opentimestampStart': weekAgo.isoformat(),
    'opentimestampEnd': midnight.isoformat(),
    'gFilters': 'true',
    'gFiltersIncludeMissing': 'true',
    'state[]': 'special:all',
    'gPageSize': 'all'
  }
)

if (r.status_code == 200):
  print(r.text)
else:
  print('Failed: ' + str(r.status_code))