为`requests`提供async-await支持。
项目描述
此作品已被http3项目取代: https://www.encode.io/http3/
我们现在建议使用http3.AsyncClient()
,以具有与requests兼容的API来支持async/await。
注意: 由于它支持await
,请使用ipython
在控制台尝试此操作。
>>> import http3
>>> client = http3.AsyncClient()
>>> r = await client.get('https://www.example.org/')
>>> r.status_code
200
>>> r.text
'<!doctype html>\n<html>\n<head>\n<title>Example Domain</title>...'
requests-async
将async
/await
语法引入Python的出色requests
库。
需求
- Python 3.6+
安装
$ pip install requests-async
使用
只需使用标准requests API,但在发出请求时使用await
。
注意: 由于它支持await
,请使用ipython
在控制台尝试此操作。
import requests_async as requests
response = await requests.get('https://example.org')
print(response.status_code)
print(response.text)
或者使用显式的会话,并使用异步上下文管理器。
import requests_async as requests
async with requests.Session() as session:
response = await session.get('https://example.org')
print(response.status_code)
print(response.text)
requests_async
包是requests
的子类,因此您将获得所有预期的标准行为和API。
流式响应和请求
iter_content()
和iter_lines()
方法都是异步迭代器。
response = await requests.get('https://example.org', stream=True)
async for chunk in response.iter_content():
...
方法签名与标准requests
API相同
iter_content(chunk_size=1, decode_unicode=False)
iter_lines(chunk_size=512, decode_unicode=False, delimiter=None)
如果设置decode_unicode
并且响应包含编码,则方法将产生文本。否则,方法将产生字节。
您还可以流式传输请求正文。为此,您应使用产生字节的异步生成器。
async def stream_body():
...
response = await requests.post('https://example.org', data=stream_body())
模拟请求
在某些情况下,例如在测试Web应用程序时,您可能不想发出实际的出站网络请求,而是希望模拟端点。
您可以使用 ASGISession
来实现,这允许您连接到任何 ASGI 应用程序,而不是实际进行网络请求。
import requests_async
# Create a mock service, with Starlette, Responder, Quart, FastAPI, Bocadillo,
# or any other ASGI web framework.
mock_app = ...
if TESTING:
# Issue requests to the mocked application.
requests = requests_async.ASGISession(mock_app)
else:
# Make live network requests.
requests = requests_async.Session()
测试客户端
您还可以将 ASGISession
作为任何 ASGI 应用的测试客户端使用。
您可能需要安装 pytest
和 pytest-asyncio
或类似工具,以便编写 async
测试用例。
from requests_async import ASGISession
from myproject import app
import pytest
@pytest.mark.asyncio
async def test_homepage():
client = ASGISession(app)
response = await client.get("/")
assert response.status_code == 200
替代方案
项目详情
关闭
requests-async-0.6.2.tar.gz 的散列
算法 | 散列摘要 | |
---|---|---|
SHA256 | 8f128b66dec2974aa263e24410914b1829d3e1506497364c2edfa10112f55e0b |
|
MD5 | b03f7a535f906eaef28439b8d744d726 |
|
BLAKE2b-256 | 4ae7afbe47fa17068c80a5e80d75e1e960a98a5cfefe3590108ba76d9560cefc |