如何在Python中使用httpx (VS请求)进行异步HTTP请求?

Question

我是httpx和async的新手，我在将使用requests调用外部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)

但我不确定这是一个有效的代码和如何测试它…

变更功能前的测试示例(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

您的函数看起来很好。你正在运行async函数吗?当我第一次尝试使用它来替换项目中的请求时，我遇到了这个问题。而不是直接叫它:results = async_funtion(parameter)它必须是:results = asyncio.run(async_funtion(parameter))在第一个例子中，您只是创建了一个协程对象，第二个例子实际运行了它。