Ausnahmen werden von asyncio queue.join () verborgen, die versteckt werden. ⇐ Python

[Anonymous]

Ich verwende einen API-Client, der von einem Anbieter (OKTA) geliefert wird, der sehr schlechte/alte Beispiele für das Laufen mit Async hat-beispielsweise (in der Python-Dokumentation heißt es, Get_Event_loop () ):
">"> ">"> ">"> ">"> ">

Code: Select all

from okta.client import Client as OktaClient
import asyncio

async def main():
client = OktaClient()
users, resp, err = await client.list_users()
while True:
for user in users:
print(user.profile.login) # Add more properties here.
if resp.has_next():
users, err = await resp.next()
else:
break

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
< /code>
Dies funktioniert, aber ich muss die zurückgegebenen Ergebnisse durchgehen und verschiedene Links befolgen, um zusätzliche Informationen zu erhalten. Ich habe eine Warteschlange mit Asyncio 

erstellt und habe die Arbeiterschleife, bis die Warteschlange leer ist. Dies funktioniert auch.

Code: Select all

async def handle_queue(name, queue: asyncio.Queue, okta_client: OktaClient):
"""Handle queued API requests"""
while True:
log.info("Queue size: %d", queue.qsize())
api_req = await queue.get()
log.info('Worker %s is handling %s', name, api_req)

api_func = getattr(okta_client, f"list_{api_req['endpoint']}")
api_procs = getattr(sys.modules[__name__], api_req['processor'])
log.info('Worker %s is handling %s with api_func %s, api_proc %s', name, api_req, api_func, api_proc)

resp_data, resp, err = await api_func(**api_req['params'])
log.debug(resp_data)
while True:
for i in resp_data:
await api_proc(i, queue)
if resp.has_next():
resp_data, err = await resp.next()
else:
break

queue.task_done()

async def create_workers(queue: asyncio.Queue):
"""Reusable worker creation process"""
log.info('Creating workers')
workers = []
async with OktaClient() as okta_client:
for i in range(NUM_WORKERS):
log.info('Creating worker-%d', i)
worker = asyncio.create_task(handle_queue(f'worker-{i}', queue, okta_client))
workers.append(worker)

await queue.join()
for worker in workers:
worker.cancel()

await asyncio.gather(*workers, return_exceptions=True)

async def main():
"""Load Access Policies and their mappings and rules"""
queue = asyncio.Queue()
queue.put_nowait({'endpoint': 'policies', 'params': {'query_params': {'type': 'ACCESS_POLICY'}}, 'processor': 'process_policy'})

await create_workers(queue)

metadata['policy_count'] = len(data)
print(yaml.dump({'_metadata': metadata, 'data': data}))

if __name__ == '__main__':
try:
asyncio.run(main())
except KeyboardInterrupt:
# Hide the exception for a Ctrl-C
log.info('Keyboard Interrupt')

Wenn eine Ausnahme in handle_queue (oder eine der von ihm aufgerufenen Funktionen) geworfen wird, hängt das Programm. Wenn ich auf Strg-C drücke, erhalte ich die Ausnahme zusammen mit einer Nachricht, die Asyncio-Task-Ausnahme nie abgerufen wurde . Ich verstehe, dass dies daran liegt, dass queue.join () auf queue.task_done () so oft wie Queue.put () aufgerufen werden soll, aber ich verstehe nicht, warum die Ausnahme nicht gefangen wird. PrettyPrint-Override ">async def handle_queue(name, queue: asyncio.Queue, okta_client: OktaClient):
"""Handle queued API requests"""
while True:
try:
# REST OF THE FUNCTION
except Exception as e:
queue.task_done()
raise e
queue.task_done()
< /code>
Auf diese Weise wird die Programmausführung abgeschlossen, aber die Ausnahme verschwindet immer noch.>