FastAPI+SQLAlchemy+Celery集成遇UnboundExecutionError问题求助

我基于FastAPI搭建API，使用SQLAlchemy定义模型并操作Postgres数据库，集成Celery作为任务管理器时，含SQLAlchemy调用的Celery任务抛出UnboundExecutionError，错误信息如下：

UnboundExecutionError('Could not locate a bind configured on mapper Mapper[CampaignORM(campaign)], SQL expression or this Session.')

已尝试禁用会话池、在celery.py中初始化数据库会话，但问题仍未解决。

1. Celery Worker是独立于FastAPI的进程，全局的app.db会话无法跨进程共享，即便在celery.py中初始化了app，Session的绑定关系在Worker进程中也可能未正确建立
2. SQLAlchemy的Session并非线程/进程安全，复用全局Session会导致绑定失效或连接异常
3. 使用async_to_sync调用异步函数时，原异步函数依赖的全局app.db在Celery的同步环境中缺少正确的绑定上下文

1. 为每个Celery任务创建独立数据库会话 #

禁止复用FastAPI的全局Session，在Celery任务内部单独初始化数据库连接和Session，确保每个任务拥有独立且绑定正确的Session实例。

2. 重构数据库操作函数，通过参数传入Session #

修改异步的campaign_list函数，取消对全局app.db的依赖，改为接收Session参数，使其在FastAPI和Celery环境中都能适配不同的Session实例。

3. 修正Celery任务中的异步调用逻辑 #

在Celery任务内初始化Session后，将其传入campaign_list函数，避免依赖全局上下文。

第一步：修改campaign_list函数，新增Session参数 #

async def campaign_list(db_session, campaign_id, user_id, response_offset = 0, response_limit=20, include_prompt=False):
    stmt_retr = (select(schema.CampaignORM, func.count(schema.CampaignORM.campaign_id).over().label("total"))
                 .join(schema.AccessMembershipORM, schema.CampaignORM.access_id == schema.AccessMembershipORM.access_id)
                 .where(schema.AccessMembershipORM.user_id == user_id)
                 )
    
    # 分组筛选目标 campaign
    stmt_retr = (stmt_retr.group_by(schema.CampaignORM.campaign_id)
                 .order_by(schema.CampaignORM.timestamp_created.desc())
                 .offset(response_offset).limit(response_limit)
                 )

    campaign_list = []
    total_count = 0
    # 使用传入的db_session执行查询
    for db_item, _local_count in db_session.execute(stmt_retr):
        total_count = _local_count
        stmt_retr = (select(schema.CampaignORM.campaign_id, schema.CampaignORM.campaign_type)
                     .where(schema.CampaignORM.component_list_id == db_item.component_list_id)
                     )
        list_components = {x.campaign_id: x.campaign_type for x in db_session.execute(stmt_retr).all()}

        # 其余业务逻辑保持不变
        campaign_info = CampaignIdentifier(...lots of code...)
    return resp_new

第二步：修改Celery任务，初始化独立Session并传入 #

@celery.task(name="generate content")
def generate_content(campaign_id, user_id, num_responses, media_type, simulate_only, clone_source, num_historical):
    # 在任务内部初始化数据库连接和Session
    from ..database.schema import engine_init_config, session_init
    # 加载数据库配置（需根据实际项目实现get_config方法）
    config = get_config()
    sql_engine = engine_init_config(config)
    if not sql_engine:
        raise Exception("数据库初始化失败")
    db_session = session_init(sql_engine)
    
    try:
        # 将独立Session传入campaign_list
        resp_campaign = async_to_sync(campaign_list)(db_session, campaign_id, user_id)
        obj_campaign_source = resp_campaign.campaign_list[0]
        return obj_campaign_source
    finally:
        # 任务结束后关闭Session，避免连接泄漏
        db_session.close()

第三步：FastAPI路由中调用campaign_list时传入全局Session #

# 示例FastAPI路由
@app.get("/campaigns/{campaign_id}")
async def get_campaign(campaign_id: str, user_id: str):
    resp = await campaign_list(app.db, campaign_id, user_id)
    return resp

• 禁止在多进程/多线程环境（如Celery）中使用全局SQLAlchemy Session
• 每个数据库操作的Session应在使用完成后及时关闭，避免连接池耗尽
• 若使用SQLAlchemy 2.0+，可考虑搭配Celery异步任务（Celery 5.0+支持）和异步数据库驱动（如asyncpg代替psycopg2），进一步优化异步流程

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