11-05-2021 10:47 AM
Solved! Go to Solution.
11-08-2021 01:57 AM
11-07-2021 08:24 AM
import os
import requests
import json
import pprint
import pandas as pd
import time
import getpass
# Page size for each API request
limit = 1000
# Sleep time in seconds between each API call when mutiple API calls are needed for paginaation
sleep_time_seconds = 1
# Do not change any variables below
# URL of the API gateway
base_url = "https://api.extremecloudiq.com"
# Initialization of global variables
data = []
data_connected_only = []
device_without_locations = []
headers = {}
payload = {}
access_token = ''
def print_error_and_exit(desc, error):
print("*** Error description:", desc)
print("*** Error content:", error)
print("*** Exiting ...")
exit()
def display_usage():
print('''
This script will list all devices that are connected and without location assigned.
When the script runs, it will ask for the login credential of the the VIQ default admin.
The result will be exported to a csv file named "result.csv" placed under the same directory where this script runs.
The script uses the New API offered by XIQ.
See linked pages below for more details about the New API usage:
https://api.extremecloudiq.com/swagger-ui/index.html?configUrl=/openapi/swagger-config#/
https://extremecloudiq.com/api-docs/api-docs.html
''')
def get_access_token():
username = input("Login email address of the default admin of the VIQ: ").strip()
password = getpass.getpass()
url = base_url + "/login"
payload = json.dumps({
'username': username,
'password': password
})
headers['Content-Type'] = 'application/json'
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code != 200:
print_error_and_exit("Status code is not 200", response.text)
global access_token
access_token = json.loads(response.text)["access_token"]
print(access_token)
def list_devices(connected_only=None):
# Page ID of the first API reqeust. Do not change this.
page = 1
# API request to get all devices
url = base_url + "/devices?page=" + str(page) + "&limit=" + str(limit)
payload = {}
headers['Authorization'] = 'Bearer ' + access_token
response = requests.request("GET", url, headers=headers, data=payload)
if response.status_code != 200:
print_error_and_exit("Status code is not 200", response.text)
# Parse the API resposne as the json data
response_json = json.loads(response.text)
curent_page = response_json["page"]
print("Current page:", curent_page)
total_pages = response_json["total_pages"]
print("Total number of pages:", total_pages)
# Save the devices data to the global variable "data"
global data
data = (response_json["data"])
# Make API calls pages other the the first page and aggregte the results
print("Looping trhough to get the data")
while curent_page < total_pages:
time.sleep(sleep_time_seconds)
curent_page += 1
print("currrent_page:", curent_page)
url = base_url + "/devices?page=" + str(curent_page) + "&limit=" + str(limit)
response = requests.request("GET", url, headers=headers, data=payload)
if response.status_code != 200:
print_error_and_exit("Status code is not 200", response.text)
response_json = json.loads(response.text)
data.extend(response_json["data"])
print("Data of all devices:")
pprint.pprint(data)
print("Number of items in data before filtered:", len(data))
if connected_only:
print("*** Find the devcies in connected status only")
global data_connected_only
for i in data:
if i["connected"] == True:
data_connected_only.append(i)
print("Devices with connected status:")
pprint.pprint(data_connected_only)
print("Number of devices with connected status:", len(data_connected_only))
def get_devices_without_locations(connected_only=None):
if connected_only == True:
print("*** Get devices wth connected status only. ")
data_devices = data_connected_only
msg_connection_status = '(connected)'
else:
print("*** Get devices regardless of the connection status")
data_devices = data
msg_connection_status = '(regardless of connection status)'
print("*** Get the device IDs list")
# List of the IDs of all devices
device_ids = []
for i in data_devices:
device_ids.append(str(i["id"]))
print("*** IDs of all devices:", device_ids, msg_connection_status)
print("*** Number of IDs of all devices:", len(device_ids), msg_connection_status)
url = base_url + "/devices/location/:query"
payload = json.dumps({
"ids": device_ids
})
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code != 200:
print_error_and_exit("Status code is not 200", response.text)
# Parse the API resposne as the json data
response_json = json.loads(response.text)
print("Devices with locations:")
pprint.pprint(response_json)
print("*** Numbe of devices with locations:", len(response_json), msg_connection_status)
device_ids_with_locations = list(response_json.keys())
print(device_ids_with_locations)
device_ids_without_locations = list(set(device_ids) - set(device_ids_with_locations))
print("*** IDs of devices without locations:", device_ids_without_locations, msg_connection_status)
print("*** Number of devices without locations:", len(device_ids_without_locations), msg_connection_status)
global device_without_locations
for i in data_devices:
if str(i["id"]) in device_ids_without_locations:
device_without_locations.append(i)
print("*** Devices without locations:", msg_connection_status)
pprint.pprint(device_without_locations)
print("*** Number of devices without locations:", len(device_without_locations), msg_connection_status)
def export_csv():
df = pd.json_normalize(device_without_locations)
df.to_csv("result.csv", index=False)
if os.name == "nt":
os.system("type result.csv")
else:
print("Print the firt 3 lines of result.csv")
os.system("cat result.csv")
print("*** Location of the exported file: " + os.getcwd() + "/result.csv")
# main
display_usage()
get_access_token()
list_devices(connected_only=True)
get_devices_without_locations(connected_only=True)
export_csv()
11-08-2021 01:57 AM
11-08-2021 06:10 AM