Как мне делать асинхронные HTTP-запросы с использованием httpx (VS-запросы) в Python?

Question

Я новичок и async и у меня возникли проблемы с преобразованием функции, которая вызывает внешний API, использующийся, в асинхронную функцию с использованием httpx.

Это исходная функция:

      import requests


def get_api_data(url: str) -> Optional[List[str]]:
    """Get the data from an API url.
    Args:
        url (str):  The API url.
    Returns:
        list: The data.
    """
    try:
        resp = requests.get(url)
        return resp.json()
    except requests.exceptions.HTTPError as errh:
        print("Http Error:", errh)
    except requests.exceptions.ConnectionError as errc:
        print("Error Connecting:", errc)
    except requests.exceptions.Timeout as errt:
        print("Timeout Error:", errt)
    except requests.exceptions.RequestException as err:
        print("OOps: Something Else", err)

и вот что я пробовал:

      import httpx


async def get_api_data(url: str) -> Optional[List[Any]]:
    """Get the data from an API url.
    Args:
        url (str):  The API url.
    Returns:
        list: The data.
    """
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(url)
            return resp.json()
    except httpx.HTTPError as errh:
        print("Http Error:", errh)
    except httpx.ConnectError as errc:
        print("Error Connecting:", errc)
    except httpx.TimeoutException as errt:
        print("Timeout Error:", errt)
    except httpx.RequestError as err:
        print("OOps: Something Else", err)

Но я не уверен, что это правильный код и как его проверить ...

Пример теста перед изменением функции ( requestsверсия); Оно использует patch издеваться:

      import pytest
import requests

from unittest.mock import patch


def test_api_http_error():
    with patch('app.internal.world_clock.requests.get', side_effect=requests.exceptions.HTTPError):
        assert not get_api_data(TIMEZONES_BASE_URL)

Я потратил много часов, пытаясь понять это, поэтому, если у вас есть какие-либо предложения о том, как правильно выполнить это преобразование и как его протестировать, я был бы очень признателен.

Answer 1

Ваша функция выглядит нормально. Вы используете свою асинхронную функцию? У меня была эта проблема, когда я впервые попытался использовать ее для замены запросов в моем проекте. Вместо того, чтобы просто называть это так: results = async_funtion(parameter)Это должно было быть: results = asyncio.run(async_funtion(parameter))В первом примере вы просто создаете объект сопрограммы, второй фактически запускает его.