FastAPI中回调通知的跨端点复用及数据库会话同步问题

问题根源 #

原调用端点和回调端点属于两个独立的HTTP请求，各自使用独立的数据库会话。回调端点更新数据后，若原端点复用旧会话查询，会读取会话缓存的旧数据；即使原端点用新会话，若依赖轮询数据库，效率也极低。

可行解决办法 #

1. 强制刷新数据库会话（适用于轮询场景） #

如果原端点采用轮询方式等待状态更新，每次查询时必须从数据库重新加载数据，而非依赖会话缓存：

from sqlalchemy.orm import Session
from models import Task

def fetch_latest_task_status(db: Session, task_id: str):
    task = db.query(Task).filter(Task.id == task_id).first()
    db.refresh(task)  # 强制从数据库拉取最新数据，忽略会话缓存
    return task.status

• 注意：避免长时间持有同一个数据库会话，每次查询都应获取新会话，防止会话超时或资源泄漏。

2. 异步事件通知（推荐方案） #

用跨请求的消息传递机制，让回调端点直接通知原端点状态变化，无需轮询数据库：

单实例部署（用asyncio.Queue）

from fastapi import FastAPI, BackgroundTasks
import asyncio
from typing import Dict
from sqlalchemy.orm import Session
from models import Task
from database import get_db

# 全局存储任务ID与对应等待队列的映射
task_wait_queues: Dict[str, asyncio.Queue] = {}

app = FastAPI()

# 原调用第三方API的端点
@app.post("/initiate-task")
async def initiate_task(task_id: str, background_tasks: BackgroundTasks):
    # 为当前任务创建等待队列
    task_wait_queues[task_id] = asyncio.Queue(maxsize=1)
    # 后台调用第三方API
    background_tasks.add_task(call_third_party_service, task_id)
    
    # 等待回调通知，设置超时避免无限阻塞
    try:
        status = await asyncio.wait_for(task_wait_queues[task_id].get(), timeout=300)
    except asyncio.TimeoutError:
        status = "timeout"
    finally:
        # 清理队列，避免内存泄漏
        task_wait_queues.pop(task_id, None)
    
    return {"task_id": task_id, "final_status": status}

# 第三方回调接收端点
@app.post("/task-callback")
async def task_callback(task_id: str, status: str):
    # 先更新数据库
    db: Session = next(get_db())
    task = db.query(Task).filter(Task.id == task_id).first()
    if task:
        task.status = status
        db.commit()
    
    # 通知等待中的原端点
    if task_id in task_wait_queues:
        await task_wait_queues[task_id].put(status)
    
    return {"ack": "success"}

# 模拟调用第三方API的函数
async def call_third_party_service(task_id: str):
    # 这里替换为实际调用第三方API的逻辑
    pass

多实例部署（用Redis Pub/Sub）

如果FastAPI是多实例部署，内存队列无法跨实例传递消息，改用Redis的发布订阅机制：

# 原端点等待逻辑改为订阅Redis频道
import redis.asyncio as redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)

@app.post("/initiate-task")
async def initiate_task(task_id: str):
    channel = f"task_status:{task_id}"
    pubsub = redis_client.pubsub()
    await pubsub.subscribe(channel)
    
    # 调用第三方API逻辑...
    
    # 等待回调消息
    try:
        message = await asyncio.wait_for(pubsub.get_message(ignore_subscribe_messages=True), timeout=300)
        status = message["data"].decode()
    except asyncio.TimeoutError:
        status = "timeout"
    finally:
        await pubsub.unsubscribe(channel)
    
    return {"task_id": task_id, "final_status": status}

# 回调端点发布消息到Redis频道
@app.post("/task-callback")
async def task_callback(task_id: str, status: str):
    # 更新数据库逻辑...
    
    channel = f"task_status:{task_id}"
    await redis_client.publish(channel, status)
    return {"ack": "success"}

3. 确保回调端点正确提交数据库会话 #

回调端点必须提交会话，否则更新仅存在于当前会话的缓存中，不会写入数据库：

def update_task_status(db: Session, task_id: str, status: str):
    task = db.query(Task).filter(Task.id == task_id).first()
    if task:
        task.status = status
        db.add(task)
        db.commit()  # 必须执行提交，数据才会持久化到数据库
        db.refresh(task)  # 可选，刷新会话中的对象为最新状态

关键注意事项 #

• 同步端点需改为异步定义（async def）才能使用异步等待逻辑，若需保持同步，可通过线程池处理等待逻辑。
• 多实例场景下必须使用外部消息中间件（Redis、RabbitMQ等），内存队列仅适用于单实例。
• 所有等待逻辑必须设置超时，避免请求无限阻塞占用资源。

内容的提问来源于stack exchange，提问作者SATNAM SINGH