Fluent pattern with async methods

Question

This class has async and sync methods (i.e. isHuman):

class Character:

  def isHuman(self) -> Self:
    if self.human:
      return self
    raise Exception(f'{self.name} is not human')

  async def hasJob(self) -> Self:
    await asyncio.sleep(1)
    return self

  async def isKnight(self) -> Self:
    await asyncio.sleep(1)
    return self

If all methods were sync, I'd have done:

# Fluent pattern

jhon = (
  Character(...)
  .isHuman()
  .hasJob()
  .isKnight()
)

I know I could do something like:

jhon = Character(...).isHuman()
await jhon.hasJob()
await jhon.isKnight()

But I'm looking for something like this:

jhon = await (
  Character(...)
  .isHuman()
  .hasJob()
  .isKnight()
)

Answer 1

Edit

Now support parameters.

---

Yup you totally can. Almost nothing is impossible in python - every single int, float, etc are objects too after all. await call ain't untouchable.

This method involves following decorator line on target class

@add_chain_call
class Character:
    ...


chara = Character(True)


# --- using obj ---
return_self = await chara.chain.is_human(1, a=1).has_job(2, 3, b=2).is_knight(4, c=3, d=4)
assert return_self is chara
print("That character was a formidable knight who has job and is human.")

return_str = await chara.chain.is_knight().non_self_returning(name="hina", first_name="sorasaki")
print("And that knight said,", return_str)

... or if that class isn't yours, then you can do this. Decorator's just function after all.

from some_module import Character

chara = add_chain_call(Character)(*args, **kwargs)


# --- using obj ---
return_self = await chara.chain.is_human(1, a=1).has_job(2, 3, b=2).is_knight(4, c=3, d=4)
assert return_self is chara
print("That character was a formidable knight who has job and is human.")

return_str = await chara.chain.is_knight().non_self_returning(name="hina", first_name="sorasaki")
print("And that knight said,", return_str)

I intentionally added .chain accessor to indicate explicitly that we're chaining method. This also comes in benefit that it does not affect any other non-chain calls.

However, ultimate downside would be:

• additional .chain access.
• only works with await keyword, even if all calls are non-async. (Since this only override __await__)

If you are only using for your own classes where you can inherit freely, and be released from .chain accessor and become Truely Free™ - Check @jsbueno's answer as he probably spent amazing amount of effort and knowledge to implement it.

---

Tested on python 3.12 but should work up to 3.9 regardless.

Chaining wrapper class & Chaining decorator:

import inspect


class ChainCall:
    def __init__(self, target):
        self.target_obj = target
        self.chained_calls = []
        self.params = []

    def __getattr__(self, item):
        # try if it's target object's
        try:
            attribute = getattr(self.target_obj, item)

        except AttributeError as err:
            try:
                # it's our method then.
                return getattr(super(), item)

            except AttributeError:
                # we catch & reraise because it otherwise returns 'super' as object name
                # which is kinda confusing and misleading.
                raise AttributeError(
                    f"'{type(self.target_obj).__name__}' object has no attribute '{item}'",
                    name=item, obj=self.target_obj
                ) from err

        # check if it's non method attributes (ignoring callable here)
        if not (inspect.ismethod(attribute) or inspect.iscoroutinefunction(attribute)):
            raise TypeError(f"'{type(self.target_obj).__name__}.{item}' is not a method.")

        self.chained_calls.append(attribute)

        # return lambda that returns this Chainable, with param
        def wrapper(*args, **kwargs) -> "ChainCall":
            self.params.append((args, kwargs))
            return self

        return wrapper

    async def _await_calls(self):
        """Starts calling pending calls.
        Also checks if returning object is identical, excluding last call."""
        last_return = None

        for idx, (method, (arg, kwarg)) in enumerate(zip(self.chained_calls, self.params), start=1):
            if inspect.iscoroutinefunction(method):
                last_return = await method(*arg, **kwarg)
            else:
                last_return = method(*arg, **kwarg)

            # raise if identity is wrong, just in case.
            # ignored on the last call.
            if (last_return is not self.target_obj) and idx != len(self.chained_calls):
                raise TypeError(
                    f"'{type(self.target_obj).__name__}.{method.__name__}' did not return self."
                )

        return last_return

    def __await__(self):
        return self._await_calls().__await__()


def add_chain_call(tgt_class):
    """Decorate class with chain property."""

    def chain(self) -> ChainCall:
        """Start chaining following calls."""
        return ChainCall(self)

    tgt_class.chain = property(chain)

    return tgt_class

Full code:

import asyncio
import inspect


class ChainCall:
    def __init__(self, target):
        self.target_obj = target
        self.chained_calls = []
        self.params = []

    def __getattr__(self, item):
        # try if it's target object's
        try:
            attribute = getattr(self.target_obj, item)

        except AttributeError as err:
            try:
                # it's our method then.
                return getattr(super(), item)

            except AttributeError:
                # we catch & reraise because it otherwise returns 'super' as object name
                # which is kinda confusing and misleading.
                raise AttributeError(
                    f"'{type(self.target_obj).__name__}' object has no attribute '{item}'",
                    name=item, obj=self.target_obj
                ) from err

        # check if it's non method attributes (ignoring callable here)
        if not (inspect.ismethod(attribute) or inspect.iscoroutinefunction(attribute)):
            raise TypeError(f"'{type(self.target_obj).__name__}.{item}' is not a method.")

        self.chained_calls.append(attribute)

        # return lambda that returns this Chainable, with param
        def wrapper(*args, **kwargs) -> "ChainCall":
            self.params.append((args, kwargs))
            return self

        return wrapper

    async def _await_calls(self):
        """Starts calling pending calls.
        Also checks if returning object is identical, excluding last call."""
        last_return = None

        for idx, (method, (arg, kwarg)) in enumerate(zip(self.chained_calls, self.params), start=1):
            if inspect.iscoroutinefunction(method):
                last_return = await method(*arg, **kwarg)
            else:
                last_return = method(*arg, **kwarg)

            # raise if identity is wrong, just in case.
            # ignored on the last call.
            if (last_return is not self.target_obj) and idx != len(self.chained_calls):
                raise TypeError(
                    f"'{type(self.target_obj).__name__}.{method.__name__}' did not return self."
                )

        return last_return

    def __await__(self):
        return self._await_calls().__await__()


def add_chain_call(tgt_class):
    """Decorate class with chain property."""

    def chain(self) -> ChainCall:
        """Start chaining following calls."""
        return ChainCall(self)

    tgt_class.chain = property(chain)

    return tgt_class


@add_chain_call
class Character:
    def __init__(self, human):
        self.human = human

    def is_human(self, *arg, **kwarg) -> "Character":
        print("Checking is_human! Param:", arg, kwarg)
        if self.human:
            return self
        raise Exception("Behold! I am no human.")

    async def has_job(self, *arg, **kwarg) -> "Character":
        print("Checking has_job! Param:", arg, kwarg)
        await asyncio.sleep(1)
        return self

    async def is_knight(self, *arg, **kwarg) -> "Character":
        print("Checking is_knight! Param:", arg, kwarg)
        await asyncio.sleep(1)
        return self

    async def non_self_returning(self, *arg, **kwarg) -> str:
        print("In non_self_returning! Param:", arg, kwarg)
        await asyncio.sleep(1)
        return "Tho facing of a ambiguity, shall refuse the temptation to guess."


async def demo():
    chara = Character(True)

    return_self = await chara.chain.is_human(1, a=1).has_job(2, 3, b=2).is_knight(4, c=3, d=4)
    assert return_self is chara
    print("That character was a formidable knight who has job and is human.")

    return_str = await chara.chain.is_knight().non_self_returning(name="hina", first_name="sorasaki")
    print("And that knight said,", return_str)


asyncio.run(demo())

Checking is_human! Param: (1,) {'a': 1}
Checking has_job! Param: (2, 3) {'b': 2}
Checking is_knight! Param: (4,) {'c': 3, 'd': 4}
That character was a formidable knight who has job and is human.
Checking is_knight! Param: () {}
In non_self_returning! Param: () {'name': 'hina', 'first_name': 'sorasaki'}
And that knight said, Tho facing of a ambiguity, shall refuse the temptation to guess.

---

Below are some tailored error output sample for some common expected errors. Outputs are generated before arg support, but should work nontheless.

On calling method that doesn't return self (cant undo the call tho):

@add_chain_call
class Character:
    ...
    async def has_job(self) -> "Character":
        ...
        return Character(False)
...

Checking is_human!
Checking has_job!
Traceback (most recent call last):
...
TypeError: 'Character.has_job' did not return self.

On calling method that isn't method:

chara = Character(True)
await chara.chain.human().has_job().is_knight()
                # ^^^^^
...


Traceback (most recent call last):
...
TypeError: 'Character.human' is not a method.

On calling method that doesn't exists:

await chara.chain.is_vulcan().has_job().is_knight()
                # ^^^^^^^^^
...

Traceback (most recent call last):
...
AttributeError: 'Character' object has no attribute 'is_vulcan'. Did you mean: 'is_human'?

---

P.S. IDE won't like it but can't help, such is what makes python dynamic. Either add editor-specific suppressions to silence it or leave it as a reminder to self that we're doing hush-hush things!

(And chaining async funcs would show warning anyway!)

Answer 2

Tricky - but can be done using some advanced properties in Python. Usually, the big problem, even with all Python capabilities, implementing an automated way to curry methods like you are doing is to know when the chain stops (i.e. when the result of the last method will be actually used, and no longer used with a . to call the next method).

But if the whole expression is to be used with await that is resolved, we use the __await__ method to finish up and execute the chain.

I needed some back-and forth to get it all working, and your Character class will have to inherit from asyncio.Future (so, beware of clashing names for methods and attributes) - other than that, the code bellow did work.

(May need some clean-up - sorry, there where some approaches, I can clean-up later)

from inspect import isawaitable
from types import MethodType
from inspect import iscoroutinefunction
from typing import Self
from collections import deque

import asyncio


class LazyCallProxy:
    def __init__(self, parent, callable):
        self.__callable = callable
        self.__parent = parent

    def __call__(self, *args, **kw):
        result = self.__callable(*args, **kw)
        self.__result = result
        return self.__parent

    def __getattribute__(self, attr):
        if attr == "_result":
            return self.__result
        if attr.startswith(f"_{__class__.__name__}"):
            return super().__getattribute__(attr)
        return getattr(self.__parent, attr)


class AwaitableCurryMixin(asyncio.Future):

    def __init__(self, *args, **kw):
        self._to_be_awaited = deque()
        super().__init__(*args, **kw)

    def __await__(self):
        tasks = []
        for proxy in self._to_be_awaited:
            result = proxy._result
            if isawaitable(result):
                tasks.append(asyncio.create_task(result))
        if not tasks:
            return super().__await__()
        def mark_done(task):
            self.set_result(task.result())
        tasks[-1].add_done_callback(mark_done)
        return super().__await__()


    def __getattribute__(self, attr):
        obj = super().__getattribute__(attr)
        _to_be_awaited = super().__getattribute__("_to_be_awaited")
        if not iscoroutinefunction(obj) and not (_to_be_awaited := super().__getattribute__("_to_be_awaited")):
            # if not in the midle of a chaincall, and this is not
            # an async method, just return the attribute:
            return obj
        if isinstance(obj, MethodType) and obj.__annotations__["return"] in (Self, type(self)):
            _to_be_awaited.append(proxy:=LazyCallProxy(self, obj))
            return proxy
        return obj


class Character(AwaitableCurryMixin):


    def __init__(self, name):
        super().__init__()
        self.name = name
        self.human = True

    def isHuman(self) -> Self:
        if self.human:
            return self
        raise Exception(f'{self.name} is not human')

    async def hasJob(self) -> Self:
        await asyncio.sleep(1)
        return self

    async def isKnight(self) -> Self:
        await asyncio.sleep(1)
        return self
    def __repr__(self):
        return self.name


async def main():
    john = await (
    Character("john")
    .isHuman()
    .hasJob()
    .isKnight()
    )
    print(john)

if __name__ == "__main__":
    asyncio.run(main())

Answer 3

The closest thing you can do is:

result = await (await Character().isHuman().hasJob()).isKnight()

If it's a normal function call it and chain it like normal, if it's an async function call it and await the whole expression and wrap it inside a parenthesis. The chain of async functions would be:

(await (await (await async_fn1()).async_fn2()).async_fn3())

Full working code:

import asyncio
from typing import Self


class Character:
    def __init__(self, name: str, is_human: bool) -> None:
        self.name = name
        self.human = is_human

    def isHuman(self) -> Self:
        if self.human:
            return self
        raise Exception(f"{self.name} is not human")

    async def hasJob(self) -> Self:
        await asyncio.sleep(1)
        return self

    async def isKnight(self) -> Self:
        await asyncio.sleep(1)
        return self


async def main():
    result = await (await Character("foo", is_human=True).isHuman().hasJob()).isKnight()
    print(result)  # <__main__.Character object at 0x104af5f10>


asyncio.run(main())