FastApi 与其他 Api 的通信答案

【问题标题】：FastApi communication with other ApiFastApi 与其他 Api 的通信
【发布时间】：2020-07-09 20:20:45
【问题描述】：

我最近在使用 fastapi，作为练习，我想将我的 fastapi api 与其他服务器上的验证服务连接起来......但我不知道该怎么做，我还没有找到可以帮助我的东西官方文档..我必须用python代码来做吗？或者有什么办法？

FastApi docs

感谢您的帮助，请原谅我的英语。

【问题讨论】：

要连接到其他 REST 服务，请使用requests library

标签： python api rest fastapi

【解决方案1】：

您需要使用 Python 对其进行编码。

如果您使用异步，则应使用也是异步的 HTTP 客户端，例如 aiohttp。

import aiohttp

@app.get("/")
async def slow_route():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://validation_service.com") as resp:
            data = await resp.text()
            # do something with data

【讨论】：

【解决方案2】：

接受的答案当然有效，但它不是一个有效的解决方案。对于每个请求，ClientSession 都会关闭，因此我们失去了ClientSession 的优势[0]：连接池、keepalives 等。

我们可以使用FastAPI中的startup和shutdown事件[1]，分别在服务器启动和关闭时触发。在这些事件中，可以创建一个ClientSession 实例并在整个应用程序的运行时使用它（从而充分发挥其潜力）。

ClientSession 实例存储在应用程序状态中。 [2]

这里我在aiohttp服务器的上下文中回答了一个非常相似的问题：https://stackoverflow.com/a/60850857/752142

from __future__ import annotations

import asyncio
from typing import Final

from aiohttp import ClientSession
from fastapi import Depends, FastAPI
from starlette.requests import Request

app: Final = FastAPI()


@app.on_event("startup")
async def startup_event():
    setattr(app.state, "client_session", ClientSession(raise_for_status=True))


@app.on_event("shutdown")
async def shutdown_event():
    await asyncio.wait((app.state.client_session.close()), timeout=5.0)


def client_session_dep(request: Request) -> ClientSession:
    return request.app.state.client_session


@app.get("/")
async def root(
    client_session: ClientSession = Depends(client_session_dep),
) -> str:
    async with client_session.get(
        "https://example.com/", raise_for_status=True
    ) as the_response:
        return await the_response.text()

【讨论】：