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