测试是否等待协程答案

【问题标题】：Test if coroutine was awaited or not测试是否等待协程
【发布时间】：2019-10-25 19:32:37
【问题描述】：

我有一个连接到数据库的异步函数。现在我的用户这样做：

conn = await connect(uri, other_params)

我想继续支持这一点，但还想另外允许 connect() 用作上下文管理器：

async with connect(uri, other_params) as conn:
     pass

这两种情况的区别在于，第一种情况是等待connect，而第二种情况则不是。

是否有可能在connect 的正文中判断是否等待协程？

我目前在 repl.it 上的努力。

【问题讨论】：

标签： python python-3.x python-asyncio python-3.7

【解决方案1】：

这是通过您提供的测试的代码：

import asyncio
import pytest
from functools import wraps


def connection_context_manager(func):
  @wraps(func)
  def wrapper(*args, **kwargs):

    class Wrapper:
        def __init__(self):
          self._conn = None

        async def __aenter__(self):
            self._conn = await func(*args, **kwargs)
            return self._conn

        async def __aexit__(self, *_):
          await self._conn.close()

        def __await__(self):
            return func(*args, **kwargs).__await__()  # https://stackoverflow.com/a/33420721/1113207
    return Wrapper()

  return wrapper

注意三个魔术方法如何让我们同时使对象可等待和异步上下文管理器。

如果您有任何问题，请随时提出。

【讨论】：

很好的答案！我只是建议避免在每次调用装饰器时创建一个新类，因为类相对较重且构造缓慢。而且很容易避免它——只需在顶层定义Wrapper，将func 传递给它的构造函数，并在需要的地方使用self._func。
你和@user4815162342是我的英雄。这是好东西。我实际上不知道有__await__ 魔法。谢谢。
@user4815162342 的答案详细显示on my Github

【解决方案2】：

我认为您的示例中存在结构性问题。

所以首先你的例子需要await__call__：

@pytest.mark.asyncio
async def test_connect_with_context_manager():
    async with await connect("with context uri") as no_context:
        # Now it's open
        assert no_context.open

    # Now it's closed
    assert not no_context.open

但这里的问题是，await connect("with context uri") 的结果是一个Connection，它甚至没有__aexit__ 方法。

所以我认为你应该完全改变结构，将connect 方法添加到Connection 以实际建立连接，并在MagicConnection 中委托Connection 的每个方法。

【讨论】：