¿Cómo puedo utilizar logging.exception() para las excepciones capturadas por asyncio.gather()?

Answer

1. 

   davy.ai

   August 2, 2023 at 3:00 am

   Una forma de lograr esto es mediante el uso del método sys.exc_info(), que devuelve información sobre la última excepción lanzada. Aquí tienes un código de ejemplo:

   import logging
   import sys
   import asyncio
   
   logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')
   
   async def my_coroutine():
       raise ValueError('Ocurrió un error')
   
   async def main():
       try:
           await asyncio.gather(my_coroutine(), return_exceptions=True)
       except Exception:
           exc_info = sys.exc_info()
           logging.error('Se produjo una excepción no controlada', exc_info=exc_info)
   
   if __name__ == '__main__':
       asyncio.run(main())
   

   En el código anterior, definimos una corutina que lanza un ValueError. Luego utilizamos asyncio.gather() con return_exceptions=True para ejecutar la corutina y capturar cualquier excepción que pueda ocurrir.

   Envolvemos la llamada a asyncio.gather() en un bloque try-except para capturar cualquier excepción no controlada. Cuando se produce una excepción no controlada, llamamos a sys.exc_info() para obtener detalles sobre la excepción y luego la registramos utilizando logging.error() con el parámetro exc_info configurado como exc_info. Esto registrará la excepción con la misma información que logging.exception().

   Este enfoque se puede utilizar para registrar cualquier excepción que ocurra en una aplicación asyncio, ya sea que se lancen en una corutina o en otro lugar.

Comments are closed.