Python3：当需要向我们公开经典+异步接口时，在项目中重用代码的最佳方法

Question

该项目是python3库（包），实现自定义流二进制协议。考虑一对流处理类：

MyEncodingWriter(dst_stream)-将Python对象转换为原始字节流 MyDecodingReader(src_stream)-将原始编码字节转换为消息或Python对象

我需要同时支持阻塞同步流和asyncio StreamReader + StreamWriter（可能是从它们派生子类化）

我知道我不能直接使用coroutines作为常规同步函数。在

我想模仿异步读写器API（read(), readexactly(),...） 在我的类MyEncodingWriter()和MyDecodingReader()中，唯一的区别是异步版本中额外的async/await关键字。在

对我来说很奇怪，我会有两个副本或回购协议，代码非常相似，我不能重用。在

现在我只有这个想法。。。 …试着在元类的帮助下从async（通过剥离async/wait关键字）动态生成同步代码？在

有更好的方法吗？在

对于这种双重实现的任何建议或漂亮的样本/回购。在

Answer 1

使阻塞版本成为异步版本的子类，并包装调用。未测试，可能有语法错误，但大致如下：

class MyBlockingEncodingReader(MyEncodingReader):
    def readexactly(self, n):
        return asyncio.get_event_loop().run_until_complete(super().readexactly(n))