- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Python Get API randomly hits limit
Hi,
working on a script for Meraki to count all wireless clients in a organization and there capabilities. But the code run randomly in to the API limit which is 5 per second. As a test is added allot of sleep timer to slow down the code but even that doesn’t work and i cannot figure out why.
import requests
import time
import os.path
headers = {'X-Cisco-Meraki-API-Key': xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'}
se = requests.session()
orgid = {'Test': '111111'}
cnt_80211b = 0
cnt_80211n = 0
cnt_80211ac = 0
cnt_80211ax = 0
try:
timer = 0
client_info = []
for org in orgid:
networks = 'https://api.meraki.com/api/v0/organizations/{}/networks'.format(str(orgid[org]))
time.sleep(1)
networks_data = se.get(networks, headers=headers).json()
time.sleep(1)
response = requests.request("GET", networks, headers=headers)
time.sleep(1)
print ("networks_data")
if response.status_code == 200:
time.sleep(1)
if networks_data == 'errors':
time.sleep(5)
else:
for each in networks_data:
print(".....................................................")
print (each['name'])
print(".....................................................")
time.sleep(1)
clients = 'https://api.meraki.com/api/v0/networks/{}/clients'.format(str(each['id']))
clients_data = se.get(clients, headers=headers).json()
time.sleep(1)
response = requests.request("GET", clients, headers=headers)
time.sleep(1)
if response.status_code == 200:
time.sleep(1)
for each_cli in clients_data:
if clients_data == 'errors':
time.sleep(5)
else:
if each_cli['ssid'] is not None:
time.sleep(1)
clientsurl = 'https://api.meraki.com/api/v0/networks/{}/clients/{}'.format(str(each['id']), str(each_cli['id']))
clientsdata = se.get(clientsurl, headers=headers).json()
time.sleep(1)
response = requests.request("GET", clientsurl, headers=headers)
time.sleep(1)
if response.status_code == 200:
if clientsdata == 'error':
print ("Error")
print (clientsdata)
time.sleep(5)
else:
time.sleep(0.3)
if "802.11b" in clientsdata['wirelessCapabilities']:
print ("802.11b")
cnt_80211b = cnt_80211b + 1
elif "802.11n" in clientsdata['wirelessCapabilities']:
print ("802.11n")
cnt_80211n = cnt_80211n + 1
elif "802.11ac" in clientsdata['wirelessCapabilities']:
print ("802.11ac")
cnt_80211ac = cnt_80211ac + 1
elif "802.11ax" in clientsdata['wirelessCapabilities']:
print ("802.11ax")
cnt_80211ax = cnt_80211ax + 1
elif response.status_code == 429:
time.sleep((int(response.headers["Retry-After"]))+ 1)
elif response.status_code == 429:
time.sleep((int(response.headers["Retry-After"]))+ 1)
elif response.status_code == 429:
time.sleep((int(response.headers["Retry-After"]))+ 1)
print ("...............................")
print ("802.11b")
print (cnt_80211b)
print ("...............................")
print ("...............................")
print ("802.11n")
print (cnt_80211n)
print ("...............................")
print ("...............................")
print ("802.11ac")
print (cnt_80211ac)
print ("...............................")
print ("...............................")
print ("802.11ax")
print (cnt_80211ax)
print ("...............................")
except Exception as ex:
print ("ooopsss")
print(str(ex))
Solved! Go to solution.
- Labels:
-
Code Sample
-
Dashboard API
-
Scanning API
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Found the solution and it works like a charm.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Indeed thanks for the heads up!
Adjusted
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
If have this in my code.
elif r.status_code == 429:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
continue
So perhaps add the keyword continue.
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Thank you
Testing it now
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Also, a bit of advice, use environment variables for your API key, so you don't have it stored directly in your code. Adds some security, and flexibility, so that others can use the same script, without having to modify the source code directly.
LIke this.
import os
[...]
API_KEY = os.environ.get('MERAKI_API_KEY', None)
if API_KEY is None:
raise SystemExit('Please set environment variable MERAKI_API_KEY')
else:
API_KEY = os.environ['MERAKI_API_KEY']
Assumes you have created the env-var MERAKI_API_KEY on your system.
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Thank you!
nice feature to add!
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
tested and the code runs a bit longer but still after some time:
clients_data
{'errors': ['API rate limit exceeded for organization']}
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
I think you should clean up your code, for starters.
Instead of mixing requests() and sessions() use only one.
Use functions, and segment your code, for simplicity.
def GetOrgs(p_apiKey):
for _ in range(MAX_RETRIES):
try:
r = session.get(
baseURL+'/organizations',
headers = {
'Authorization': f'Bearer {p_apiKey}',
'Content-Type': 'application/json',
"Accept": "application/json",
}
)
if r.status_code == 200:
return r
elif r.status_code == 429:
print(f'Rate limited - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
continue
else:
raise SystemExit(f'Unexpected status code: {r.status_code} returned from server.')
except Exception as e:
pprint(e)
(Note: I'm using the v1 beta api..)
For every session.get(), check the status code.
If it's 429, put in a sleep.
It's not the script that's being rate limited, it's the request that's being throttled.
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
totally agree on cleaning up the code and using functions. Not that good at Python but learning it on the go 🙂
do you perhaps have a complete code that i could use and modify?
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Not for your specific case, unfortunately.
But perhaps you can modify this to siut your needs.
#!/usr/bin/env python3
from pprint import pprint
import os, time, sys
import argparse
import requests
from requests import Session
import json
MAX_RETRIES = 5
API_KEY = None
apiVersion = 'v1'
baseURL = 'https://api.meraki.com/api/'+apiVersion
class NoRebuildAuthSession(Session):
def rebuild_auth(self, prepared_request, response):
'''
No code here means requests will always preserve the Authorization header when redirected.
Be careful not to leak your credentials to untrusted hosts!
'''
session = NoRebuildAuthSession()
def GetOrgs(p_apiKey):
for _ in range(MAX_RETRIES):
try:
r = session.get(
baseURL+'/organizations',
headers = {
'Authorization': f'Bearer {p_apiKey}',
'Content-Type': 'application/json',
"Accept": "application/json",
}
)
if r.status_code == 200:
return r
elif r.status_code == 429:
print(f'Rate limited - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
continue
else:
raise SystemExit(f'Unexpected status code: {r.status_code} returned from server.')
except Exception as e:
pprint(e)
# Main routine function
def main(arg_org_id,arg_get_nw,arg_debug):
if arg_debug is not None:
print(locals())
print()
API_KEY = os.environ.get('MERAKI_API_KEY', None)
if API_KEY is None:
raise SystemExit('Please set environment variable MERAKI_API_KEY')
else:
API_KEY = os.environ['MERAKI_API_KEY']
if arg_org is None:
print('Getting list of organisations...')
r = GetOrgs(API_KEY)
rjson = r.json()
print('Success!')
pprint(rjson)
def run(args):
parser = argparse.ArgumentParser()
group = parser.add_mutually_exclusive_group()
parser.add_argument('-o', '--org-id', dest='arg_org_id', help='Set the Organisation ID')
group.add_argument('-n', '--get-networks', dest='arg_get_nw', help='Get Networks')
parser.add_argument('-d', '--debug', dest='arg_debug')
options = parser.parse_args(sys.argv[1:])
main(**vars(options))
if __name__ == "__main__":
run(sys.argv[1:])
Edit: Added sys package.
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Build a new script last night it works much better but still hits the API limit.
😅
# This is a Python 3 script to count the total unique client MAC addresses connected to MR access points for
# an organization during the last month.
#
# Usage:
# clientcount.py -k <api key> [-o <org name>]
#
# Parameters:
# -k <api key> : Mandatory. Your Meraki Dashboard API key
# -o <org name> : Optional. Name of the organization you want to process. Use keyword "/all" to explicitly
# specify all orgs. Default is "/all"
#
# Example:
# clientcount.py -k 1234 -o "Big Industries Inc"
#
# Notes:
# * In Windows, use double quotes ("") to enter command line parameters containing spaces.
# * This script was built for Python 3.7.1.
# * Depending on your operating system, the command to start python can be either "python" or "python3".
#
# Required Python modules:
# Requests : http://docs.python-requests.org
#
# After installing Python, you can install these additional modules using pip with the following commands:
# pip install requests
#
# Depending on your operating system, the command can be "pip3" instead of "pip".
#
#
import sys, getopt, requests, json, time, datetime, os, sqlite3, threading
from ratelimit import limits
#SECTION: GLOBAL VARIABLES: MODIFY TO CHANGE SCRIPT BEHAVIOUR
#SECTION: GLOBAL VARIABLES AND CLASSES: DO NOT MODIFY
ARG_APIKEY = '' #DO NOT STATICALLY SET YOUR API KEY HERE
ARG_ORGNAME = '' #DO NOT STATICALLY SET YOUR ORGANIZATION NAME HERE
ORG_LIST = None #list of organizations, networks and MRs the used API key has access to
MAX_CLIENT_TIMESPAN = 2592000 #maximum timespan GET clients Dashboard API call supports
class c_Net:
def __init__(self):
id = ''
name = ''
shard = 'api.meraki.com'
devices = []
class c_Organization:
def __init__(self):
id = ''
name = ''
shard = 'api.meraki.com'
nets = []
#SECTION: General use functions
#make it work nice across threads
def RateLimited(max_per_second):
'''
Decorator that make functions not be called faster than
'''
lock = threading.Lock()
minInterval = 1.0 / float(max_per_second)
def decorate(func):
lastTimeCalled = [0.0]
def rateLimitedFunction(*args):
lock.acquire()
elapsed = time.perf_counter() - lastTimeCalled[0]
leftToWait = minInterval - elapsed
if leftToWait>0:
time.sleep(leftToWait)
lock.release()
ret = func(*args)
lastTimeCalled[0] = time.perf_counter()
return ret
return rateLimitedFunction
return decorate
def printhelp():
print('This is a Python 3 script to count the total unique client MAC addresses connected to MR access points for')
print(' an organization during the last month.')
print('')
print('Usage:')
print(' clientcount.py -k <api key> [-o <org name>]')
print('')
print('Parameters:')
print(' -k <api key> : Mandatory. Your Meraki Dashboard API key')
print(' -o <org name> : Optional. Name of the organization you want to process. Use keyword "/all" to explicitly')
print(' specify all orgs. Default is "/all"')
print('')
print('Example:')
print(' clientcount.py -k 1234 -o "Big Industries Inc"')
print('')
print('Notes:')
print(' * In Windows, use double quotes ("") to enter command line parameters containing spaces.')
#SECTION: Meraki Dashboard API communication functions
@RateLimited(0.3)
def getInventory(p_org):
#returns a list of all networks in an organization
try:
r = requests.get('https://api.meraki.com/api/v0/organizations/%s/inventory' % (p_org.id), headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 06: Unable to contact Meraki cloud')
return(None)
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
return(r.json())
@RateLimited(0.3)
def getNetworks(p_org):
#returns a list of all networks in an organization
try:
r = requests.get('https://api.meraki.com/api/v0/organizations/%s/networks' % (p_org.id), headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 07: Unable to contact Meraki cloud')
return(None)
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
return(r.json())
@RateLimited(0.3)
def getNetworks_clients(p_org, p_dev):
#returns a list of all networks in an organization
try:
r = requests.get('https://api.meraki.com/api/v0/organizations/%s/networks/%s/clients' % (p_org.id, p_dev), headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 08: Unable to contact Meraki cloud')
return(None)
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
return(r.json())
@RateLimited(0.3)
def getNetworks_clients_id(p_org, p_dev, p_id):
#returns a list of all networks in an organization
try:
r = requests.get('https://api.meraki.com/api/v0/organizations/%s/networks/%s/clients/%s' % (p_org.id, p_dev, p_id), headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 08: Unable to contact Meraki cloud')
return(None)
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
return(r.json())
@RateLimited(0.3)
def getOrgs():
#returns the organizations' list for a specified admin, with filters applied
try:
r = requests.get('https://api.meraki.com/api/v0/organizations', headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 01: Unable to contact Meraki cloud')
return(None)
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
rjson = r.json()
orglist = []
listlen = -1
if ARG_ORGNAME.lower() == '/all':
for org in rjson:
orglist.append(c_Organization())
listlen += 1
orglist[listlen].id = org['id']
orglist[listlen].name = org['name']
else:
for org in rjson:
if org['name'] == ARG_ORGNAME:
orglist.append(c_Organization())
listlen += 1
orglist[listlen].id = org['id']
orglist[listlen].name = org['name']
return(orglist)
@RateLimited(0.3)
def getShardHost(p_org):
#Looks up shard URL for a specific org. Use this URL instead of 'api.meraki.com'
# when making API calls with API accounts that can access multiple orgs.
#On failure returns None
try:
r = requests.get('https://api.meraki.com/api/v0/organizations/%s/snmp' % p_org.id, headers={'X-Cisco-Meraki-API-Key': ARG_APIKEY, 'Content-Type': 'application/json'} )
except:
print('ERROR 08: Unable to contact Meraki cloud')
return None
if r.status_code != requests.codes.ok:
print(f'Rate limited activated - Retrying after {r.headers["Retry-After"]}.')
time.sleep(int(r.headers['Retry-After']))
#return(None)
raise Exception('API response: {}'.format(r.status_code))
return response
rjson = r.json()
return(rjson['hostname'])
@RateLimited(0.3)
def refreshOrgList():
global ORG_LIST
cnt_80211b = 0
cnt_80211n = 0
cnt_80211ac = 0
cnt_80211ax = 0
cnt = 0
print('INFO: Starting org list refresh at %s...' % datetime.datetime.now())
orglist = getOrgs()
if not orglist is None:
for org in orglist:
print('INFO: Processing org "%s"' % org.name)
org.shard = 'api.meraki.com'
orgshard = getShardHost(org)
if not orgshard is None:
org.shard = orgshard
netlist = getNetworks(org)
devlist = getInventory(org)
for device in netlist:
if device['name'] is not None:
print (device['name'])
clidev = device['id']
client_listlist = getNetworks_clients (org, clidev)
for client in client_listlist:
if client['ssid'] is not None:
client_id = client['id']
client_spec = getNetworks_clients_id(org, clidev, client_id)
if "802.11b" in client_spec['wirelessCapabilities']:
cnt_80211b = cnt_80211b + 1
elif "802.11n" in client_spec['wirelessCapabilities']:
cnt_80211n = cnt_80211n + 1
elif "802.11ac" in client_spec['wirelessCapabilities']:
cnt_80211ac = cnt_80211ac + 1
elif "802.11ax" in client_spec['wirelessCapabilities']:
cnt_80211ax = cnt_80211ax + 1
print ("802.11b")
print (cnt_80211b)
print ("802.11n")
print (cnt_80211n)
print ("802.11ac")
print (cnt_80211ac)
print ("802.11ax")
print (cnt_80211ax)
LAST_ORGLIST_REFRESH = datetime.datetime.now()
print('INFO: Refresh complete at %s' % LAST_ORGLIST_REFRESH)
return None
#SECTION: main
def main(argv):
global ARG_APIKEY
global ARG_ORGNAME
#initialize command line arguments
ARG_APIKEY = ''
ARG_ORGNAME = ''
arg_numresults = ''
arg_mode = ''
arg_filter = ''
#get command line arguments
try:
opts, args = getopt.getopt(argv, 'hk:o:m:')
except getopt.GetoptError:
printhelp()
sys.exit(2)
for opt, arg in opts:
if opt == '-h':
printhelp()
sys.exit()
elif opt == '-k':
ARG_APIKEY = arg
elif opt == '-o':
ARG_ORGNAME = arg
elif opt == '-m':
arg_mode = arg
#check that all mandatory arguments have been given
if ARG_APIKEY == '':
printhelp()
sys.exit(2)
#set defaults for empty command line arguments
if ARG_ORGNAME == '':
ARG_ORGNAME = '/all'
refreshOrgList()
if __name__ == '__main__':
main(sys.argv[1:])
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
The thing is, in the end, you can not avoid getting rate limited. If you submit more than 5 requests per second, it triggers the rate limit.
What you can do, is just be sure to handle it accordingly. That, or you can make sure to throttle your script manually but putting in sleep timers.
Or you can look into Action Batches
Like what you see? - Give a Kudo ## Did it answer your question? - Mark it as a Solution 🙂
All code examples are provided as is. Responsibility for Code execution lies solely your own.
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Hello! What was the solution? The link is no longer usable
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Hello RaresPauna,
As mentioned in the chat here is a Python class i created to help with the Meraki API ratelimit.
import sys, requests, os, json
import pandas as pd
from requests_futures.sessions import FuturesSession
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
class MyHTTPSession:
def __init__(self, ARG_APIKEY=None, URL_LIST=[]):
assert ARG_APIKEY, f'{self.__class__.__name__} ARG_APIKEY cannot be empty'
self.ARG_APIKEY = ARG_APIKEY
self.BASE_URL = "https://api.meraki.com/api/v1"
# Create list of urls for further processing
self.urls = list(self.divide_chunks(URL_LIST))
retry_strategy = Retry(
total=4,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.http = requests.Session()
self.http.mount("https://", adapter)
self.http.mount("http://", adapter)
self.headers={
'X-Cisco-Meraki-API-Key': self.__dict__['ARG_APIKEY'],
'Content-Type': 'application/json',
'Accept-Encoding': 'gzip'
}
def divide_chunks(self, l):
n = 5
# looping till length l
for i in range(0, len(l), n):
yield l[i:i + self.n]
def count_len(self, l):
count = 0
for i in range(0, len(l), self.n):
count+=1
self.netlist_len = count
def clear_console(self):
platforms = {
'linux1' : 'Linux',
'linux2' : 'Linux',
'win32' : 'Windows'
}
if sys.platform not in platforms:
raise f" Platform:{sys.platform} is not in the list clear_console"
else:
if platforms[sys.platform] == 'linux':
os.system('clear')
elif platforms[sys.platform] == 'Windows':
os.system('cls')
def process_bar(self):
try:
precent = 100 * (self.cnt / self.netlist_len)
except ZeroDivisionError:
precent = 0
bar = '#' * int(precent) + '-' * (100 - int(precent))
print(f"\r|{bar}| {precent:.2f}%", end="\r")
def get_data(self):
self.clear_console()
df = pd.DataFrame()
end_df = pd.DataFrame()
self.__dict__.setdefault('json_dept', 1)
self.__dict__.setdefault('data', '')
self.cnt = 0
self.process_bar()
self.cnt += 1
for urls in self.urls:
futures = [(url[1],self.http.get(self.BASE_URL + url[0], data=json.dumps(self.data), headers=self.headers)) for url in urls]
self.process_bar()
self.cnt +=1
for future in futures:
try:
r = future[1]
if r.status_code == 200:
if len(r.json()) != 0:
df = pd.json_normalize(r.json(), max_level=self.json_dept, errors='ignore')
for key, value in future[0].items():
df[key] = value
df = df.fillna(value=np.nan)
df = df.astype(str)
while 'next' in r.links:
r = self.http.get(r.links['next']['url'], headers=self.headers)
if len(r.json()) != 0:
dff = pd.json_normalize(r.json(), max_level=self.json_dept, errors='ignore')
for key, value in future[0].items():
dff[key] = value
dff = dff.fillna(value=np.nan)
dff = dff.astype(str)
df = pd.concat([df, dff])
elif r.status_code == 404:
print("Page not found")
except requests.exceptions.RequestException as err:
print(f"Error occurred during HTTP request: {err}")
continue
end_df = pd.concat([end_df, df])
self.response = end_df
This class takes a list of lists containing urls to check.
As an example, I'm using the url endpoint '/organizations/{Organization IDs}/networks' to collect all networks inside an organization.
lst = []
records = <Source from where to select>
for row in records:
URL = f"/organizations/{<Organization IDs>}/networks"
lst.append([URL, {}])
my_meraki_session = MyHTTPSession(ARG_APIKEY='123456789', URL_LIST=lst)
df_result = my_meraki_session.get_data()
Records is a list of the organization IDs that the API key can access.
There's an extra dictory [] after the url in the lst. This dictionary can be used to add more information to the Pandas dataframe later on. Things like organization names
Example:
records = Mysql_connection(dictionary=None).Select('ORGIDS', 'id', 'name' ,DISTINCT='DISTINCT')
for row in records:
URL = f"/organizations/{row[0]}/devices"
lst.append([URL, {'organizationId': row[0], 'organizationName':row[1]}])
MyHTTPSession is a custom HTTP session class that extends the requests.Session class. The class is initialized with two parameters: ARG_APIKEY and URL_LIST. The ARG_APIKEY is a mandatory parameter that represents the API key for the Meraki dashboard API. URL_LIST is an parameter that is a list of URLs that will be processed.
The class has several methods.
Init(self, ARG_APIKEY=None, URL_LIST=[]): This is the class constructor that initializes the class properties ARG_APIKEY, BASE_URL, urls, http, and headers. It also initializes the Retry and HTTPAdapter objects that will be used for handling retries and HTTP requests. It divides the URL_LIST into chunks and sets the chunk size to 5. It also sets default values for json_dept, data, and netlist_len.
Divide_chunks(self, l): This method is used to divide the URL_LIST into chunks of size 5. It takes in a list l as a parameter and uses a yield statement to return chunks of size 5.
Count_len(self, l): This method is used to count the number of chunks in the URL_LIST. It takes in a list l as a parameter and divides the list into chunks of size 5. It then calculates the number of chunks and sets the result to the netlist_len property.
Clear_console(self): This method is used to clear the console. It uses the os and sys modules to determine the platform and clear the console accordingly.
Process_bar(self): This method is used to display a progress bar. It calculates the percentage of completion and displays the progress bar in the console.
Get_data(self): This is the main method that retrieves data from the Meraki dashboard API. It initializes two data frames, end_df and end_df. It sets the JSON_dept and data properties if they are not already set. It sets the cnt property to 0 and displays the progress bar. It then iterates over the urls and creates a list of futures using the http.get method. It then iterates over the futures and processes the responses. If the status code is 200, it normalizes the JSON response using the PD.json_normalize method and adds any additional columns from the future. If there are more pages in the response, it continues to retrieve data until all pages are retrieved. It then concatenates the data frame df with end_df. If the status code is 404, it prints an error message. If there is an error during the HTTP request, it prints an error message and continues. Finally, it sets the response property to end_df.