兼容同步和异步客户端的Python函数

Question

我正在开发一个库，该库应该支持同步和异步用户，同时最大限度地减少库内的代码重复。理想的情况是实现库异步（因为该库执行远程 API 调用）并添加对与包装器等同步的支持。将以下函数视为库的一部分。

# Library code

async def internal_function():
  """Internal part of my library doing HTTP call"""
  pass

我的想法是提供一个包装器来检测库的用户是否使用异步。

# Library code

def api_call():
   """Public api of my library"""
    if asyncio.get_event_loop().is_running():
        # Async caller
        return internal_function()  # This returns a coroutine, so the caller needs to await it
    
    # Sync caller, so we need to run the coroutine in a blocking way
    return asyncio.run(internal_function())

起初，这似乎是解决方案。有了这个我就可以支持

• 从异步函数调用该函数的用户，
• 从笔记本调用该函数的用户（也是异步的），以及
• 从普通 python 脚本调用该函数的用户（这里它会退回到阻塞

  asyncio.run

  ）。

但是，在某些情况下，从事件循环内调用该函数，但直接调用者是同步函数。

# Code from users of the library

async def entrypoint():
    """This entrypoint is called in an event loop, e.g. within fastlib"""
    return legacy_sync_code()  # Call to sync function

def legacy_sync_code():
    """Legacy code that does not support async"""
    # Call to library code from sync function,
    # expect value not coroutine, could not use await
    api_response = api_call()
    return api_response.json()  # Needs to be value, not coroutine

最后一行，

json()

的调用失败。

api_call()

错误地推断调用者可以等待响应，并返回协程。

为了支持此类用户，我需要

• 确定直接调用函数是否是同步的并且需要一个值而不是协程，
• 有办法以阻塞方式等待

  internal_function()

  的结果。使用

  asyncio.run

  只能在纯 Python 脚本中工作，如果代码是在堆栈跟踪中较高的事件循环中调用的，则会失败。

如果我的库提供两个函数，例如

api_call_async()

和

api_call_sync()

，则可以缓解第一点。

我希望我的观点足够清楚。我不认为这从根本上是不可能的，但是，如果 Python 的设计不允许我以完全透明的方式支持同步和异步用户，我可以接受。

Answer 1

不要试图太灵活或解决尚未发生的问题。这将使您的生活变得更加困难，并且您的代码的开发和维护可能会更加复杂，结果并不像您想象的那么灵活。不要过多地照顾其他开发人员，让他们承担一些责任，并在想要同步或异步时做出自己的选择，这并没有什么错。另请注意，对您有利的内容（在实现细节等方面，例如检测同步/异步）可能对其他人来说是一个破坏因素，因为您可能正在做一些没有被要求的事情。很容易智胜自己而没有任何好处。我个人会采用单独的

api_call_sync()

和

api_call_async()

方法，并确保我的文档清晰并讨论您现在想到的所有边缘情况，以便我的用户全面了解。