如何在FastAPI中实现耗时任务完成后回调外部API功能？

你的代码存在几个关键问题导致回调未生效，以下是问题分析和修正方案：

问题点解析 #

1. 线程初始化错误：Thread(target=do_invoice(invoice, callback_url)) 会直接在主线程执行do_invoice函数，而非将其作为线程目标。正确写法是传递函数对象，通过args参数传入调用所需值。
2. 回调请求格式不匹配：你定义的InvoiceEvent模型要求description和paid字段，但实际发送的是data数组，外部API无法解析该格式。
3. 误解回调路由作用：invoice_notification函数仅用于生成OpenAPI回调规范，告知外部API你的服务将以何种格式调用其回调接口，无需实现业务逻辑，实际回调请求由你的服务主动发起。

修正后的代码 #

from typing import Union

from fastapi import APIRouter, FastAPI, BackgroundTasks
from pydantic import BaseModel, AnyHttpUrl

import requests
import time

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[str, None] = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter()


# 该路由仅用于生成OpenAPI回调规范，无需实现业务逻辑
@invoices_callback_router.post(
    "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):
    pass


@app.post("/invoices/", callbacks=invoices_callback_router.routes)
async def create_invoice(invoice: Invoice, callback_url: Union[AnyHttpUrl, None] = None, background_tasks: BackgroundTasks = None):
    if callback_url and background_tasks:
        # 使用FastAPI内置的BackgroundTasks，比手动线程更适配异步生态
        background_tasks.add_task(do_invoice, invoice, callback_url)
    return {"msg": "Invoice received"}


def do_invoice(invoice: Invoice, callback_url: AnyHttpUrl):
    # 模拟耗时处理任务
    time.sleep(10)
    
    # 构造符合InvoiceEvent模型的回调数据
    callback_data = {
        "description": f"Invoice {invoice.id} processed successfully",
        "paid": True
    }
    
    # 拼接完整的回调URL
    url = f"{callback_url}/invoices/{invoice.id}"
    
    try:
        # 发送回调请求
        response = requests.post(url=url, json=callback_data)
        response.raise_for_status()  # 主动抛出HTTP错误
        print(f"Callback sent successfully, status code: {response.status_code}")
    except requests.exceptions.RequestException as e:
        # 捕获请求异常，避免后台任务崩溃
        print(f"Callback failed: {str(e)}")

关键改进说明 #

• 使用BackgroundTasks：FastAPI内置的后台任务管理器比手动创建线程更安全，会在请求返回后自动处理任务，完美适配FastAPI的异步机制。
• 修正线程调用（若坚持用Thread）：如果不想用BackgroundTasks，需将线程初始化改为Thread(target=do_invoice, args=(invoice, callback_url))，确保函数在子线程中执行。
• 严格匹配请求格式：按照定义的InvoiceEvent结构构造请求体，保证外部API能正确解析回调数据。
• 异常捕获：添加请求异常处理，避免后台任务崩溃影响服务稳定性，同时便于调试问题。

内容的提问来源于stack exchange，提问作者Tim Josefsson