如何编写一个 Python 程序的异步/同步变体？

Question

先开发一个简单的实现是正常的。因此，例如，我们可能从一个非并发程序开始，然后添加并发。我希望能够顺畅地来回切换。

例如单线程（伪代码）：

   results=[]
   for url in urls:
      # This then calls other functions that call yet others
      # in a call hierarchy, down to a requests.request() call.
      get_result_and_store_in_database(url) 

异步（伪代码）：

# The following calls other functions that call yet others
# in a call hierarchy, down to an asyncio ClientSession().get() call.
# It runs HTTP requests and store the results in a database.
# The multiple URLs are processed concurrently.
asyncio.run(get_results_in_parallel_and_store_in_db(urls))

使用 Python async/await，通常你用asyncio.run() 包装运行（与你在普通程序中使用的循环相比）；然后在调用层次结构的底部，使用类似aiohttp.ClientSession().get(url) 的IO 操作（与普通的requests.request() 相比。）

但是，在异步版本中，这两者之间的调用层次结构中的所有函数都必须写为async/await。因此，我需要编写两个基本相同的调用层次结构的副本，主要区别在于它们是否具有 async/await 关键字。

那是很多代码重复。

如何制作可切换的非并发/异步程序？

Answer 1

这确实是一个很大的话题，但不是一个笼统的话题。我个人有一个实现同步版本和异步版本的私有 WebDAV 项目。

首先，我的 WebDAV 客户端接受一个名为 client 的参数，它可以是 requests.Session 或 aiohttp.ClientSession 来执行同步请求或异步请求。

其次，我有一个基类来实现所有常见的逻辑，例如：

def _perform_dav_request(self, method, auth_tuple=None, client=None, **kwargs):
    auth_tuple = self._get_auth_tuple(auth_tuple)
    client = self._get_client(client)
    data = kwargs.get("data")
    headers = None
    url = None

    path = kwargs.get("path")
    if path:
        root_url = urljoin(self._base_url, self._dav_url)
        url = root_url + path

    from_path = kwargs.get("from_path")
    to_path = kwargs.get("to_path")
    if from_path and to_path:
        root_url = urljoin(self._base_url, self._dav_url)
        url = root_url + from_path
        destination = root_url + quote(to_path)

        headers = {
            "Destination": destination
        }

    return client.request(method, url, data=data, headers=headers, auth=auth_tuple)

事实上requests.Session 和aiohttp.ClientSession 都支持几乎相同的API，所以在这里我可以使用一个模棱两可的调用client.request(...)。

第三，我要导出不同的API：

# In async client
async def ls(self, path, auth_tuple=None, client=None):
    response = await self._perform_dav_request("PROPFIND", auth_tuple, client, path=path)

    if response.status == 207:
        return parse_ls(await response.read())
    raise WebDavHTTPError(response.status, await response.read())

# In sync client
def ls(self, path, auth_tuple=None, client=None):
    response = self._perform_dav_request("PROPFIND", auth_tuple, client, path=path)

    if response.status_code == 207:
        return parse_ls(response.content)
    raise WebDavHTTPError(response.status_code, response.content)

所以最后我的用户可以像dav = DAV(...) 或dav = AsyncDAV(...) 一样使用它。

这就是我处理两个不同版本的方式。我认为这个想法是您可以通过函数调用传递这些协程，并且只在最高级别评估它们。所以你只需要在最后一层写不同的代码，但在所有其他层都有相同的逻辑。