从单个异步实现中导出阻塞和异步库版本
项目描述
Python 3 对异步编程有一些惊人的支持，但它可能使得开发库变得更加困难。您是否厌倦了实现同步和异步方法来做基本上相同的事情？这可能为您提供一个简单的解决方案。
安装
pip install synchronicity
背景：为什么需要这样的东西
假设您有一个异步函数
async def f(x):
await asyncio.sleep(1.0)
return x**2
假设（出于某种原因）您想向用户提供同步API。例如，可能您想使您的代码在基本脚本中运行变得容易，或者用户正在构建一个主要是CPU密集型的项目，因此他们不想处理asyncio。
创建同步等效方法的一个“简单”方式是实现一组同步函数，它们所做的仅仅是调用asyncio.run。但这种方法对于更复杂的代码来说并不是一个好的解决方案。
- 对于每个方法/函数都这样做是一种相当繁琐的体力劳动。
- asyncio.run不能与生成器一起使用。
- 在许多情况下，需要在调用之间保持事件循环的运行。
最后一种情况尤其具有挑战性。例如，假设你正在实现一个需要持久连接的数据库客户端，并且你想使用asyncio来构建它。
class DBConnection:
def __init__(self, url):
self._url = url
async def connect(self):
self._connection = await connect_to_database(self._url)
async def query(self, q):
return await self._connection.run_query(q)
如何向这段代码公开同步接口？问题在于，在asyncio.run中将connect和query包装起来不起作用，因为你需要跨调用保留事件循环。很明显，我们需要稍微高级一些的东西。
如何使用这个库
这个库提供了一个简单的Synchronizer类，它在单独的线程上创建一个事件循环，并包装函数/生成器/类，以便在相应的线程上执行同步执行。当你调用任何东西时，它将检测你是否在一个同步或异步上下文中运行，并相应地执行。
- 在同步情况下，它将简单地阻塞，直到结果可用（注意，你还可以让它返回一个future，见后文）
- 在异步情况下，它的工作方式就像通常调用异步代码一样
from synchronicity import Synchronizer
synchronizer = Synchronizer()
@synchronizer.create_blocking
async def f(x):
await asyncio.sleep(1.0)
return x**2
# Running f in a synchronous context blocks until the result is available
ret = f(42) # Blocks
print('f(42) =', ret)
async def g():
# Running f in an asynchronous context works the normal way
ret = await f(42)
print('f(42) =', ret)
更高级的示例
生成器
这个装饰器也适用于生成器
@synchronizer.create_blocking
async def f(n):
for i in range(n):
await asyncio.sleep(1.0)
yield i
# Note that the following runs in a synchronous context
# Each number will take 1s to print
for ret in f(10):
print(ret)
同步整个类
它还会通过包装类上的每个方法来作用于类
@synchronizer.create_blocking
class DBConnection:
def __init__(self, url):
self._url = url
async def connect(self):
self._connection = await connect_to_database(self._url)
async def query(self, q):
return await self._connection.run_query(q)
# Now we can call it synchronously, if we want to
db_conn = DBConnection('tcp://localhost:1234')
db_conn.connect()
data = db_conn.query('select * from foo')
返回future
你也可以通过将_future=True添加到任何调用中来让函数返回一个Future对象。如果你想要从阻塞上下文中调度许多调用，但又想大致并行地解决它们，这可能很有用
from synchronicity import Synchronizer
synchronizer = Synchronizer()
@synchronizer.create_blocking
async def f(x):
await asyncio.sleep(1.0)
return x**2
futures = [f(i, _future=True) for i in range(10)] # This returns immediately
rets = [fut.result() for fut in futures] # This should take ~1s to run, resolving all futures in parallel
print('first ten squares:', rets)
与其他异步代码一起使用
如果你有多个事件循环，或者你有一些CPU密集型的部分，或者有一些你想要在单独的线程上运行以保障安全的临界代码，这个库在纯异步设置中也可能很有用。所有同步函数/生成器的调用都是线程安全的，这是它的设计。这使得它在简单事情上成为loop.run_in_executor的一个有用的替代品。然而，请注意，每个同步器只运行一个线程。
上下文管理器
你可以像同步任何其他类一样同步上下文管理器类，并且特殊方法将被正确处理。
还有一个函数装饰器@synchronizer.asynccontextmanager，它的行为就像https://docs.pythonlang.cn/3/library/contextlib.html#contextlib.asynccontextmanager一样，但在同步和异步上下文中都可以工作。
需要注意的问题
- 它适用于是上下文管理器的类，但不适用于返回上下文管理器的函数
- 它在包装类时创建了一个新的类（具有相同的名称），这可能导致在未同步使用相同类的情况下出现类型问题
- 不清楚它与类型注解的交互方式
- 如果一个类是“同步的”，它将包装类上的所有方法，但这通常意味着你不能访问属性并在其上运行异步代码：你可能会得到“附加到不同的循环”之类的错误
- 请注意，所有同步代码都将在一个不同的线程和不同的事件循环上运行，因此调用代码可能会有些微额外的开销
待办事项
- 支持相反的情况，即您有一个阻塞函数/生成器/类/对象，并且想要异步调用它（对于普通函数来说，使用 loop.run_in_executor 来做这件事相对简单，但Python没有内置对生成器的支持，能够将整个类转换为异步调用会更好。
- 更多文档
- 允许有选择性地注释方法以返回future
- 可能允许在运行时同步对象，而不仅仅是类
这个库处于非常边缘的肢体截肢状态
这是我从个人项目中提取的代码，并且还没有经过实战测试。你可以使用pytest运行一个小的测试套件。
发布流程
应该自动化这个...
- 从主分支创建一个新的分支
release-X.Y.Z - 在pyproject.toml中将版本号升级到
X.Y.Z - 提交这个更改并创建一个PR
- 一旦绿色，就合并PR
- 检出主分支
git tag -a vX.Y.Z -m "* release bullets"- git push --tags
TWINE_USERNAME=__token__ TWINE_PASSWORD="$PYPI_TOKEN_SYNCHRONICITY" make publish
下载文件
下载适合您平台的文件。如果您不确定选择哪一个，请了解更多关于 安装包 的信息。
