Upgrade to Pro — share decks privately, control downloads, hide ads and more …

New in Python 3.11: asyncio.TaskGroup and “Hell...

New in Python 3.11: asyncio.TaskGroup and “Hello-ish World” of Asyncio

New in Python 3.11: asyncio.TaskGroup and “Hello-ish World” of Asyncio
[PyCon JP 2022](https://2022.pycon.jp/timetable/?id=J7S98E)

Junya Fukuda

October 18, 2022
Tweet

More Decks by Junya Fukuda

Other Decks in Technology

Transcript

  1. Who are you? •Junya Fukudaʢ@JunyaFffʣ • Japan belongs to Japan

    System Laboratories, Inc.(JSL) Company in Nagano Prefecture •Web Engineer •GEEKLAB.NAGANO - Community space management •Valuebooks, inc •Aiming to be a service loved by book lovers (used book purchase and sales) Here is today's code
  2. Python Practical Recipes •Launched in January 2022 Gijutsu-Hyoron Co., Ltd

    •Python 3 Engineer Certi fi cation Practice Exam •Use it like a dictionary! • I am responsible for the following chapters • Chapter2 coding rule • Chapter5 Type hints • Chapter19 Concurrency, Parallel Processing
  3. Review - asyncio basics •Of fi cial library for writing

    asynchronous processing code •Evolved with each version High level API - very easy to write •I gave such a talk in my book and at PyCon JP 2021. Please do! "4(*ΞϓϦέʔγϣϯೖ໳͜Θ͘ͳ͍BTZODJPجૅͱඇಉظ*0'BTU"1*Λྫʹ 1Z$PO+1 •Handle multiple tasks concurrent •Effective for I/O-bound processing (e.g., DB access/HTTP requests)
  4. asyncio I've been trying to think of reasons not to

    use it. •The synchronous process I'm used to is suf fi cient. •No IO (mostly machine learning and data analysis) (lots of CPU hard work) •Asynchronous processing seems to have unpredictable behavior. •Write in other languages (Go, JS(nodejs/deno...), Rust, C#, Swift, etc.)
  5. Do you have anything to say or be said in

    a review? "What happens when an exception occurs in this process?"
  6. Are you using asyncio? •Asynchronous processing seems to have unpredictable

    behavior. •that's true •Likely to move unpredictably •I don't know what happens when an exception happens. •What happens if I EXCEPT this parallel running task?" •When an exception occurs, what are the other tasks... •A new feature that solves this
  7. Today's Goal •Exception handling and cancellation in using asyncio •Previous

    (Python 3.10 and earlier) and future (Python 3.11), respectively •Exceptionally easy to write! •This is how cancellations work. •How to write "predictable" and "safe" asyncio •“Hello-ish world” •TaskGroup Let's give it a try.
  8. Motivation (why are we talking about this?) •Two big ones.

    •Python Core Developer Yury Selivanov tweetʢasync/await Authorʣ •When Taskgroup was MERGE, they were posting a series of excited tweets. •I don't understand it after reading the documentation. •I don't know what's so great about it 🤔
  9. Motivation (why are we talking about this?) •Using Asyncio in

    Python, a book that has been a great help to us. •There was a code I didn't understand. •Two big ones. •Quickstart - Example 3-2.
  10. •Quickstart - Example 3-2. import asyncio import time async def

    main(): print(f"{time.ctime()} Hello!") await asyncio.sleep(1.0) print(f"{time.ctime()} GoodBye!") loop = asyncio.get_event_loop() task = loop.create_task(main()) loop.run_until_complete(task) pending = asyncio.all_tasks(loop=loop) for task in pending: task.cancel() group = asyncio.gather(*pending, return_exceptions=True) loop.run_until_complete(group) loop.close()
  11. Motivation (why are we talking about this?) •Using Asyncio in

    Python, a book that has been a great help to us. •There was a code I didn't understand. •Two big ones. •Quickstart - Example 3-2. The “Hello-ish World” of Asyncio •I would like to revamp this code with asyncio.TaskGroup
  12. What to say/not to say •Talk. •Exception handling in asyncio

    (also new except* syntax) •Cancel Task •asyncio's New Hello World •Don't talk. •asyncio basics no more •Speci fi c examples in the application •What's happening at ASGI IUUQTHJUIVCDPNKSGLUBML
  13. Attention! •Python 3.11 has not been released yet! •It was

    scheduled for release on October 3, but is now scheduled for October 24. •Operation is con fi rmed with Python 3.11.0rc2 IUUQTQFQTQZUIPOPSHQFQ •I don't think so, but there is no possibility that asyncio.TaskGroup is a dream function... •This talk is designed to disappear from your mind.
  14. •Let's start with the of fi cial documentation •Features of

    asyncio.TaskGroup •Handle multiple tasks at once •It's in Python 3.10 and earlier. •asyncio.gather() and asyncio.wait() asycio.TaskGroup
  15. import asyncio async def main(): async with asyncio.TaskGroup() as tg:

    task1 = tg.create_task(some_coro(...)) task2 = tg.create_task(another_coro(...)) print("Both tasks have completed now.") asycio.TaskGroup
  16. •Features of asyncio.TaskGroup •Handle multiple tasks at once •It's in

    Python 3.10 and earlier. •asyncio.gather() and asyncio.wait() •There are two major features that distinguish it from them asycio.TaskGroup •Catching Concurrent Exceptions •Cancel task when exception occurs
  17. •Handle multiple tasks at once Catching Concurrent Exceptions •Naturally, exceptions

    may occur at the same time •Let's see how to capture it in Python 3.10 or earlier •Using the high-level API asyncio.gather() added in Python 3.7 •Options for exceptions to asyncio.gather()ʮreturn_exceptionsʯ •return_exceptions False •return_exceptions True •Simultaneous execution of speci fi ed tasks (asynchronous functions)
  18. •Prepare asynchronous functions to run concurrently - 3 coroutines Advance

    preparation - no change import asyncio async def coro_success(): return "੒ޭ" async def coro_value_err(): raise ValueError async def coro_type_err(): raise TypeError
  19. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  20. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  21. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  22. $ python310 1_asyncio_gather_except.py async def main(): """return_exceptionsͳ͠""" try: results =

    await asyncio.gather( coro_success(), coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  23. $ python310 1_asyncio_gather_except.py err=ValueError() $ python310 async def main(): """return_exceptionsͳ͠"""

    try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  24. $ python310 1_asyncio_gather_except.py err=ValueError() $ python310 async def main(): """return_exceptionsͳ͠"""

    try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions ValueError only TypeError cannot be caught
  25. $ python310 1_asyncio_gather_except.py err=ValueError() $ python310 ValueError only TypeError cannot

    be caught Only the fi rst exception is captured The rest are suppressed. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}") asyncio.run(main()) asyncio.gather - return_exceptions False Catching Concurrent Exceptions
  26. •asyncio.gather - return_exceptions False •asyncio.gather - return_exceptions True return_exceptions If

    speci fi ed Get a list of results after all tasks have been executed Need to check the return list.
  27. async def main(): """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(),

    coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  28. async def main(): """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(),

    coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  29. async def main(): """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(),

    coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  30. $ python310 2_asyncio_gather_except_true.py async def main(): """return_exceptions=True͋Γ""" results = await

    asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  31. $ python310 2_asyncio_gather_except_true.py async def main(): """return_exceptions=True͋Γ""" results = await

    asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") results=['੒ޭ', ValueError(), TypeError()] $ python310 asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  32. $ python310 2_asyncio_gather_except_true.py async def main(): """return_exceptions=True͋Γ""" results = await

    asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") results=['੒ޭ', ValueError(), TypeError()] $ python310 Get all results in a list Need to see what exceptions were raised asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  33. for result in results: match result: case ValueError(): print("ValueError") case

    TypeError(): print("TypeError") async def main(): """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  34. for result in results: match result: case ValueError(): print("ValueError") case

    TypeError(): print("TypeError") async def main(): """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  35. $ python310 2_asyncio_gather_except_true.py async def main(): """return_exceptions=True͋Γ""" results = await

    asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") for result in results: match result: case ValueError(): print("ValueError") case TypeError(): print("TypeError") asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  36. $ python310 2_asyncio_gather_except_true.py async def main(): """return_exceptions=True͋Γ""" results = await

    asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") results=['੒ޭ', ValueError(), TypeError()] ValueError() for result in results: match result: case ValueError(): print("ValueError") case TypeError(): print("TypeError") TypeError() asyncio.gather - return_exceptions True Catching Concurrent Exceptions
  37. asyncio.gather - return_exceptions True $ python310 2_asyncio_gather_except_true.py async def main():

    """return_exceptions=True͋Γ""" results = await asyncio.gather( coro_success(), coro_value_err(), coro_type_err(), return_exceptions=True) print(f"{results=}") results=['੒ޭ', ValueError(), TypeError()] ValueError() for result in results: match result: case ValueError(): print("ValueError") case TypeError(): print("TypeError") TypeError() Need to check results All tasks need to be performed Catching Concurrent Exceptions
  38. •For example, a process that makes 10 HTTP requests at

    the same time •Exception occurred in the fi rst task •We'll have to wait until the remaining nine that might fail are complete. (even if all 10 result in the same error) Catching Concurrent Exceptions Python 3.10 and earlier issues •Unable to capture except for the fi rst exception. •Cannot log all exceptions that occur •Need to check the return list. •All tasks must be performed •return_exceptions False •return_exceptions True
  39. Catching Concurrent Exceptions Python 3.10 and earlier issues •Unable to

    capture except for the fi rst exception. •Cannot log all exceptions that occur •Need to check the return list. •All tasks must be performed •return_exceptions False •return_exceptions True •The solution is asyncio.TaskGroup and the new "except*" syntax
  40. •Prepare asynchronous functions to run concurrently - 3 coroutines Advance

    preparation - no change import asyncio async def coro_success(): return "੒ޭ" async def coro_value_err(): raise ValueError async def coro_type_err(): raise TypeError
  41. async def main(): async with asyncio.TaskGroup() as tg: task1 =

    tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] Catching Exceptions in asycio.TaskGroup
  42. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] Catching Exceptions in asycio.TaskGroup
  43. async def main(): async with asyncio.TaskGroup() as tg: task1 =

    tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") try: Catching Exceptions in asycio.TaskGroup
  44. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Catching Exceptions in asycio.TaskGroup
  45. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Can be captured by except as a synchronous process Catching Exceptions in asycio.TaskGroup
  46. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Catching Exceptions in asycio.TaskGroup
  47. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") $ python311 3_asyncio_taskgroup_except.py Catching Exceptions in asycio.TaskGroup
  48. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") $ python311 3_asyncio_taskgroup_except.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) err=ExceptionGroup('unhandled errors in a TaskGroup', [TypeError()]) $ python311 Catching Exceptions in asycio.TaskGroup
  49. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") $ python311 3_asyncio_taskgroup_except.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) err=ExceptionGroup('unhandled errors in a TaskGroup', [TypeError()]) $ python311 Two Exception are captured. Catching Exceptions in asycio.TaskGroup
  50. asyncio.TaskGroup new syntaxʮexcept*ʯ •Can catch concurrent exceptions •Easy to read,

    written in the same way as synchronous processing This is the fi rst feature of asyncio.TaskGroup
  51. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Catching Exceptions in asycio.TaskGroup
  52. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Catching Exceptions in asycio.TaskGroup
  53. try: async def main(): async with asyncio.TaskGroup() as tg: task1

    = tg.create_task(coro_success()) task2 = tg.create_task(coro_value_err()) task3 = tg.create_task(coro_type_err()) results = [task1.result(), task2.result(), task3.result()] except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}") Catching Exceptions in asycio.TaskGroup
  54. except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f"{err=}")

    except* except* Catching Exceptions in asycio.TaskGroup
  55. Catching Exceptions in asycio.TaskGroup except* ValueError as err: print(f"{err=}") except*

    TypeError as err: print(f"{err=}") except* except* err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) err=ExceptionGroup('unhandled errors in a TaskGroup', [TypeError()])
  56. except* except* •PEP 654: Exception Groups and except*. •Python 3.11new

    syntax •Use with ExceptionGroup (added 3.11) for built-in exception types •ExceptionGroup •Use when you need to raise multiple unrelated exceptions •try except •try except* •ExceptionGroup(“message”, [ValueError(1), TypeError(2)]) new syntax
  57. asycio.TaskGroup IUUQTHJUIVCDPNQZUIPODQZUIPOCMPCNBJO-JCBTZODJPUBTLHSPVQTQZ class TaskGroup: ... async def __aexit__(self, et, exc,

    tb): ... if self._errors: # Exceptions are heavy objects that can have object # cycles (bad for GC); let's not keep a reference to # a bunch of them. errors = self._errors self._errors = None me = BaseExceptionGroup('unhandled errors in a TaskGroup', errors) raise me from None ExceptionGroup
  58. class TaskGroup: ... async def __aexit__(self, et, exc, tb): ...

    if self._errors: # Exceptions are heavy objects that can have object # cycles (bad for GC); let's not keep a reference to # a bunch of them. errors = self._errors self._errors = None me = BaseExceptionGroup('unhandled errors in a TaskGroup', errors) raise me from None IUUQTHJUIVCDPNQZUIPODQZUIPOCMPCNBJO-JCBTZODJPUBTLHSPVQTQZ asycio.TaskGroup ExceptionGroup
  59. IUUQTHJUIVCDPNQZUIPODQZUIPOCMPCNBJO-JCBTZODJPUBTLHSPVQTQZ class TaskGroup: ... async def __aexit__(self, et, exc, tb):

    ... if self._errors: # Exceptions are heavy objects that can have object # cycles (bad for GC); let's not keep a reference to # a bunch of them. errors = self._errors self._errors = None me = BaseExceptionGroup('unhandled errors in a TaskGroup', errors) raise me from None asycio.TaskGroup ExceptionGroup
  60. new syntax except* except* •PEP 654: Exception Groups and except*.

    •PEP654 Motivation 4 cases •Concurrent errors Born for what? •There is no good way to handle exceptions sent by multiple tasks. •asyncio.gather()
  61. •except* syntax is implemented for asynchronous processing ? •Python 3.11ͷ৽ػೳ(ͦͷ4ʣPEP

    654 ྫ֎άϧʔϓͱTaskGroup •https://www.python.jp/news/wnpython311/except-grp.html •The 2021 Python Language Summit: PEP 654 — Exception Groups and except* •https://pyfound.blogspot.com/2021/05/the-2021-python-language-summit-pep-654.html • How Exception Groups Will Improve Error Handling in AsyncIO - Łukasz Langa | Power IT Conference •https://www.youtube.com/watch?v=Lfe2zsGS0Js new syntax except* Born for what?
  62. •Catching Concurrent Exceptions •Cancel task when exception occurs •How do

    you cancel a task? •Python 3.10 or earlier - asyncio.gather() •Python 3.11 - asyncio.TaskGroup() Features of asyncio.TaskGroup
  63. Cancel task when exception occurs •If an exception occurs, what

    happens to the remaining tasks that were running at the same time? •In some cases, the remaining tasks are executed as they are. •Let's take asyncio.gather() as an example of a case that remains
  64. import asyncio async def coro_success(): return "੒ޭ" async def coro_value_err():

    raise ValueError async def coro_type_err(): raise TypeError มߋͳ͠ มߋͳ͠ •Add one asynchronous function - wait a bit Preparation - add one asynchronous function
  65. •Add one asynchronous function - wait a bit Preparation -

    add one asynchronous function import asyncio async def coro_success(): return "੒ޭ" async def coro_value_err(): raise ValueError async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" มߋͳ͠ มߋͳ͠ ௥Ճ
  66. Cases where tasks remain - asyncio.gather - return_exceptions false async

    def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main())
  67. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) Cases where tasks remain - asyncio.gather - return_exceptions false
  68. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) Cases where tasks remain - asyncio.gather - return_exceptions false
  69. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) Cases where tasks remain - asyncio.gather - return_exceptions false
  70. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py Cases where tasks remain - asyncio.gather - return_exceptions false
  71. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ Cases where tasks remain - asyncio.gather - return_exceptions false
  72. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ Cases where tasks remain - asyncio.gather - return_exceptions false
  73. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ If an exception occurs Other tasks remain running Hard to predict behavior Cases where tasks remain - asyncio.gather - return_exceptions false
  74. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ I want to stop. I want to roll back Cases where tasks remain - asyncio.gather - return_exceptions false If an exception occurs Other tasks remain running Hard to predict behavior
  75. λεΫ͕࢒ͬͯ͠·͏έʔε - asyncio.gather - return_exceptions ࢦఆͳ͠ async def main(): """return_exceptionsͳ͠"""

    try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ ɾࢭΊ͍ͨ ɾϩʔϧόοΫ͍ͨ͠ ྫ֎͕ൃੜͨ͠৔߹ ଞͷλεΫ͸ಈ͍ͨ·· 
 ಈ࡞Λ༧ଌͮ͠Β͍ I want to stop. I want to roll back If an exception occurs Other tasks remain running Hard to predict behavior
  76. Cancel task when exception occurs •If an exception occurs, what

    happens to the remaining tasks that were running at the same time? •In some cases, the remaining tasks are executed as they are. •Cancel a task (Cancellations were not added on 3.11) •Cancellation Basics •Cancel when using asyncio.gather() before 3.10 •Let's fi x the process that leaves the task
  77. Cancel task when exception occurs •Let's fi x the process

    that leaves the task •Asynchronous function (coro_long()) to be canceled •Instructions to check and cancel remaining tasks •Wait for post-cancellation processing to complete. Added post-processing for cancellations •There are three perspectives for modi fi cation
  78. •task.cancel() •Cancel request to a task •CancelledError exception is sent

    to the cancelled task •receive asyncio.CancelledError in an asynchronous function •Resend asyncio.CancelledError •I'm a little confused. Let's try to implement it. Added post-processing for cancellations
  79. import asyncio async def coro_success(): return "੒ޭ" async def coro_value_err():

    raise ValueError async def coro_long(): มߋͳ͠ มߋͳ͠ มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" •Therefore, it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  80. async def coro_long(): มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" •Therefore,

    it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  81. async def coro_long(): มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" try:

    •Therefore, it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  82. async def coro_long(): มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except

    asyncio.CancelledError as err: try: •Therefore, it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  83. async def coro_long(): มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except

    asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError •Therefore, it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  84. Cancel task when exception occurs •Let's fi x the process

    that leaves the task •Asynchronous function (coro_long()) to be canceled •Instructions to check and cancel remaining tasks •Wait for post-cancellation processing to complete. Added post-processing for cancellations •There are three perspectives for modi fi cation
  85. Cancel task when exception occurs •Let's fi x the process

    that leaves the task •Asynchronous function (coro_long()) to be canceled •Instructions to check and cancel remaining tasks •Wait for post-cancellation processing to complete. Added post-processing for cancellations •There are three perspectives for modi fi cation
  86. •Request a task to cancel - Task.cancel() Instructions to check

    and cancel remaining tasks •Determine completion of a task - Task.done() •Wait for post-cancel process to complete - asyncio.sleep() •Let's modify the code.
  87. async def main(): """return_exceptionsͳ͠""" try: print(f"{results=}") except ValueError as err:

    print(f"{err=}") except TypeError as err: print(f”{err=}”) results = await asyncio.gather( coro_success(), coro_value_err(), coro_long() ) asyncio.run(main()) print("ऴྃ") Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  88. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  89. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) λεΫͷঢ়ଶΛ ֬ೝ͠΍͘͢͢Δ ॲཧͱͯ͠͸ಉ͡ print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  90. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") for task in [task1, task2, task3]: if task.done() is False: Task.done()Ͱ ະ׬ྃͷλεΫΛಛఆ Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  91. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ ະ׬ྃͷλεΫʹର͠ Task.cancel()Λ࣮ߦ print("ऴྃ") Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  92. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ ׬ྃ଴ͪͷ asyncio.sleep() print("ऴྃ") await asyncio.sleep(1) Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  93. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ શମ͜Μͳײ͡ print("ऴྃ") await asyncio.sleep(1) Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  94. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ Cases where tasks remain - asyncio.gather - return_exceptions false Review Execution before modi fi cation
  95. async def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(),

    coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ Cases where tasks remain - asyncio.gather - return_exceptions false Review Execution before modi fi cation Output from remaining tasks
  96. Cases where tasks remain - asyncio.gather - return_exceptions false async

    def main(): """return_exceptionsͳ͠""" try: results = await asyncio.gather( coro_success(), coro_value_err(), coro_long() ) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) print("ऴྃ") asyncio.run(main()) $ python310 4_remaining_tasks.py err=ValueError() ऴྃ ׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢ Review Execution before modi fi cation Output from remaining tasks
  97. async def coro_long(): มߋ await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except

    asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError •Therefore, it is necessary to receive asyncio.CancelledError in an asynchronous function •In addition, resend asyncio.CancelledError in an asynchronous function Advance preparation - add post-processing in case of cancellation
  98. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) ࣮ߦͯ͠Έ·͠ΐ͏ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  99. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) $ python310 5_canceling_task_gather.py async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  100. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) $ python310 5_canceling_task_gather.py err=ValueError() async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  101. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) $ python310 5_canceling_task_gather.py err=ValueError() Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  102. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) $ python310 5_canceling_task_gather.py err=ValueError() Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError ऴྃ Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  103. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) Ωϟϯηϧ͞ΕΔͱ ඇಉظؔ਺ଆͰ ΩϟϯηϧΛัଊ $ python310 5_canceling_task_gather.py err=ValueError() Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError ऴྃ Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  104. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) If canceled On the asynchronous function side Catch Cancel $ python310 5_canceling_task_gather.py err=ValueError() Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError ऴྃ basic principle of cancellation Cases where tasks remain → Add instructions to con fi rm and cancel tasks Python 3.10 or earlier - asyncio.gather()
  105. Cases where tasks remain → Add instructions to con fi

    rm and cancel tasks async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 = asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) asyncio.run(main()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) Python 3.10 or earlier - asyncio.gather() $ python310 5_canceling_task_gather.py err=ValueError() Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError ऴྃ Checking task status Waiting for cancellation process
  106. Cancel task when exception occurs •Let's fi x the process

    that leaves the task •Asynchronous function (coro_long()) to be canceled •Instructions to check and cancel remaining tasks •Wait for post-cancellation processing to complete. Added post-processing for cancellations •There are three perspectives for modi fi cation
  107. •At the time of exiting the context manager, •Remaining tasks

    are canceled. •Wait for cancellation process to complete. •That's all! •In short, nothing special needs to be done. It will be canceled. Let's refactor the code we just wrote. Cancellation process in asycio.TaskGroup
  108. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) asyncio.run(main()) print(f"{results=}") for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) Cancellation process in asycio.TaskGroup Let's refactor it. results = await asyncio.gather(*[task1, task2, task3]) except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”)
  109. async def main(): """return_exceptionsͳ͠""" try: task1 = asyncio.create_task(coro_success()) task2 =

    asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) asyncio.run(main()) for task in [task1, task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) results = await asyncio.gather(*[task1, task2, task3]) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) Cancellation process in asycio.TaskGroup Let's refactor it.
  110. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: task1 = asyncio.create_task(coro_success()) task2 = asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) results = await asyncio.gather(*[task1, task2, task3]) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) Cancellation process in asycio.TaskGroup Let's refactor it.
  111. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: task1 = asyncio.create_task(coro_success()) task2 = asyncio.create_task(coro_value_err()) task3 = asyncio.create_task(coro_long()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) Cancellation process in asycio.TaskGroup Let's refactor it.
  112. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) Cancellation process in asycio.TaskGroup Let's refactor it.
  113. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  114. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: print(f"{results=}") except ValueError as err: print(f"{err=}") except TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  115. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  116. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) for task in [task1,

    task2, task3]: if task.done() is False: task.cancel() # ະ׬ྃͷλεΫΛΩϟϯηϧ print("ऴྃ") await asyncio.sleep(1) async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  117. async def main(): """return_exceptionsͳ͠""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup()

    as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  118. async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup()

    as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup Let's refactor it.
  119. Let's run it. async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return

    "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) $ python311 6_cancelled_task_taskgroup.py Cancellation process in asycio.TaskGroup
  120. Let's run it. async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return

    "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) $ python311 6_cancelled_task_taskgroup.py Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ Cancellation process in asycio.TaskGroup
  121. Let's run it. async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return

    "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) $ python311 6_cancelled_task_taskgroup.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ Cancellation process in asycio.TaskGroup
  122. Let's run it. async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return

    "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) $ python311 6_cancelled_task_taskgroup.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ ऴྃ Cancellation process in asycio.TaskGroup
  123. Catching Exceptions and Processing after canceling The order of catching

    async def coro_long(): await asyncio.sleep(1) print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) $ python311 6_cancelled_task_taskgroup.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ ऴྃ Cancellation process in asycio.TaskGroup
  124. In TaskGroup complete Predictable behavior async def coro_long(): await asyncio.sleep(1)

    print("׬͍ྃͯ͠ͳ͍λεΫ͕ग़ྗ͍ͯ͠·͢") return "੒ޭʁ" except asyncio.CancelledError as err: try: print("Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢") raise asyncio.CancelledError async def main(): """TaskGroupͰͷΩϟϯηϧ""" try: asyncio.run(main()) print("ऴྃ") async with asyncio.TaskGroup() as g: print(f"{results=}") except* ValueError as err: print(f"{err=}") except* TypeError as err: print(f”{err=}”) results = [task1.result(), task2.result(), task3.result()] task1 = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_long()) Cancellation process in asycio.TaskGroup $ python311 6_cancelled_task_taskgroup.py err=ExceptionGroup('unhandled errors in a TaskGroup', [ValueError()]) Ωϟϯηϧ͞ΕͨλεΫ͕ग़ྗ͍ͯ͠·͢ ऴྃ
  125. asycio.TaskGroup Features •Catching Concurrent Exceptions •Cancel task when exception occurs

    •Simply catch exceptions in a conventional way of writing •Cancel without indication •Except* will not spare any concurrent exceptions. •No longer need to be aware of tasks when executing them •How to write "predictable," "safe," and readable asyncio •New “Hello-ish world”
  126. import asyncio async def main(): async with asyncio.TaskGroup() as tg:

    task1 = tg.create_task(some_coro(...)) task2 = tg.create_task(another_coro(...)) New “Hello-ish world”
  127. import asyncio async def main(): async with asyncio.TaskGroup() as tg:

    task1 = tg.create_task(some_coro(...)) task2 = tg.create_task(another_coro(...)) try: except* BaseException as e: print(f"{e=}") asyncio.run(main()) New “Hello-ish world”
  128. New “Hello-ish world” import asyncio async def main(): async with

    asyncio.TaskGroup() as tg: task1 = tg.create_task(some_coro(...)) task2 = tg.create_task(another_coro(...)) try: except* BaseException as e: print(f"{e=}") asyncio.run(main()) Python 3.11
  129. •Quickstart - Example 3-2. The “Hello-ish World” of Asyncio import

    asyncio import time async def main(): print(f"{time.ctime()} Hello!") await asyncio.sleep(1.0) print(f"{time.ctime()} GoodBye!") loop = asyncio.get_event_loop() task = loop.create_task(main()) loop.run_until_complete(task) pending = asyncio.all_tasks(loop=loop) for task in pending: task.cancel() group = asyncio.gather(*pending, return_exceptions=True) loop.run_until_complete(group) loop.close()
  130. •Quickstart - Example 3-2. import asyncio import time async def

    main(): print(f"{time.ctime()} Hello!") await asyncio.sleep(1.0) print(f"{time.ctime()} GoodBye!") The “Hello-ish World” of Asyncio Checking task status Waiting for cancellation process loop = asyncio.get_event_loop() task = loop.create_task(main()) loop.run_until_complete(task) pending = asyncio.all_tasks(loop=loop) for task in pending: task.cancel() group = asyncio.gather(*pending, return_exceptions=True) loop.run_until_complete(group) loop.close()
  131. •Quickstart - Example 3-2. import asyncio import time async def

    coro(): print(f"{time.ctime()} Hello!") await asyncio.sleep(1.0) print(f"{time.ctime()} GoodBye!") The “Hello-ish World” of Asyncio Checking task status Waiting for cancellation process async def main(): async with asyncio.TaskGroup() as tg: task = tg.create_task(coro(...)) try: except* BaseException as e: print(f"{e=}") asyncio.run(main())
  132. •Catching and canceling exceptions in asyncio.TaskGroup •asyncio for external I/O

    •Because of external dependence, exceptions are likely to occur. •Easy to write and read •Let's touch TaskGroup. •Exception/cancellation triggers a review of the existing code. •A more comfortable asyncio life
  133. •Task Status Appendix: Task Status •PENDING •FINISHED •CANCELLED •You can

    check with task._state •The sample code also outputs the state. Please try to run it. Here is the code IUUQTHJUIVCDPNKSGLUBML
  134. Appendix: HTTPX with TaskGroup Here is the code IUUQTHJUIVCDPNKSGLUBML import

    asyncio from time import time import httpx URL = "https://pokeapi.co/api/v2/pokemon/" async def access_url_poke(client, num: int) -> str: r = await client.get(f"{URL}{num}") pokemon = r.json() # JSONύʔε return pokemon["name"] # ϙέ໊Λൈ͘ async def main_poke(): """httpxͰϙέϞϯ151ඖҾͬ͜ൈ͘""" start = time() async with httpx.AsyncClient() as client: try: async with asyncio.TaskGroup() as tg: tasks = [ tg.create_task(access_url_poke(client, number)) for number in range(1, 151) ] except* BaseException as e: print(f"{e=}") results = [task.result() for task in tasks] print(results) print("time: ", time() - start) asyncio.run(main_poke())
  135. Appendix: New features that are still available! Here is the

    code IUUQTHJUIVCDPNKSGLUBML import asyncio async def main(): try: async with asyncio.TaskGroup() as tg: task1 = tg.create_task(some_coro(...)) task2 = tg.create_task(another_coro(...)) except* BaseException as e: print(f"{e=}") with asyncio.Runner() as runner: runner.run(main()) 3.11 What's New What is asyncio.Runner! For interactive sessions? run multiple times.
  136. Appendix: it was a scheduled draft 🙏. •asyncio.timeout() •Sneakily Added

    Python 3.11 New Features •Born of a wait_for bugs? •Implement asyncio.timeout() context manager #90927 IUUQTHJUIVCDPNQZUIPODQZUIPOJTTVFT …
  137. Appendix: it was a scheduled draft 🙏. •asyncio evolution history

    •Asynchronous function asyncore, which has actually been around since Python 1.5 •Third-party asynchronous implementation •curio, trio , AnyIO, quattro, etc. … …
  138. thanks •Using Asyncio in Python - Oreilly & Associates Inc

    •Python 3.11 Preview: Task and Exception Groups - realpython.com •https://realpython.com/python311-exception-groups/ •Python 3.11ͷ৽ػೳ(ͦͷ4ʣPEP 654 ྫ֎άϧʔϓͱTaskGroup •https://www.python.jp/news/wnpython311/except-grp.html •The 2021 Python Language Summit: PEP 654 — Exception Groups and except* •https://pyfound.blogspot.com/2021/05/the-2021-python-language-summit-pep-654.html • How Exception Groups Will Improve Error Handling in AsyncIO - Łukasz Langa | Power IT Conference •https://www.youtube.com/watch?v=Lfe2zsGS0Js