如何在python中选择异步或同步方法变体？

Question

假设我有一个用于执行 I/O 操作的类：

class CommunicationStack:
    def __init__(self, socket):
        self.socket = socket

    def sync_get_data(self):
        ...

    def sync_send_data(self):
        ...

    async def async_get_data(self):
        ...

    async def async_send_data(self):
        ...

如您所见，它具有用于相同操作的同步和异步变体，但手动编写 async_get_data 或 sync_get_data 会有点不方便。我正在寻找一种具有相同界面的智能方法，例如

def get_data(self):
    ...  # call sync variant or return an awaitable, depending on caller type

def send_data(self):
    ...  # call sync variant or return an awaitable, depending on caller type

所以它可以方便地使用：

stack = CommunicationStack(...)

def thread_procedure():
    data = stack.get_data()  # get_data returns data

async def task_procedure():
    data = await stack.get_data()  # get_data returns an awaitable

我相信它可以通过检查或black magic 一些棘手的方式完成：

def is_caller_coroutine():
    return sys._getframe(2).f_code.co_flags & 0x380

检查调用者是协程还是函数，但是弄乱python的胆量似乎是一个糟糕的设计。

---

问题是：选择合适的变体的好方法是什么？或者有没有更好的方法来设计所有东西，比如使用adapters 或开发两个独立的AsyncCommunicationStack 和SyncCommunicationStack 类？

Answer 1

如果您想以与常规函数相同的方式调用异步函数，您可能有兴趣使用gevent。

asyncio 希望您将函数显式标记为async，并在发生异步事件的任何地方显式使用await。换句话说，asyncio 希望你有不同的接口用于同步和异步代码。

这是intentional decision 用于解决并发问题，当代码的异步性质被隐藏时（如在 gevent 中），这将更难实现。

所以是的 - 如果您想支持这两个世界，两个不同的独立 AsyncCommunicationStack 和 CommunicationStack 类是一种方法。虽然一旦你有了异步版本，你就可以使用它来编写关键代码并使其同步，只需使用asyncio.run() 运行它。唯一的问题是在那之后您将无法返回异步世界。