时间戳

首页 > 解决方案 > 如何从我的 FastAPI 应用程序向另一个站点 (API) 发送 HTTP 请求？

python - 如何从我的 FastAPI 应用程序向另一个站点 (API) 发送 HTTP 请求？

问题描述

我正在尝试http://httpbin.org/uuid使用以下代码片段一次向服务器发送 100 个请求

from fastapi import FastAPI
from time import sleep
from time import time
import requests
import asyncio

app = FastAPI()

URL= "http://httpbin.org/uuid"


# @app.get("/")
async def main():
    r = requests.get(URL)
    # print(r.text)
    
    return r.text

async def task():
    tasks = [main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main()]
    # print(tasks)
    # input("stop")
    result = await asyncio.gather(*tasks)
    print (result)

@app.get('/')
def f():
    start = time()
    asyncio.run(task())
    print("time: ",time()-start)

我正在使用带有 Asyncio 的 FastAPI 来实现大约 3 秒或更短的最短时间，但使用上述方法，我得到的总时间为 66 秒，超过一分钟。我还想保留该main功能以在r.text. 我知道要达到如此短的时间，需要并发，但我不确定我在这里犯了什么错误。

标签： pythonasync-awaithttprequestpython-asynciofastapi

解决方案

requests是一个同步库。您需要使用asyncio基于 - 的库来异步发出请求。

httpx

httpx.AsyncClient通常在 FastAPI 应用程序中用于请求外部服务。它也用于应用程序的异步测试。默认使用它。

from fastapi import FastAPI
from time import time
import httpx
import asyncio

app = FastAPI()

URL = "http://httpbin.org/uuid"


async def request(client):
    response = await client.get(URL)
    return response.text


async def task():
    async with httpx.AsyncClient() as client:
        tasks = [request(client) for i in range(100)]
        result = await asyncio.gather(*tasks)
        print(result)


@app.get('/')
async def f():
    start = time()
    await task()
    print("time: ", time() - start)

输出

['{\n  "uuid": "65c454bf-9b12-4ba8-98e1-de636bffeed3"\n}\n', '{\n  "uuid": "03a48e56-2a44-48e3-bd43-a0b605bef359"\n}\n',...
time:  0.5911855697631836

aiohttp

aiohttp也可以在 FastAPI 应用程序中使用，但如果你真的需要它，请这样做。

from fastapi import FastAPI
from time import time
import aiohttp
import asyncio

app = FastAPI()

URL = "http://httpbin.org/uuid"


async def request(session):
    async with session.get(URL) as response:
        return await response.text()


async def task():
    async with aiohttp.ClientSession() as session:
        tasks = [request(session) for i in range(100)]
        result = await asyncio.gather(*tasks)
        print(result)


@app.get('/')
async def f():
    start = time()
    await task()
    print("time: ", time() - start)

如果要限制并行执行的请求数，可以asyncio.semaphore这样使用：

MAX_IN_PARALLEL = 10
limit_sem = asyncio.Semaphore(MAX_IN_PARALLEL)


async def request(client):
    async with limit_sem:
        response = await client.get(URL)
        return response.text

推荐阅读