跳转到主要内容

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

Build Status Coverage Package version

需求

  • 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 应用的测试客户端使用。

您可能需要安装 pytestpytest-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

替代方案

  • http3 提供了同步和异步 HTTP 客户端,具有与 requests 兼容的 API。
  • aiohttp 提供了用于发送异步 HTTP 请求的替代客户端。

项目详情


下载文件

下载您平台上的文件。如果您不确定选择哪个,请了解有关 安装包 的更多信息。

源分发

requests-async-0.6.2.tar.gz (12.9 kB 查看散列)

上传时间

由...

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF 赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误日志 StatusPage StatusPage 状态页