Skip to content

Instantly share code, notes, and snippets.

@flyte
Last active March 14, 2021 17:15
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save flyte/1d4321108b4d2511e84026a19c6c0fff to your computer and use it in GitHub Desktop.
Save flyte/1d4321108b4d2511e84026a19c6c0fff to your computer and use it in GitHub Desktop.
Add a results property to a nursery, containing the return values of the functions which have been started (and finished).
from functools import wraps
import trio
def set_result(results, key, func):
@wraps(func)
async def wrapper(*args, **kwargs):
results[key] = await func(*args, **kwargs)
return wrapper
class ResultNurseryManager:
def __init__(self):
self.nursery_manager = trio.open_nursery()
async def __aenter__(self):
nursery = await self.nursery_manager.__aenter__()
nursery.results = {}
def start_soon_result(name, async_fn, *args):
nursery.start_soon(
set_result(nursery.results, name, async_fn), *args, name=name
)
nursery.start_soon_result = start_soon_result
return nursery
async def __aexit__(self, etype, exc, tb):
await self.nursery_manager.__aexit__(etype, exc, tb)
async def get_status(msg):
return f"{msg} status"
async def get_statuses():
async with ResultNurseryManager() as nursery:
nursery.start_soon_result("a", get_status, "aaa")
nursery.start_soon_result("b", get_status, "bbb")
print(f"Status A: {nursery.results['a']}")
print(f"Status B: {nursery.results['b']}")
trio.run(get_statuses)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment