无法在Celery/ARQ任务中使用Beanie ODM API的解决方案咨询

核心原因 #

任务代码（如异步任务、后台任务）运行在独立的执行上下文，Beanie默认的MongoDB连接会话没有绑定到这个上下文，导致无法识别集合初始化状态。手动传递会话可以让Beanie在指定上下文里操作数据库。

具体实现步骤 #

1. 获取Beanie的会话对象 #

在已经完成Beanie初始化的上下文（比如FastAPI请求上下文、主程序初始化后的代码）中，可通过Beanie提供的get_session()方法获取会话：

from beanie import get_session

async def get_db_session():
    async with get_session() as session:
        yield session

2. 修改任务函数接收会话参数 #

调整任务函数，添加session参数，并在所有Beanie文档操作中显式传入该会话：

from beanie import Document

class MyDocument(Document):
    name: str

# 改造后的任务函数
async def task_function(session, doc_id: str):
    # 查询操作传入session
    doc = await MyDocument.get(doc_id, session=session)
    doc.name = "Updated"
    # 保存操作传入session
    await doc.save(session=session)

3. 不同场景下的会话传递方式 #

场景一：FastAPI后台任务

通过依赖注入获取会话，再传递给后台任务：

from fastapi import FastAPI, BackgroundTasks, Depends

app = FastAPI()

@app.post("/update-doc/{doc_id}")
async def trigger_task(doc_id: str, background_tasks: BackgroundTasks, session=Depends(get_db_session)):
    background_tasks.add_task(task_function, session, doc_id)
    return {"status": "task started"}

场景二：Celery异步任务（AsyncIO版本）

主程序初始化时保存MongoDB客户端实例，任务中基于该客户端创建会话并传入：

# 主程序初始化环节
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie

client = AsyncIOMotorClient("mongodb://localhost:27017")
await init_beanie(database=client["my_db"], document_models=[MyDocument])

# Celery任务定义
import asyncio
from celery import Celery

celery = Celery("tasks", broker="redis://localhost:6379/0")

@celery.task(name="update_doc_task")
def update_doc_task(doc_id: str):
    async def run_task():
        async with client.start_session() as session:
            await task_function(session, doc_id)
    asyncio.run(run_task())

场景三：通用异步任务队列

无论使用哪种任务框架，核心逻辑都是基于已初始化的MongoDB客户端创建会话，再传递给Beanie操作：

async def run_async_task(doc_id: str):
    # 确保客户端已全局初始化
    async with client.start_session() as session:
        await task_function(session, doc_id)

注意事项 #

• 确保MongoDB客户端在任务进程/线程中全局可用，避免重复初始化连接
• Beanie所有文档操作方法（get()、save()、find()等）均支持传入session参数
• 若使用Beanie事务，会话是必需的，事务操作必须绑定到同一个会话

内容的提问来源于stack exchange，提问作者timur zolotov