Тествайте дали сърутина е била изчаквана или не

Имам асинхронна функция, която се свързва с база данни. В момента моите потребители правят:

conn = await connect(uri, other_params)

Искам да продължа да поддържам това, но искам допълнително да разреша connect() да се използва като контекстен мениджър:

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

Разликата между тези два сценария е, че в първия случай се чака connect, а във втория не е.

Възможно ли е да се каже, в рамките на тялото на connect, дали съпрограмата е била изчаквана или не?

Настоящите ми усилия в това на repl.it.


python-3.x python python-3.7 python-asyncio
person LondonRob    schedule 25.10.2019    source източник

Отговори (2)


arrow_upward
2
arrow_downward

Ето код, който преминава предоставените от вас тестове:

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

Обърнете внимание как три магически метода ни позволяват да направим обекта чакащ и асинхронен контекстен мениджър едновременно.

Чувствайте се свободни да задавате въпроси, ако имате такива.

person Mikhail Gerasimov    schedule 26.10.2019
comment
Страхотен отговор! Просто бих препоръчал да се избягва създаването на нов клас всеки път, когато се извиква декораторът, защото класовете са сравнително тежки и бавни за конструиране. И е доста лесно да го избегнете - просто дефинирайте Wrapper на най-високо ниво, предайте func на неговия конструктор и използвайте self._func където е необходимо. - person user4815162342; 26.10.2019
comment
Вие и @user4815162342 сте моите герои. Това е добро нещо. Всъщност не знаех, че има __await__ магия. Благодаря и на двамата. - person LondonRob; 26.10.2019
comment
Отговорът на @user4815162342 е показан подробно в моя Github - person LondonRob; 26.10.2019

arrow_upward
0
arrow_downward

Мисля, че във вашия пример има структурен проблем.

Така че първо вашият пример трябва да 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.

person Sraw    schedule 25.10.2019