在实现同时用于同步和异步应用程序的类时,我发现自己为这两种用例维护了几乎相同的代码。
举个例子,考虑一下:
from time import sleep
import asyncio
class UselessExample:
def __init__(self, delay):
self.delay = delay
async def a_ticker(self, to):
for i in range(to):
yield i
await asyncio.sleep(self.delay)
def ticker(self, to):
for i in range(to):
yield i
sleep(self.delay)
def func(ue):
for value in ue.ticker(5):
print(value)
async def a_func(ue):
async for value in ue.a_ticker(5):
print(value)
def main():
ue = UselessExample(1)
func(ue)
loop = asyncio.get_event_loop()
loop.run_until_complete(a_func(ue))
if __name__ == '__main__':
main()
在这个例子中,还不错,UselessExample
的 ticker
方法很容易串联维护,但是你可以想象异常处理和更复杂的功能可以快速发展一个方法并使其成为一个更大的问题,即使这两种方法实际上可以保持相同(仅用异步对应物替换某些元素)。
假设没有实质性差异值得完全实现,那么维护这样的类并避免不必要的重复的最佳(也是最 Pythonic)方法是什么?
最佳答案
要使基于 asyncio 协程的代码库可从传统的同步代码库中使用,没有一条万能的道路。您必须根据代码路径做出选择。
从一系列工具中挑选:
使用asyncio.run()
的同步版本
在协程周围提供同步包装器,它会阻塞直到协程完成。
即使是异步生成器函数,例如ticker()
,也可以在循环中以这种方式处理:
class UselessExample:
def __init__(self, delay):
self.delay = delay
async def a_ticker(self, to):
for i in range(to):
yield i
await asyncio.sleep(self.delay)
def ticker(self, to):
agen = self.a_ticker(to)
try:
while True:
yield asyncio.run(agen.__anext__())
except StopAsyncIteration:
return
可以使用辅助函数生成这些同步包装器:
from functools import wraps
def sync_agen_method(agen_method):
@wraps(agen_method)
def wrapper(self, *args, **kwargs):
agen = agen_method(self, *args, **kwargs)
try:
while True:
yield asyncio.run(agen.__anext__())
except StopAsyncIteration:
return
if wrapper.__name__[:2] == 'a_':
wrapper.__name__ = wrapper.__name__[2:]
return wrapper
然后只需在类定义中使用 ticker = sync_agen_method(a_ticker)
。
直接协程方法(不是生成器协程)可以用:
def sync_method(async_method):
@wraps(async_method)
def wrapper(self, *args, **kwargs):
return async.run(async_method(self, *args, **kwargs))
if wrapper.__name__[:2] == 'a_':
wrapper.__name__ = wrapper.__name__[2:]
return wrapper
分解出通用组件
将同步部分重构为生成器、上下文管理器、实用函数等。
对于您的具体示例,将 for
循环拉出到一个单独的生成器中可以最大程度地减少重复代码以实现两个版本的休眠方式:
class UselessExample:
def __init__(self, delay):
self.delay = delay
def _ticker_gen(self, to):
yield from range(to)
async def a_ticker(self, to):
for i in self._ticker_gen(to):
yield i
await asyncio.sleep(self.delay)
def ticker(self, to):
for i in self._ticker_gen(to):
yield i
sleep(self.delay)
虽然这在这里没有太大区别,但它可以在其他情况下工作。
抽象语法树转换
使用 AST 重写和映射将协程转换为同步代码。如果您不注意如何识别诸如 asyncio.sleep()
与 time.sleep()
之类的实用函数,这可能会非常脆弱:
import inspect
import ast
import copy
import textwrap
import time
asynciomap = {
# asyncio function to (additional globals, replacement source) tuples
"sleep": ({"time": time}, "time.sleep")
}
class AsyncToSync(ast.NodeTransformer):
def __init__(self):
self.globals = {}
def visit_AsyncFunctionDef(self, node):
return ast.copy_location(
ast.FunctionDef(
node.name,
self.visit(node.args),
[self.visit(stmt) for stmt in node.body],
[self.visit(stmt) for stmt in node.decorator_list],
node.returns and ast.visit(node.returns),
),
node,
)
def visit_Await(self, node):
return self.visit(node.value)
def visit_Attribute(self, node):
if (
isinstance(node.value, ast.Name)
and isinstance(node.value.ctx, ast.Load)
and node.value.id == "asyncio"
and node.attr in asynciomap
):
g, replacement = asynciomap[node.attr]
self.globals.update(g)
return ast.copy_location(
ast.parse(replacement, mode="eval").body,
node
)
return node
def transform_sync(f):
filename = inspect.getfile(f)
lines, lineno = inspect.getsourcelines(f)
ast_tree = ast.parse(textwrap.dedent(''.join(lines)), filename)
ast.increment_lineno(ast_tree, lineno - 1)
transformer = AsyncToSync()
transformer.visit(ast_tree)
tranformed_globals = {**f.__globals__, **transformer.globals}
exec(compile(ast_tree, filename, 'exec'), tranformed_globals)
return tranformed_globals[f.__name__]
虽然上面的内容可能远远不足以满足所有需求,并且转换 AST 树可能令人生畏,但上面的内容可以让您只维护异步版本并将该版本直接映射到同步版本:
>>> import example
>>> del example.UselessExample.ticker
>>> example.main()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/.../example.py", line 32, in main
func(ue)
File "/.../example.py", line 21, in func
for value in ue.ticker(5):
AttributeError: 'UselessExample' object has no attribute 'ticker'
>>> example.UselessExample.ticker = transform_sync(example.UselessExample.a_ticker)
>>> example.main()
0
1
2
3
4
0
1
2
3
4
关于python - 同步和异步实现的代码重复,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55152952/