标签: httpx

我有一个 FastAPI 应用程序，在多种不同的场合，需要调用外部 API。我使用 httpx.AsyncClient 进行这些调用。关键是我不完全理解应该如何使用它。

从httpx 的文档中我应该使用上下文管理器，

async def foo():
    """"
    I need to call foo quite often from different 
    parts of my application
    """
    async with httpx.AsyncClient() as aclient:
        # make some http requests, e.g.,
        await aclient.get("http://example.it")

但是，我明白，通过这种方式，每次我调用 时都会生成一个新客户端foo()，而这正是我们首先希望通过使用客户端来避免的情况。

我想另一种选择是在某处定义一些全局客户端，然后在需要时导入它，就像这样

aclient = httpx.AsyncClient()

async def bar():
    # make some http requests using the global aclient, e.g.,
    await aclient.get("http://example.it")

不过，第二个选项看起来有点可疑，因为没有人负责关闭会话等。

所以问题是：如何httpx.AsyncClient()在 FastAPI 应用程序中正确（重新）使用？

我们有一个 FastApi 应用程序并使用 httpx AsyncClient 进行测试。我们遇到一个问题，单元测试在本地运行良好，但在 CI 服务器上失败（Github Actions）。

经过进一步研究，我们通过设置找到了这个建议的解决方案。raise_server_exceptions=FalseFalse

client = TestClient(app, raise_server_exceptions=False)

但这适用于同步客户端。我们正在使用异步客户端。

@pytest.fixture
async def client(test_app):
    async with AsyncClient(app=test_app, base_url="http://testserver") as client:
        yield client

AsyncClient 不支持该raise_app_exceptions=False选项。

有人对此有经验吗？谢谢

我正在尝试优化我制作的一个简单的网络抓取工具。它从主页上的表中获取 URL 列表，然后转到每个“子”URL 并从这些页面获取信息。我能够成功地同步编写它并使用concurrent.futures.ThreadPoolExecutor(). 然而，我正在尝试优化它的使用asyncio，httpx因为它们对于发出数百个 http 请求来说似乎非常快。

我使用编写了以下脚本asyncio，但是httpx，我不断收到以下错误：

httpcore.RemoteProtocolError: Server disconnected without sending a response.

RuntimeError: The connection pool was closed while 4 HTTP requests/responses were still in-flight.

当我运行脚本时，我似乎不断失去连接。我什至尝试运行它的同步版本并得到相同的错误。我以为远程服务器阻止了我的请求，但是，我能够运行我的原始程序并毫无问题地从同一 IP 地址访问每个网址。

什么会导致此异常以及如何解决它？

import httpx
import asyncio

async def get_response(client, url):
    resp = await client.get(url, headers=random_user_agent()) # Gets a random user agent.
    html = resp.text
    return html


async def main():
    async with httpx.AsyncClient() as client:
        tasks = []

        # …

例子

这是我的代码：

from typing import  List
from fastapi import FastAPI, File, UploadFile
import asyncio
import concurrent.futures

app = FastAPI()
@app.post("/send_images")
async def update_item(
    files: List[UploadFile] = File(...),
):
    return {"res": len(files)}

我向该服务器发送请求（这些特定网址仅作为示例）：

import requests
import os 
import json
import numpy as np
import time
import io
import httpx
import asyncio
from datetime import datetime
import pandas as pd

from random import shuffle
import pandas as pd
import urllib
import io
from PIL import Image
from matplotlib import pyplot as plt …

我想使用httpx从协程内的多个同时发生的 HTTP 流请求中读取数据，并将数据返回给运行事件循环的非异步函数，而不仅仅是返回最终数据。

但是如果我让我的异步函数产生而不是返回，我会收到抱怨asyncio.as_completed()并loop.run_until_complete()期望一个协程或一个 Future，而不是一个异步生成器。

因此，我可以让它工作的唯一方法是收集每个协程内的所有流数据，一旦请求完成就返回所有数据。然后收集所有协程结果，最后将其返回给非异步调用函数。

这意味着我必须将所有内容都保存在内存中，并等到最慢的请求完成后才能获取所有数据，这与流式传输 http 请求的全部意义相悖。

有什么办法可以完成这样的事情吗？我目前的愚蠢实现是这样的：

def collect_data(urls):
    """Non-async function wishing it was a non-async generator"""

    async def stream(async_client, url, payload):
        data = []
        async with async_client.stream("GET", url=url) as ar:
            ar.raise_for_status()
            async for line in ar.aiter_lines():
                data.append(line)
                # would like to yield each line here
        return data

    async def execute_tasks(urls):
        all_data = []
        async with httpx.AsyncClient() as async_client:
            tasks = [stream(async_client, url) for url in urls]
            for coroutine …

考虑这个向 API 端点发出简单 GET 请求的函数：

import httpx 

def check_status_without_session(url : str) -> int:
    response = httpx.get(url)
    return response.status_code

每次check_status_without_session调用该函数时，运行该函数都会打开一个新的 TCP 连接。现在， HTTPX 文档的这一Client部分建议在向同一 URL 发出多个请求时使用API。下面的函数可以做到这一点：

import httpx

def check_status_with_session(url: str) -> int:
    with httpx.Client() as client:
        response = client.get(url)
        return response.status_code

根据文档，使用Client将确保：

...客户端实例使用 HTTP 连接池。这意味着当您向同一主机发出多个请求时，客户端将重用底层 TCP 连接，而不是为每个请求重新创建一个连接。

我的问题是，在第二种情况下，我将Client上下文管理器包装在一个函数中。如果我check_status_with_session使用相同的 URL 多次调用，那么每次调用函数时是否都会创建一个新的连接池？这意味着它实际上并没有重用连接。由于函数执行后函数堆栈被销毁，因此Client对象也应该被销毁，对吗？这样做有什么好处或者有更好的方法吗？

我需要为一个用于从 API 获取数据的函数编写测试用例。在那里我使用 httpx.AsyncClient() 作为上下文管理器。但我不明白如何为该功能编写测试用例。

async def make_dropbox_request(url, payload, dropbox_token):
async with httpx.AsyncClient(timeout=None, follow_redirects=True) as client:
    headers = {
        'Content-Type': 'application/json',
        'authorization': 'Bearer '+ dropbox_token
    }
    # make the api call
    response = await client.post(url, headers=headers, json=payload)
    
    if response.status_code not in [200]:
        print('Dropbox Status Code: ' + str(response.status_code))

    if response.status_code in [200, 202, 303]:
        return json.loads(response.text)

    elif response.status_code == 401:
        raise DropboxAuthenticationError()

    elif response.status_code == 429:
        sleep_time = int(response.headers['Retry-After'])
        if sleep_time < 1*60:
            await asyncio.sleep(sleep_time)
            raise DropboxMaxRateLimitError()
        raise DropboxMaxDailyRateLimitError()

    raise …

我的项目包括使用构建在 aws lambda 服务之上的 api。从技术上讲，构建 api 的负责人告诉我，由于服务是弹性的，因此没有固定的请求限制，但重要的是要考虑 api 可以支持的每秒请求数。

为了控制每秒请求的限制（同时），我正在开发的 python 脚本使用 asyncio 和 httpx 来同时使用 api，并利用httpx.Limits的 max_connections 参数，我试图找到最佳值，以便API 不会冻结。

我的问题是，我不知道我是否误解了 max_connections 参数的使用，因为当使用值 1000 进行测试时，我的理解告诉我，每秒我会同时向 api 发出 1000 个请求，但即便如此， api 在一定时间后冻结。

我希望能够控制每秒请求的限制，而无需使用第三方库。

我怎样才能做到呢？

这是我的 MWE

async def consume(client, endpoint: str = '/create', reg):
  data = {"param1": reg[1]}

  response = await client.post(url=endpoint, data=json.dumps(data))

  return response.json()

async def run(self, regs):
  # Empty list to consolidate all responses
  results = []

  # httpx limits configuration
  limits = httpx.Limits(max_keepalive_connections=None, max_connections=1000)
  timeout = …

我正在尝试访问具有机器人防护功能的网站。

使用以下脚本使用请求我可以访问该网站。

request = requests.get(url,headers={**HEADERS,'Cookie': cookies})

我得到了所需的 HTML。但是当我使用 aiohttp 时

async def get_data(session: aiohttp.ClientSession,url,cookies):
    async with session.get(url,timeout = 5,headers={**HEADERS,'Cookie': cookies}) as response:
        text = await response.text()
        print(text)

我收到机器人预防页面作为响应。

这是我用于所有请求的标头。

HEADERS = {
    'User-Agent': 'PostmanRuntime/7.29.0',
    'Host': 'www.dnb.com',
    'Connection': 'keep-alive',
    'Accept': '/',
    'Accept-Encoding': 'gzip, deflate, br'
} 

我比较了 requests.get 和 aiohttp 的请求标头，它们是相同的。

结果不同有什么原因吗？如果是这样为什么？

编辑：我检查了 httpx 模块，问题也出现在httpx.Client()那里httpx.AsyncClient()。

response = httpx.request('GET',url,headers={**HEADERS,'Cookie':cookies})

效果不太好。（不是异步的）

安装正确的软件包后出现上述错误

Python --版本

Python 3.6.9

安装命令

pip3 安装 httpx

点 3 列表

anyio (3.6.1)
async-generator (1.10)
Brotli (1.0.9)
certifi (2022.6.15)
charset-normalizer (2.1.0)
contextvars (2.4)
dataclasses (0.8)
dnspython (2.2.1)
email-validator (1.2.1)
h11 (0.12.0)
httpcore (0.14.7)
httpx (0.22.0)
idna (3.3)
immutables (0.18)
MarkupSafe (2.0.1)
pip (9.0.1)
pkg-resources (0.0.0)
pydantic (1.9.1)
python-dateutil (2.8.2)
rfc3986 (1.5.0)
setuptools (39.0.1)
six (1.16.0)
sniffio (1.2.0)
typing-extensions (4.1.1)
validator (0.7.1)

在虚拟环境交互式 shell 中，该包也可以正常工作

    (env) PEOPLE\saurabhkamble@lp7948:/var/www/vip_select_shaadi_api$ uvicorn main:app --reload
    INFO:     Will watch for changes in these directories: ['/var/www/vip_select_shaadi_api']
    INFO: …