时间戳

首页 > 解决方案 > 并发启动函数并提前返回

python - 并发启动函数并提前返回

问题描述

我需要从我的 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'}}

标签： pythonpython-asynciofastapi

解决方案

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

解决方案#1

我们可以await task按照 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'}}

解决方案#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'}}

这甚至可以在没有async之前使用def slow_request(data):

推荐阅读