fork(3) download
  1. import asyncio
  2.  
  3.  
  4. def create_interval(coro, ms):
  5. async def inner():
  6. i = 0
  7. while True:
  8. print(f'{i} round')
  9. await asyncio.sleep(ms/1000)
  10. await coro()
  11. i += 1
  12. return asyncio.create_task(inner())
  13.  
  14.  
  15. def task_error_handler(cb):
  16. def decorator(func):
  17. async def wrapper(*args, **kwargs):
  18. try:
  19. result = await func(*args, **kwargs)
  20. except Exception as exc:
  21. result = await cb(exc)
  22. return result
  23. return wrapper
  24. return decorator
  25.  
  26.  
  27. async def simple_err_handler(exc):
  28. result = {'error': str(type(exc))}
  29. print(f'EXC OCCURED: {exc}')
  30. await asyncio.sleep(0.5)
  31. return result
  32.  
  33.  
  34. async def test1():
  35. print('START TEST1')
  36.  
  37. @task_error_handler(simple_err_handler)
  38. async def inner(name):
  39. print(f'{name} STARTED')
  40. await asyncio.sleep(0.5)
  41. raise RuntimeError('u suck')
  42. print('NEVER SHOULD BE PRINTED')
  43. await asyncio.sleep(0.5)
  44.  
  45. t1 = asyncio.create_task(inner('Beta'))
  46. await asyncio.sleep(2)
  47. t1.cancel()
  48. print('TASK CANCELLED')
  49. await asyncio.sleep(0.5)
  50. print('END TEST1')
  51.  
  52.  
  53. async def test2():
  54. print('START TEST2')
  55.  
  56. @task_error_handler(simple_err_handler)
  57. async def inner(name):
  58. print(f'{name} STARTED')
  59. await asyncio.sleep(0.5)
  60. raise RuntimeError('u suck')
  61. print('NEVER SHOULD BE PRINTED')
  62. await asyncio.sleep(0.5)
  63.  
  64. t1 = create_interval(inner('Charly'), 500)
  65. await asyncio.sleep(2)
  66. t1.cancel()
  67. print('TASK CANCELLED')
  68. await asyncio.sleep(0.5)
  69. print('END TEST2')
  70.  
  71.  
  72. def main():
  73. asyncio.run(test1()) # не упадет
  74. asyncio.run(test2()) # тихо сдыхает
  75.  
  76.  
  77. if __name__ == '__main__':
  78. main()
Success #stdin #stdout #stderr 0.09s 15936KB
stdin
Standard input is empty
stdout
START TEST1
Beta STARTED
EXC OCCURED: u suck
TASK CANCELLED
END TEST1
START TEST2
0 round
TASK CANCELLED
END TEST2
stderr
/usr/lib/python3.7/asyncio/events.py:88: RuntimeWarning: coroutine 'task_error_handler.<locals>.decorator.<locals>.wrapper' was never awaited
  self._context.run(self._callback, *self._args)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback