Fluent pattern with async methods

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()
)

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())