AnySync

一个轻量级库，允许以同步方式调用异步函数。

import asyncio
from anysync import anysync


@anysync
async def f():
    return 42


assert f().run() == 42


async def main():
    assert await f() == 42


asyncio.run(main())

只需pip install anysync即可！

使用方法

协程

anysync的主要用途是允许以同步方式调用异步函数。您只需将anysync.coroutine装饰器添加到您的异步函数中即可

import asyncio
import anysync


@anysync.coroutine
async def f():
    return 42


assert f().run() == 42

生成器

您还可以使用anysync与异步生成器一起使用

import asyncio
import anysync


@anysync.generator
async def gen():
    yield 1
    yield 2
    yield 3


assert list(gen()) == [1, 2, 3]

注意，在这种情况下，您不需要调用run()。生成器将自动检测其使用方式并相应地运行协程。

上下文管理器

您甚至可以在异步上下文管理器中使用AnySync。

import anysync


@anysync.contextmanager
async def cm():
    yield 42


with cm() as x:
    assert x == 42

您还可以从AnySyncContextManager类继承

from anysync import AnySyncContextManager


class CM(AnySyncContextManager):
    async def __aenter__(self):
        return 42

    async def __aexit__(self, exc_type, exc, tb):
        pass


with CM() as x:
    assert x == 42

比较

asyncio.run

与asyncio.run不同，即使事件循环已经在运行，AnySync对象也可以被run()。

例如，以下代码将引发一个RuntimeError

import asyncio


async def f():
    return 42


async def test_async():
    assert asyncio.run(f()) == 42


asyncio.run(test_async())

然而，使用AnySync，以下代码将按预期工作

import asyncio
from anysync import anysync


@anysync
async def f():
    return 42


async def test_async():
    assert f().run() == 42


asyncio.run(test_async())

unsync

AnySync与unsync类似，允许在需要时以同步方式调用异步函数。主要区别在于AnySync可以通过anyio与类型检查器和其他异步库（如trio）以及异步生成器和上下文管理器一起使用。

自动检测

处理混合同步和异步代码挑战的另一种方法是自动推断函数是否应该在异步上下文中运行。这种方法被像Prefect的sync_compatible装饰器这样的库所采用。主要缺点是函数的行为会根据上下文动态变化，可能导致意外的行为。

例如，下面的代码在同步上下文中调用work()时按预期运行

from prefect.utilities.asyncutils import sync_compatible


@sync_compatible
async def request():
    ...
    return "hello"


def work():
    response = request()
    ...
    return response.upper()


def test_sync():
    assert work() == "HELLO"


test_sync()

然而，如果我们现在从异步上下文中调用work()，行为会发生变化

import asyncio


async def test_async():
    assert work() == "HELLO"  # AttributeError: 'coroutine' object has no attribute 'upper'


asyncio.run(test_async())

因为work()现在是从异步上下文中调用的，所以request()自动返回一个协程对象，导致work()失败。

其他考虑因素

工作原理

AnySync通过检测运行的事件循环来工作。如果已经存在一个，那么AnySync将使用一个单独的线程来运行协程。尽可能的情况下，AnySync会尝试重用单个全局后台线程，该线程仅在需要时创建。但是，如果在异步上下文中程序反复尝试同步运行协程，AnySync会每次都创建一个新线程。

例如，您可以在两种不同的情况下计算使用的线程数。第一种情况是反复重用相同的全局线程。

from threading import current_thread

import anysync

threads = set()


@anysync.coroutine
async def f():
    threads.add(current_thread())  # runs in the main thread
    return g().run()


@anysync.coroutine
async def g():
    threads.add(current_thread())  # runs in anysync's global background thread
    return 42


f().run()
f().run()

main_thread = current_thread()
assert len(threads - {main_thread}) == 1

在第二种情况下，除了AnySync的全局后台线程外，还会创建两个线程，因为g()在全局后台线程中运行，而h()每次都在新线程中运行。

from threading import current_thread

import anysync

threads = set()


@anysync.coroutine
async def f():
    threads.add(current_thread())  # runs in the main thread
    return g().run()


@anysync.coroutine
async def g():
    threads.add(current_thread())  # runs in anysync's global background thread
    return h().run()


@anysync.coroutine
async def h():
    threads.add(current_thread())  # runs in a new thread each time
    return 42


f().run()
f().run()

main_thread = current_thread()
assert len(threads - {main_thread}) == 3

与contextvars交互

AnySync包装的协程或上下文管理器不会从异步上下文中传播到同步上下文的contextvars更改。这是因为contextvars在线程或事件循环之间不共享，并且AnySync必须创建这些才能以同步方式运行协程。因此，以下内容不受支持

from contextvars import ContextVar
from anysync import anysync


var = ContextVar("var", default=0)


@anysync
async def f():
    var.set(42)


f().run()
assert var.get() == 42  # AssertionError: 0 != 42