如何在Python中为含异步方法的类实现Fluent pattern？

现有如下Python类，包含同步方法isHuman与异步方法hasJob、isKnight：

from typing import Self
import asyncio

class Character:
    def __init__(self, name: str, human: bool):
        self.name = name
        self.human = 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

当所有方法为同步时，可通过Fluent模式实现链式调用：

# 同步链式调用示例
john = (
    Character("John", human=True)
    .isHuman()
    .hasJob()
    .isKnight()
)

目前异步方法只能分步调用：

# 现有异步调用方式
john = Character("John", human=True).isHuman()
await john.hasJob()
await john.isKnight()

期望实现类似同步的链式调用写法，通过一次await得到最终实例：

# 目标异步链式调用写法
john = await (
    Character("John", human=True)
    .isHuman()
    .hasJob()
    .isKnight()
)

核心思路是让异步方法返回的对象同时支持链式调用和被await执行，以下提供两种可行实现方式：

方案一：通用包装类封装调用链 #

创建一个AsyncFluentWrapper类，负责拦截方法调用、处理同步/异步逻辑，并实现__await__方法让自身可被await：

from typing import Any, Coroutine
import asyncio

class AsyncFluentWrapper:
    def __init__(self, instance: Any):
        self._instance = instance

    def __getattr__(self, name: str) -> Any:
        attr = getattr(self._instance, name)
        if asyncio.iscoroutinefunction(attr):
            # 异步方法：执行后返回新的包装器
            async def wrapped(*args, **kwargs):
                result = await attr(*args, **kwargs)
                return AsyncFluentWrapper(result)
            return wrapped
        else:
            # 同步方法：执行后更新实例并返回自身维持链式
            def wrapped(*args, **kwargs):
                self._instance = attr(*args, **kwargs)
                return self
            return wrapped

    def __await__(self):
        # 让包装器可被await，最终返回原始实例
        async def _await():
            return self._instance
        return _await().__await__()

# 修改Character的isHuman方法，返回包装器
class Character:
    def __init__(self, name: str, human: bool):
        self.name = name
        self.human = human

    def isHuman(self) -> AsyncFluentWrapper:
        if self.human:
            return AsyncFluentWrapper(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():
    john = await (
        Character("John", human=True)
        .isHuman()
        .hasJob()
        .isKnight()
    )
    print(john.name)  # 输出: John

asyncio.run(main())

方案二：直接修改原类异步方法 #

如果不想额外引入包装类，可直接改造异步方法，让其返回一个兼具链式调用和可await特性的内部类实例：

from typing import Self, Coroutine
import asyncio

class Character:
    def __init__(self, name: str, human: bool):
        self.name = name
        self.human = 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

    # 对外暴露的链式方法
    def hasJob(self):
        class Wrapper:
            def __init__(self, coro: Coroutine):
                self.coro = coro
                self.result = None

            async def _get_result(self):
                if self.result is None:
                    self.result = await self.coro
                return self.result

            def __getattr__(self, name: str):
                attr = getattr(self.result, name) if self.result else getattr(Character, name)
                if asyncio.iscoroutinefunction(attr):
                    async def wrapped(*args, **kwargs):
                        await self._get_result()
                        coro = attr(self.result, *args, **kwargs)
                        return Wrapper(coro)
                    return wrapped
                else:
                    def wrapped(*args, **kwargs):
                        self._get_result()
                        self.result = attr(self.result, *args, **kwargs)
                        return self
                    return wrapped

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

        return Wrapper(self._hasJob())

    # isKnight方法做同样改造
    async def _isKnight(self) -> Self:
        await asyncio.sleep(1)
        return self

    def isKnight(self):
        class Wrapper:
            def __init__(self, coro: Coroutine):
                self.coro = coro
                self.result = None

            async def _get_result(self):
                if self.result is None:
                    self.result = await self.coro
                return self.result

            def __getattr__(self, name: str):
                attr = getattr(self.result, name) if self.result else getattr(Character, name)
                if asyncio.iscoroutinefunction(attr):
                    async def wrapped(*args, **kwargs):
                        await self._get_result()
                        coro = attr(self.result, *args, **kwargs)
                        return Wrapper(coro)
                    return wrapped
                else:
                    def wrapped(*args, **kwargs):
                        self._get_result()
                        self.result = attr(self.result, *args, **kwargs)
                        return self
                    return wrapped

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

        return Wrapper(self._isKnight())

调用方式完全符合预期：

async def main():
    john = await (
        Character("John", human=True)
        .isHuman()
        .hasJob()
        .isKnight()
    )
    print(john.name)

asyncio.run(main())

方案对比 #

• 方案一的通用包装类更灵活，可复用给其他需要异步链式调用的类；
• 方案二更直接，无需额外依赖，但需要逐个改造异步方法。

内容的提问来源于stack exchange，提问作者ZC13