在 Uvicorn/FastAPI 内部发出下游 HTTP 请求的正确方法是什么?

2024-12-03 08:44:00
admin
原创
220
摘要:问题描述:我有一个 API 端点(FastAPI / Uvicorn)。除其他事项外,它还会向另一个 API 发出信息请求。当我使用多个并发请求加载我的 API 时,我开始收到以下错误:h11._util.LocalProtocolError: can't handle event type Connecti...

问题描述:

我有一个 API 端点(FastAPI / Uvicorn)。除其他事项外,它还会向另一个 API 发出信息请求。当我使用多个并发请求加载我的 API 时,我开始收到以下错误:

h11._util.LocalProtocolError: can't handle event type ConnectionClosed when role=SERVER and state=SEND_RESPONSE

在正常环境中,我会利用request.session,但我知道它不是完全线程安全的。

那么,在 FastAPI 等框架中使用请求的正确方法是什么,其中多个线程会requests同时使用该库?


解决方案 1:

除了使用requests,您还可以使用httpx,它也提供了asyncAPI (在执行测试时httpx也在 FastAPI 的文档中建议async使用,并且 FastAPI/Starlette 最近将 HTTP 客户端TestClient从替换requestshttpx)。

下面的示例基于httpx文档中给出的示例,演示了如何使用库发出异步 HTTP(s) 请求,随后将响应流回客户端。httpx.AsyncClient()您可以用 代替requests.Session(),这在向同一主机发出多个请求时很有用,因为底层 TCP 连接将被重用,而不是为每个请求都重新创建一个连接 - 因此,可以显着提高性能。 此外,它允许您在请求之间重用headers和其他设置(例如proxiestimeout)以及保留。每次需要时,cookies您都会生成一个并重用它。 您可以使用在完成后显式关闭客户端(您可以在事件处理程序中执行此操作)。 在此答案中也可以找到示例和更多详细信息。Client`await client.aclose()`shutdown

例子

from fastapi import FastAPI
from starlette.background import BackgroundTask
from fastapi.responses import StreamingResponse
import httpx


app = FastAPI()


@app.on_event("startup")
async def startup_event():
    app.state.client = httpx.AsyncClient()


@app.on_event('shutdown')
async def shutdown_event():
    await app.state.client.aclose()


@app.get('/')
async def home():
    client = app.state.client
    req = client.build_request('GET', 'https://www.example.com/')
    r = await client.send(req, stream=True)
    return StreamingResponse(r.aiter_raw(), background=BackgroundTask(r.aclose))

示例(已更新)

由于startupshutdown现已弃用(将来可能会完全删除),因此您可以使用lifespan处理程序来初始化httpx客户端,并在关闭时关闭客户端实例,类似于此答案中演示的内容。 Starlette 在其文档页面中特别提供了一个使用lifespan处理程序和httpx客户端的示例。 如Starlette 的文档中所述:

有的lifespan概念state,它是一个字典,可以用来在生命周期和请求之间共享对象。

state请求上接收到的是生命周期处理程序上接收到的状态的浅表副本。

因此,可以使用 在端点内访问添加到生命周期处理程序中的状态的对象request.state。下面的示例使用流式响应与外部服务器通信,并将响应发送回客户端。有关(即、、等)响应流式方法的更多详细信息,请参阅此处。async`httpxaiter_bytes()aiter_text()`aiter_lines()

如果您想将 用作media_typeStreamingResponse则可以使用原始响应中的 ,如下所示:media_type=r.headers['content-type']。但是,正如本答案中所述,您需要确保media_type未设置为text/plain;否则,内容将不会按预期在浏览器中传输,除非您禁用MIME 嗅探(请查看链接的答案以了解更多详细信息和解决方案)。

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
import httpx


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialise the Client on startup and add it to the state
    async with httpx.AsyncClient() as client:
        yield {'client': client}
        # The Client closes on shutdown 


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def home(request: Request):
    client = request.state.client
    req = client.build_request('GET', 'https://www.example.com')
    r = await client.send(req, stream=True)
    return StreamingResponse(r.aiter_raw(), background=BackgroundTask(r.aclose)) 

如果由于某种原因,您需要在服务器端逐块读取内容,然后再响应客户端,那么您可以按如下方式执行此操作:

@app.get('/')
async def home(request: Request):
    client = request.state.client
    req = client.build_request('GET', 'https://www.example.com')
    r = await client.send(req, stream=True)
    
    async def gen():
        async for chunk in r.aiter_raw():
            yield chunk
        await r.aclose()
        
    return StreamingResponse(gen())

如果您不想使用流式响应,而是首先读取响应(这会将响应数据存储到服务器的 RAM 中;因此,您应该确保有足够的空间来容纳数据),则可以使用以下方法。请注意,使用httpx适用r.json()于响应数据为 JSON 格式的情况;否则,您可以直接返回PlainTextResponse或自定义Response,如下所示。

from fastapi import Response
from fastapi.responses import PlainTextResponse

@app.get('/')
async def home(request: Request):
    client = request.state.client
    req = client.build_request('GET', 'https://www.example.com')
    r = await client.send(req)
    content_type = r.headers.get('content-type')
    
    if content_type == 'application/json':
        return r.json()
    elif content_type == 'text/plain':
        return PlainTextResponse(content=r.text)
    else:
        return Response(content=r.content) 

使用asyncAPIhttpx意味着您必须使用 定义您的端点async def;否则,您必须使用标准同步 API(对于defvs,async def请参阅此答案),如此github 讨论中所述:

是的。HTTPX 旨在实现线程安全,是的,跨所有线程的单个客户端实例在连接池方面会比使用每个线程一个实例做得更好。

limits您还可以使用关键字参数控制连接池大小Client(请参阅池限制配置)。例如:

limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
client = httpx.Client(limits=limits)
相关推荐
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化,即信息技术应用创新国产化,是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新,实现信息技术应用的自主可控,减少对外部技术的依赖,并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧,以及某些国家对中国在科技领域的打压,信创国产化显...
工程项目管理   1579  
  为什么项目管理通常仍然耗时且低效?您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议?这确实是耗费时间和精力。借助软件工具的帮助,您可以一目了然地全面了解您的项目。如今,国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件?项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   1355  
  信创产品在政府采购中的占比分析随着信息技术的飞速发展以及国家对信息安全重视程度的不断提高,信创产业应运而生并迅速崛起。信创,即信息技术应用创新,旨在实现信息技术领域的自主可控,减少对国外技术的依赖,保障国家信息安全。政府采购作为推动信创产业发展的重要力量,其对信创产品的采购占比情况备受关注。这不仅关系到信创产业的发展前...
信创和国产化的区别   8  
  信创,即信息技术应用创新产业,旨在实现信息技术领域的自主可控,摆脱对国外技术的依赖。近年来,国货国用信创发展势头迅猛,在诸多领域取得了显著成果。这一发展趋势对科技创新产生了深远的推动作用,不仅提升了我国在信息技术领域的自主创新能力,还为经济社会的数字化转型提供了坚实支撑。信创推动核心技术突破信创产业的发展促使企业和科研...
信创工作   9  
  信创技术,即信息技术应用创新产业,旨在实现信息技术领域的自主可控与安全可靠。近年来,信创技术发展迅猛,对中小企业产生了深远的影响,带来了诸多不可忽视的价值。在数字化转型的浪潮中,中小企业面临着激烈的市场竞争和复杂多变的环境,信创技术的出现为它们提供了新的发展机遇和支撑。信创技术对中小企业的影响技术架构变革信创技术促使中...
信创国产化   8  
热门文章
项目管理软件有哪些?
云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护,随时随地协同办公

内置subversion和git源码管理

每天备份,随时转为私有部署

免费试用