并发启动函数并提前返回

Question

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

这是我目前掌握的情况，但它是同步运行的:

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

问题其实是由两部分组成的

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

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

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

您可以使用异步超文本传输协议请求库和一些回调集合，多重处理来生成一个新进程，或者更奇特的东西，比如独立程序（可能有管道或套接字来通信）。

也许这种形式对你有用

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()

Answer 2

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

我们可以按照alex_noname的建议删除等待任务行。它将起作用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'}}

正如HTF所建议的那样，我最终使用了背景任务，因为我已经在使用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'}}

这甚至在< code > def slow _ request(data):之前没有< code>async的情况下也能工作