That is very strange. It might be worth trying to run a script from a different system just to check there is not something broken in the one you are using. The rate limiting/retry really does work. Pretty much every script in our analytics platform relies on this working flawlessly, day after day, month after month, year after year. Script below is a simple example, you can use as-is if you have some APs, or adapt it. I just ran it, there are over 1700 APs in the target org, these are all called 'at once' via aio, but the rate-limiting kicks in to save things. It takes a bit less than three minutes to complete, so the effective call rate is just under 10 calls/second, that is the wait/retry in action. If I change retries to maximum of 1, it will fail in just a few seconds. If you do not have lots of APs, just replace the call used to another one that will target something you have many of. import os
import sys
import meraki.aio
import asyncio
#import the org id and api key from the environment
#or you could hard code them, but that's less desirable
ORG_ID = os.environ.get("PARA0")
API_KEY = os.environ.get("PARA1")
async def processAp(aiomeraki: meraki.aio.AsyncDashboardAPI, ap):
try:
# get list of statuses for an AP
statuses = await aiomeraki.wireless.getDeviceWirelessStatus(ap['serial'])
except meraki.AsyncAPIError as e:
print(f'Meraki API error: {e}', file=sys.stderr)
sys.exit(0)
except Exception as e:
print(f'some other error: {e}', file=sys.stderr)
sys.exit(0)
for bss in statuses['basicServiceSets']:
if bss['enabled']:
print(f"{ap['name']},{bss['ssidName']},{bss['bssid']},{bss['band']}")
return
async def main():
async with meraki.aio.AsyncDashboardAPI(
api_key=API_KEY,
base_url='https://api.meraki.com/api/v1/',
print_console=False,
output_log=False,
suppress_logging=True,
wait_on_rate_limit=True,
maximum_retries=100
) as aiomeraki:
#get the wireless devices
try:
aps = await aiomeraki.organizations.getOrganizationDevices(ORG_ID, perPage=1000, total_pages="all", productTypes = ["wireless"])
except meraki.AsyncAPIError as e:
print(f'Meraki API error: {e}', file=sys.stderr)
sys.exit(0)
except Exception as e:
print(f'some other error: {e}', file=sys.stderr)
sys.exit(0)
# process devices concurrently
apTasks = [processAp(aiomeraki, ap) for ap in aps]
for task in asyncio.as_completed(apTasks):
await task
if __name__ == '__main__':
asyncio.run(main())
... View more
