测试是否等待协程

Question

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

conn = await connect(uri, other_params)

我想继续支持它，但还想允许 connect() 用作上下文管理器：

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

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

在 connect 的正文中，是否可以判断协程是否已等待？

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

Answer 1

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

因此首先您的示例需要 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 中委托 [=] 的每个方法14=].

Answer 2

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

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__()  # 
    return Wrapper()

  return wrapper

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

有问题欢迎提问

测试是否等待协程

Test if coroutine was awaited or not

python

python-3.x

python-asyncio

python-3.7