具有异步方法的流畅模式

Question

此类具有异步和同步方法（即

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

如果所有方法都是同步的，我就会这样做：

# Fluent pattern

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

我知道我可以做这样的事情：

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

但我正在寻找这样的东西：

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

Answer 1

您能做的最接近的事情是：

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

如果它是一个普通函数，则调用它并像平常一样链接它，如果它是一个异步函数，则调用它并等待整个表达式并将其括在括号内。异步函数链将是：

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

完整的工作代码：

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

Answer 2

是的，你完全可以，仅当约束如上所示时并且使用包装器。

在 python 3.12 上进行了测试，但无论如何都应该可以工作。

import asyncio
import inspect


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

    def __getattr__(self, item):
        # try if it's target object's
        try:
            attribute = getattr(self.target_obj, item)
        except AttributeError:
            # it's our method then
            return getattr(super(), item)

        try:
            assert inspect.ismethod(attribute) or inspect.iscoroutinefunction(attribute)
        except AssertionError as err:
            # wrong attribute
            raise TypeError(f"{type(self.target_obj).__name__}.{item} is not a method.") from err

        self.chained_calls.append(attribute)

        # return lambda that returns this Chainable
        return lambda: self

    async def _await_calls(self):
        for method in self.chained_calls:
            if inspect.iscoroutinefunction(method):
                await method()
            else:
                method()

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


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

    def __getattr__(self, item):
        print(item)

    def is_human(self) -> "Character":
        print("Checking is_human!")
        if self.human:
            return self
        raise Exception(f'{self.name} is not human')

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

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


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

    await chain_callable.is_human().has_job().is_knight()
    print("That character was a formidable knight who has job and is human.")


asyncio.run(demo())

Checking is_human!
Checking has_job!
Checking is_knight!
That character was a formidable knight who has job and is human.