Method chaining with asyncio coroutines

Question

I want to implement method chaining, but not for usual functions - for asyncio coroutines.

import asyncio class Browser: @asyncio.coroutine def go(self): # some actions return self @asyncio.coroutine def click(self): # some actions return self

"Intuitive" way to call chain wouldn't work, because single method returns coroutine (generator), not self:

@asyncio.coroutine def main(): br = yield from Browser().go().click() # this will fail loop = asyncio.get_event_loop() loop.run_until_complete(main())

Correct way to call chain is:

br = yield from (yield from Browser().go()).click()

But it looks ugly and becomes unreadable when chain grows.

Is there any way to do this better? Any ideas are welcome.

I'm not sure I understand what you're trying to do, but if you just want to iterate methods on the object you can do that by setting them into a dict, or with getattr. — user3467349
– user3467349, Commented Feb 15, 2015 at 4:34

Mikhail Gerasimov · Accepted Answer · 2015-02-17 20:05:17Z

I created solution, that do a job close to the needed. Idea is to use wrapper for Browser() which uses __getattr__ and __call__ to collect action (like getting attribute or call) and return self to catch next one action. After all actions collected, we "catch" yiled from wrapper using __iter__ and process all collected actions.

import asyncio def chain(obj): """ Enables coroutines chain for obj. Usage: text = yield from chain(obj).go().click().attr Note: Returns not coroutine, but object that can be yield from. """ class Chain: _obj = obj _queue = [] # Collect getattr of call to queue: def __getattr__(self, name): Chain._queue.append({'type': 'getattr', 'name': name}) return self def __call__(self, *args, **kwargs): Chain._queue.append({'type': 'call', 'params': [args, kwargs]}) return self # On iter process queue: def __iter__(self): res = Chain._obj while Chain._queue: action = Chain._queue.pop(0) if action['type'] == 'getattr': res = getattr(res, action['name']) elif action['type'] == 'call': args, kwargs = action['params'] res = res(*args, **kwargs) if asyncio.iscoroutine(res): res = yield from res return res return Chain()

Usage:

class Browser: @asyncio.coroutine def go(self): print('go') return self @asyncio.coroutine def click(self): print('click') return self def text(self): print('text') return 5 @asyncio.coroutine def main(): text = yield from chain(Browser()).go().click().go().text() print(text) loop = asyncio.get_event_loop() loop.run_until_complete(main())

Output:

go click go text 5

Note, that chain() doesn't return real coroutine, but object that can be used like coroutine on yield from. We should wrap result of chain() to get normal coroutine, which can be passed to any asyncio function that requires coroutine:

@asyncio.coroutine def chain_to_coro(chain): return (yield from chain) @asyncio.coroutine def main(): ch = chain(Browser()).go().click().go().text() coro = chain_to_coro(ch) results = yield from asyncio.gather(*[coro], return_exceptions=True) print(results)

Output:

go click go text [5]

Note: @asyncio.coroutine is the old method to implement an asyncio coroutine and will soon be deprecated. Use the built-in async keyword as in async def in the future. Here is the asyncio documentation.

dano · Accepted Answer · 2015-02-15 21:34:36Z

It's still not particularly pretty, but you could implement a chain function that scales a little bit better:

import asyncio @asyncio.coroutine def chain(obj, *funcs): for f, *args in funcs: meth = getattr(obj, f) # Look up the method on the object obj = yield from meth(*args) return obj class Browser: @asyncio.coroutine def go(self, x, y): return self @asyncio.coroutine def click(self): return self @asyncio.coroutine def main(): #br = yield from (yield from Browser().go(3, 4)).click() br = yield from chain(Browser(), ("go", 3, 4), ("click",)) loop = asyncio.get_event_loop() loop.run_until_complete(main())

The idea is to pass tuples in a (method_name, arg1, arg2, argX) format to the chain function, rather than actually chaining the method calls themselves. You can just pass the method names directly if you don't need to support passing arguments to any of the methods in the chain.

drukmano · Accepted Answer · 2023-09-26 12:35:50Z

I had the same issue and wrote quent to handle these cases:

from quent import ChainAttr br = await ChainAttr(Browser()).go().click().run()

It also supports method cascading without the class having to implement it itself (i.e. without return self):

from quent import CascadeAttr br = await CascadeAttr(Browser()).go().click().run()

Collectives™ on Stack Overflow

Method chaining with asyncio coroutines

3 Answers 3

1 Comment

Comments

Comments

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

Comments

Comments

Linked

Related