Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import asyncio
- import sys
- import logging
- logger = logging.getLogger()
- logger.setLevel(logging.DEBUG)
- sh = logging.StreamHandler()
- sh.setLevel(logging.DEBUG)
- logger.addHandler(sh)
- def custom_exception_handler(loop, context):
- loop.default_exception_handler(context); sys.stdout.flush()
- loop.stop()
- # exception = context.get('exception')
- # raise exception
- loop = asyncio.get_event_loop()
- loop.set_exception_handler(custom_exception_handler)
- class A:
- async def wait_and_fail(self):
- await asyncio.sleep(1)
- raise ValueError
- async def run(self, how_to_set: str):
- logger.info('task = asyncio.get_event_loop().create_task(self.wait_and_fail())')
- if how_to_set == 'field':
- self.task = asyncio.get_event_loop().create_task(self.wait_and_fail())
- elif how_to_set == 'none':
- asyncio.get_event_loop().create_task(self.wait_and_fail())
- elif how_to_set == 'var':
- task = asyncio.get_event_loop().create_task(self.wait_and_fail())
- elif how_to_set == 'list':
- [asyncio.get_event_loop().create_task(self.wait_and_fail())]
- elif how_to_set == 'list_field':
- self.tasks = [asyncio.get_event_loop().create_task(self.wait_and_fail())]
- else:
- raise ValueError
- for i in range(5):
- await asyncio.sleep(1)
- logger.info(f'i = {i}')
- logger.info('run success')
- async def run_a_and_del(how_to_set):
- logger.info('start run_a_and_del')
- a = A()
- logger.info('start await a.run(set_task=set_task)')
- await a.run(how_to_set=how_to_set)
- if hasattr(a, 'task'):
- logger.info('start del a.task')
- del a.task
- if hasattr(a, 'tasks'):
- logger.info('start del a.tasks')
- del a.tasks
- logger.info('start del a')
- del a
- async def main():
- # different approaches
- if 1: # it fails with exception immediately with exit_code=1
- a = A()
- await a.run(how_to_set='none')
- elif 0: # it finish execution and print exception in the end with exit_code=0
- a = A()
- await a.run(how_to_set='field')
- elif 0: # it finish execution, destruct A instance and print exception in the end with exit_code=0
- await run_a_and_del(how_to_set='field')
- elif 0: # it finish execution and print exception in the end with exit_code=0
- a = A()
- await a.run(how_to_set='var')
- elif 0: # it fails with exception immediately with exit_code=1
- a = A()
- await a.run(how_to_set='list')
- elif 0: # it finish execution and print exception in the end with exit_code=0
- a = A()
- await a.run(how_to_set='list_field')
- logger.info('main finish')
- if __name__ == '__main__':
- asyncio.get_event_loop().run_until_complete(main())
Add Comment
Please, Sign In to add comment