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

プロダクションでのPython非同期ユースケース - Trio/Trio-Utilを中心に

Junya Fukuda
September 28, 2024
130

プロダクションでのPython非同期ユースケース - Trio/Trio-Utilを中心に

プロダクションでのPython非同期ユースケース - Trio/Trio-Utilを中心に (PyCon JP 2024)

Description
Pythonの非同期を利用されていますか?おそらく、市場での多くのユースケースはIOを多用するWebでの活用で、FastAPIやDjangoでasyncioを利用されている方も多いのではないでしょうか。今回のトークでは、asyncioと同時期にリリースされ「structured concurrency」という概念に強く影響を受けた Python非同期のサードパーティライブラリ「Trio」について、asyncioとの比較を交えお話しします。

asyncio以外のpython非同期って何?anyioってとは?なぜTrioを選択したのか、にお答えします。世界的に見ても稀有なTrioをプロダクションで利用している事例を中心に、実際のコードを交えTrioとasyncioの比較をしながら、ユースケースをご紹介します。

Session: https://2024.pycon.jp/ja/talk/HFE3MV
X: https://x.com/JunyaFff
linkedin: https://www.linkedin.com/in/junya-fukuda-4622a863/

Junya Fukuda

September 28, 2024
Tweet

Transcript

  1. •book •Python࣮ફϨγϐ •ΤΩεύʔτPythonϓϩάϥϛϯάվగ4൛ •Junya Fukudaʢ@JunyaFffʣ •Photos 📷 post 👍 •Software

    Engineer at Groove X, inc. •࿈ࡌ: Python Monthly Topics | gihyo.jp ͖ΐ͏ͷίʔυ 🔍GitHub jrfk talk
  2. Web? ✋ Database? ✋ API calls? ✋ Python の 非

    同期を使っていますか? ✋
  3. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  4. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  5. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  6. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  7. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  8. •Structured Concurrency ͬͯԿʁ •ߏ଄Խ͞Εͨฒߦੑ •Martin Sustrik ࢯ͕2016೥2݄ʹఏএͨ֓͠೦ Structured Concurrency |

    250bpm.com : https://250bpm.com/blog:71/ Why Trio? - Structured Concurrencyとキャンセル
  9. •՝୊͸ʮΩϟϯηϧʯ main() foo() bar() Why Trio? - Structured Concurrencyとキャンセル 1.

    ଙεϨου͸ͦͷ··ॲཧ͢Δέʔε 2. Ωϟϯηϧͯ͠ཉ͍͠έʔε
  10. •՝୊͸ʮΩϟϯηϧʯ main() foo() bar() Why Trio? - Structured Concurrencyとキャンセル 1.

    ଙεϨου͸ͦͷ··ॲཧ͢Δέʔε 2. Ωϟϯηϧͯ͠ཉ͍͠έʔε
  11. •Structured Concurrency Async Rust can be a pleasure to work

    with (without `Send + Sync + 'static`) Why Trio? - Structured Concurrencyとキャンセル
  12. async def child(): print("👶ࢠίϧʔνϯ ͜Μʹͪ͸") await grandchild() print("👶 ͞Α͏ͳΒ") async

    def grandchild(): print("👶👶ଙίϧʔνϯ ͜Μʹͪ͸") await trio.sleep(0.1) print("👶👶 ͞Α͏ͳΒ")
  13. async def main(): async with trio.open_nursery() as nursery: nursery.start_soon(child) nursery.start_soon(child)

    nursery.start_soon(child) nursery.start_soon(child) print("ดԂ") •Structured Concurrency 🎉
  14. async def main(): try: async with trio.open_nursery() as nursery: nursery.start_soon(coro_success)

    nursery.start_soon(coro_value_err) nursery.start_soon(coro_type_err) nursery.start_soon(triplet_nap) # ৸ͯΔ except* Exception as err: print(f"{err.exceptions=}") await trio.sleep(5) # 👶👶👶͕print͞Εͳ͍ print("ดԂ")
  15. async def main(): try: async with trio.open_nursery() as nursery: nursery.start_soon(coro_success)

    nursery.start_soon(coro_value_err) nursery.start_soon(coro_type_err) nursery.start_soon(triplet_nap) # ৸ͯΔ except* Exception as err: print(f"{err.exceptions=}") await trio.sleep(5) # 👶👶👶͕print͞Εͳ͍ print("ดԂ")
  16. async def main(): try: async with trio.open_nursery() as nursery: nursery.start_soon(coro_success)

    nursery.start_soon(coro_value_err) nursery.start_soon(coro_type_err) nursery.start_soon(triplet_nap) # ৸ͯΔ except* Exception as err: print(f"{err.exceptions=}") await trio.sleep(5) # 👶👶👶͕print͞Εͳ͍ print("ดԂ")
  17. async def main(): try: async with trio.open_nursery() as nursery: nursery.start_soon(coro_success)

    nursery.start_soon(coro_value_err) nursery.start_soon(coro_type_err) nursery.start_soon(triplet_nap) # ৸ͯΔ except* Exception as err: print(f"{err.exceptions=}") await trio.sleep(5) # 👶👶👶͕print͞Εͳ͍ print("ดԂ")
  18. async def child(): print("👶ࢠίϧʔνϯ ͜Μʹͪ͸") await grandchild() print("👶 ͞Α͏ͳΒ") async

    def grandchild(): print("👶👶ଙίϧʔνϯ ͜Μʹͪ͸") await trio.sleep(0.1) print("👶👶 ͞Α͏ͳΒ")
  19. async def main(): async with asyncio.TaskGroup() as g: g.create_task(child()) g.create_task(child())

    g.create_task(child()) g.create_task(child(), name="Α͘৸ͯΔ") print("ดԂ")
  20. async def main(): async with asyncio.TaskGroup() as g: g.create_task(child()) g.create_task(child())

    g.create_task(child()) g.create_task(child(), name="Α͘৸ͯΔ") print("ดԂ")
  21. async def main(): async with asyncio.TaskGroup() as g: g.create_task(child()) g.create_task(child())

    g.create_task(child()) g.create_task(child(), name="Α͘৸ͯΔ") print("ดԂ")
  22. async def main(): async with asyncio.TaskGroup() as g: g.create_task(child()) g.create_task(child())

    g.create_task(child()) g.create_task(child(), name="Α͘৸ͯΔ") print("ดԂ") •Structured Concurrency 🎉
  23. async def main(): async with asyncio.TaskGroup() as g: g.create_task(child()) g.create_task(child())

    g.create_task(child()) g.create_task(child(), name="Α͘৸ͯΔ") print("ดԂ") •Structured Concurrency 🎉 •Ωϟϯηϧ΋ΈͯΈ·͠ΐ͏
  24. async def main(): try: async with asyncio.TaskGroup() as g: task1

    = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_type_err()) task4 = g.create_task(triplet_nap(), name="Α͘৸ͯΔ") results = [task1.result(), task2.result(), task3.result()] print(f"{results=}") except* Exception as err: print(f"{err.exceptions=}") await asyncio.sleep(5) # 👶👶👶͕print͞Εͳ͍͜ͱΛ֬ೝ print("ดԂ")
  25. async def main(): try: async with asyncio.TaskGroup() as g: task1

    = g.create_task(coro_success()) task2 = g.create_task(coro_value_err()) task3 = g.create_task(coro_type_err()) task4 = g.create_task(triplet_nap(), name="Α͘৸ͯΔ") results = [task1.result(), task2.result(), task3.result()] print(f"{results=}") except* Exception as err: print(f"{err.exceptions=}") await asyncio.sleep(5) # 👶👶👶͕print͞Εͳ͍͜ͱΛ֬ೝ print("ดԂ")
  26. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup Nඵޙʹ λεΫऴྃ move_on_after Timeout ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  27. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  28. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  29. async def main(): with trio.move_on_after(1) as cancel_scope: await triplet_nap() if

    cancel_scope.cancelled_caught: print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ")
  30. async def main(): with trio.move_on_after(1) as cancel_scope: await triplet_nap() if

    cancel_scope.cancelled_caught: print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ") 時間 設定
  31. async def main(): with trio.move_on_after(1) as cancel_scope: await triplet_nap() if

    cancel_scope.cancelled_caught: print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ")
  32. async def main(): with trio.move_on_after(1) as cancel_scope: await triplet_nap() if

    cancel_scope.cancelled_caught: print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ")
  33. async def main(): with trio.move_on_after(1) as cancel_scope: await triplet_nap() if

    cancel_scope.cancelled_caught: print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ") •Structured Concurrency 🎉
  34. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  35. async def main(): try: async with asyncio.timeout(1): await triplet_nap() except

    TimeoutError as err: print(f"{err=}") print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ")
  36. async def main(): try: async with asyncio.timeout(1): await triplet_nap() except

    TimeoutError as err: print(f"{err=}") print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ") 時間 設定
  37. async def main(): try: async with asyncio.timeout(1): await triplet_nap() except

    TimeoutError as err: print(f"{err=}") print("ͪΌΜͱ͓͖ͯͶ") print("͓ΘΓ") 時間 設定 •Structured Concurrency 🎉
  38. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  39. Trio asyncio 交 比 較 trio asyncio ಛ௃ ·ͱΊ࣮ͯߦ Nursery

    Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  40. Trio asyncio 交 比 較 Python >= 3.11 trio asyncio

    ಛ௃ ·ͱΊ࣮ͯߦ Nursery Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  41. Trio asyncio 交 比 較 Python < 3.11 trio asyncio

    asyncioͷಛ௃ ·ͱΊ࣮ͯߦ Nursery gather ߏ଄Խ 🙅 Ωϟϯηϧ 🤔 ϑϨϯυϦ 🤔 Nඵޙʹ λεΫऴྃ move_on_after wait_for ߏ଄Խ 🙅 Ωϟϯηϧ 🤔 ϑϨϯυϦ 🤔 ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ 今 使 ! !
  42. Trio asyncio 交 比 較 Python >= 3.11 trio asyncio

    ಛ௃ ·ͱΊ࣮ͯߦ Nursery Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  43. Trio asyncio 交 比 較 Python >= 3.11 trio asyncio

    ಛ௃ ·ͱΊ࣮ͯߦ Nursery Taskgroup ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 Nඵޙʹ λεΫऴྃ move_on_after Timeout ߏ଄Խ 🙆 Ωϟϯηϧ 🙆 ϑϨϯυϦ 🙆 ஋ͷมԽΛ଴ͭ trio-util.AsyncValue ʁ
  44. Handling Multiple Inputs •AsyncValue Class 👀 •஋΍awaitable ΛҾ਺ʹ •val =

    AyncValue(2024) Έ͍ͨͳײ͡ •ͦΕΛ·ͱΊΔ compose_event ΋͋Γ·͢
  45. Handling Multiple Inputs •AsyncValue Class 👀 •஋΍awaitable ΛҾ਺ʹ •val.wait_value(஋ or

    ؔ਺) •val.wait_transition(ۭ or ஋ or ؔ਺) •஋ͷભҠɾมԽɾؔ਺ͷঢ়ଶΛ଴ͭ
  46. Handling Multiple Inputs •AsyncValue Class 👀 •஋΍awaitable ΛҾ਺ʹ •val.wait_value(஋ or

    ؔ਺) •val.wait_transition(ۭ or ஋ or ؔ਺) •Val.eventual_value(ۭ or ஋ or ؔ਺) •৚݅Λຬͨͨ͠ΒloopΛճ͢
  47. Handling Multiple Inputs •AsyncValue Class 👀 •஋΍awaitable ΛҾ਺ʹ •val.wait_value(஋ or

    ؔ਺) •val.wait_transition(ۭ or ஋ or ؔ਺) •eventual_value(ۭ or ஋ or ؔ਺) •࣮ࡍʹͲ͏ಈ͔͘ΈͯΈ·͠ΐ͏
  48. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  49. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  50. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  51. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  52. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  53. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  54. async def main(): async with trio.open_nursery() as nursery: ws =

    await connect_websocket( nursery=nursery, host="localhost", port=8888, resource="/", ) event1 = AsyncValue("👶") event2 = AsyncValue("🍼") nursery.start_soon(fetch, ws, event1, event2) with compose_values(event1=event1, event2=event2) as composite_eve async for v in composite_event.eventual_values(): print(f"{v=}") •Continuously watch for value changes 👀
  55. Handling Multiple Inputs •Trio-util does not support anyio yet •There

    is no API that performs the same function. •asyncio-util IUUQTHJUIVCDPNKSGLBTZODJPVUJM Here is today's code
  56. •Python async ͱ ϩϘοτͰͷϢʔεέʔε •Why Python Async •Why Trio •Trio

    ͱ Trio-util, asyncio ͷAPIൺֱ •anyio ͦͯ͠ subinterpreter/free threading ͱasync
  57. anyio •asyncio ͱ trio Λܨ͙Χέϋγ •2018 2018೥10݄ Alex Grönholm ࢯ

    •࡞੒ͨ͠ओͳཧ༝ •Pythonͷඇಉظϓϩάϥϛϯάʹ͓͚Δॊೈੑͱ ߴϨϕϧͷந৅ԽΛఏڙ •ΑΓ࢖͍΍͘͢ɺΑΓҰ؏ੑͷ͋Δ ඇಉظϓϩάϥϛϯάͷܦݧΛఏڙ͔ͨͬͨ͠ #385: Higher level Python asyncio with AnyIO Transcript
  58. anyio •asyncio ͱ trio Λܨ͙Χέϋγ #385: Higher level Python asyncio

    with AnyIO Transcript •Global backend ͱ͍͏ߟ͑ํ •ଟ͘ͷඇಉظϥΠϒϥϦͰ࠾༻, FastAPI, HTTPX…
  59. # αϒΠϯλϓϦλͰ࣮ߦ͢Δؔ਺ def run_sub_interpreter(): interp_id = interpreters.create() interpreters.run_string( interp_id, """

    import asyncio async def async_task(): for i in range(5): ... uv run 6_subinterpreters_asyncio.py
  60. reference •Simplifying Python's Async with Trio •https://talkpython.fm/episodes/show/167/simplifying-pythons-async-with-trio •Higher level Python

    asyncio with AnyIO •https://talkpython.fm/episodes/show/385/higher-level-python-asyncio-with-anyio •groove-x / trio-util •https://github.com/groove-x/trio-util •Structured Concurrency - Martin Sústrik •https://250bpm.com/blog:71/