并发启动函数并提前返回

Question

我需要从我的 FastAPI 路径操作返回一个响应，但在此之前我想发送一个慢速请求并且我不需要等待该请求的结果，只要在以下情况下记录错误有的。我可以通过 Python 和 FastAPI 做到这一点吗？我不想将 Celery 添加到项目中。

这是我目前所拥有的，但它是同步运行的：

import asyncio
import requests


async def slow_request(data):
    url = 'https://external.service'
    response = requests.post(
        url=url,
        json=data,
        headers={'Auth-Header': settings.API_TOKEN}
    )
    if not response.status_code == 200:
        logger.error('response:', response.status_code)
        logger.error('data', data)


@router.post('/order/')
async def handle_order(order: Order):
    json_data = {
        'order': order
    }
    
    task = asyncio.create_task(
        slow_request(json_data)
    )
    await task

    return {'body': {'message': 'success'}}

Answer 1

好的，如果没有人想在这里发布答案，解决方案是：

解决方案 #1

我们可以按照 alex_noname 的建议删除 await task 行。它会起作用，因为create_task 安排了任务，我们不再等待它的完成。

@router.post('/order/')
async def handle_order(order: Order):
    json_data = {
        'order': order
    }
    
    task = asyncio.create_task(
        slow_request(json_data)
    )

    return {'body': {'message': 'success'}}

解决方案 #2

我最终得到了 HTF 建议的 BackgroundTasks，因为无论如何我已经在使用 FastAPI，而且这个解决方案对我来说似乎更简洁。

@router.post('/order/')
async def handle_order(order: Order, background_tasks: BackgroundTasks):
    json_data = {
        'order': order
    }
    
    background_tasks.add_task(slow_request, json_data)

    return {'body': {'message': 'success'}}

即使在def slow_request(data): 之前没有async 也可以使用

Answer 2

这个问题真的是两部分

• 请求库是同步的，所以requests.post(...) 将阻塞事件循环直到完成
• 您不需要 Web 请求的结果来响应客户端，但您当前的处理程序在请求完成之前无法响应客户端（即使它是异步的）

考虑将请求逻辑分离到另一个进程中，这样它就可以按照自己的速度发生。

关键是您可以将工作放入某种类型的队列中以最终完成，而无需直接响应客户端的结果。

您可以使用异步 http 请求库和一些回调集合，multiprocessing 来生成新进程，或者像独立程序这样更奇特的东西（可能使用管道或套接字进行通信）。

也许这种形式的东西对你有用

import base64
import json
import multiprocessing

URL_EXTERNAL_SERVICE = "https://example.com"
TIMEOUT_REQUESTS = (2, 10)  # always set a timeout for requests
SHARED_QUEUE = multiprocessing.Queue()  # may leak as unbounded

async def slow_request(data):
    SHARED_QUEUE.put(data)
    # now returns on successful queue put, rather than request completion

def requesting_loop(logger, Q, url, token):
    while True:  # expects to be a daemon
        data = json.dumps(Q.get())  # block until retrieval (non-daemon can use sentinel here)
        response = requests.post(
            url,
            json=data,
            headers={'Auth-Header': token},
            timeout=TIMEOUT_REQUESTS,
        )
        # raise_for_status() --> try/except --> log + continue
        if response.status_code != 200:
            logger.error('call to {} failed (code={}) with data: {}'.format(
                url, response.status_code,
                "base64:" + base64.b64encode(data.encode())
            ))

def startup():  # run me when starting
    # do whatever is needed for logger
    # create a pool instead if you may need to process a lot of requests
    p = multiprocessing.Process(
        target=requesting_loop,
        kwargs={"logger": logger, "Q": SHARED_QUEUE, "url": URL_EXTERNAL_SERVICE, "token": settings.API_TOKEN},
        daemon=True
    )
    p.start()