使用多个 API 处理异步请求 - Python Flask Redis

Question

我正在开发一个应用程序，该应用程序必须咨询多个 API 以获取信息，并在处理数据后将答案输出给客户端。客户端使用浏览器连接到 Web 服务器来转发请求，然后，Web 服务器将从多个 API 中查找所需的信息，并在加入来自这些 API 的响应后，然后向客户端给出答案。

Web 服务器是使用 Flask 构建的，并且还实现了一个为每个 API 提取所需信息的模块（Python）。由于每个 API 的咨询过程都需要时间，我想给 Web 服务器一个响应超时，因此，在发送请求后，只会使用那些低于时间缓冲区的请求。

我提出的解决方案： 使用 Redis Queue 和 RQ worker 将每个 API 的请求排入队列并将响应存储在 Queue 中，然后等待超时并收集能够在允许的时间内响应的响应。之后，对信息进行处理，并将响应提供给用户。

flask web 服务器的设置如下：

@app.route('/result',methods=["POST"])
def show_result():

inputText = request.form["question"]

tweetModule = Twitter()
tweeterResponse = tweetModule.ask(params=inputText)

redditObject = RedditModule()
redditResponse = redditObject.ask(params=inputText)    

edmunds = Edmunds()
edmundsJson = edmunds.ask(params=inputText)

# More APIs could be consulted here

# Send each request async and the synchronize the responses from the queue

template = env.get_template('templates/result.html')
return render_template(template,resp=resp)

工人：

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

让我们假设每个模块都处理自己的排队过程。

我可以看到前面的一些问题：

• 存储在队列中但未超时的信息会发生什么情况？
• 如何让 Flask 等待，然后从队列中提取响应？
• 如果两个客户在同一时间范围内询问，信息是否可能会混淆？
• 是否有更好的方法来处理异步请求，然后同步响应？

谢谢！

Answer 1

在这种情况下，我更喜欢HTTPX 和flask[async] 的组合

首先 - HTTPX

HTTPX 默认提供标准同步 API，但如果您需要，还可以选择 async 客户端。

Async 是一种比multi-threading 高效得多的并发模型，并且可以提供显着的性能优势并支持使用长寿命的网络连接，例如WebSockets。

如果您使用的是异步 Web 框架，那么您还需要使用 async 客户端来发送传出 HTTP requests。

>>> async with httpx.AsyncClient() as client:
...     r = await client.get('https://www.example.com/')
...
>>> r
<Response [200 OK]>

第二 - 在 flask 中使用 async 和 await

如果使用async extra (pip install flask[async]) 安装 Flask，路由、错误处理程序、请求前、请求后和拆卸函数都可以是协程函数。它需要 Python 3.7+，其中contextvars.ContextVar 可用。这允许使用async def 定义视图并使用await。

例如，你应该这样做：

import asyncio
import httpx
from flask import Flask, render_template, request


app = Flask(__name__)


@app.route('/async', methods=['GET', 'POST'])
async def async_form():

    if request.method == 'POST':

        ...

        async with httpx.AsyncClient() as client:
            tweeterResponse, redditResponse, edmundsJson = await asyncio.gather(
                client.get(f'https://api.tweeter....../id?id={request.form["tweeter_id"]}', timeout=None),
                client.get(f'https://api.redditResponse.....?key={APIKEY}&reddit={request.form["reddit_id"]}'),
                client.post(f'https://api.edmundsJson.......', data=inputText)
            )

        ...

        resp = {
            "tweeter_response" : tweeterResponse,
            "reddit_response": redditResponse,
            "edmunds_json" : edmundsJson
        }



    template = env.get_template('templates/result.html')
    return render_template(template, resp=resp)