为`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 |
