FastAPI+Celery集成Tortoise ORM遇InterfaceError：连接池正在关闭

问题场景 #

使用FastAPI搭配Tortoise ORM，后台通过Celery执行4-5次带select_for_update的数据查询与更新操作时，触发如下错误：

File "/usr/local/lib/python3.10/site-packages/tortoise/queryset.py", line 1006, in _execute
  instance_list = await self._db.executor_class(
File "/usr/local/lib/python3.10/site-packages/tortoise/backends/base/executor.py", line 130, in execute_select
  _, raw_results = await self.db.execute_query(query.get_sql())
File "/usr/local/lib/python3.10/site-packages/tortoise/backends/base_postgres/client.py", line 24, in _translate_exceptions
  return await self._translate_exceptions(func, *args, **kwargs)
File "/usr/local/lib/python3.10/site-packages/tortoise/backends/asyncpg/client.py", line 79, in _translate_exceptions
  return await func(self, *args, **kwargs)
File "/usr/local/lib/python3.10/site-packages/tortoise/backends/asyncpg/client.py", line 126, in execute_query
  async with self.acquire_connection() as connection:
File "/usr/local/lib/python3.10/site-packages/tortoise/backends/base/client.py", line 328, in __aenter__
  self.connection = await self.pool.acquire()
File "/usr/local/lib/python3.10/site-packages/asyncpg/pool.py", line 822, in _acquire
  raise exceptions.InterfaceError('pool is closing')

错误本质 #

数据库连接池正在关闭/已销毁时，Celery后台任务仍尝试获取连接。常见触发原因：

• FastAPI主进程重启/关闭时，连接池被销毁，但Celery任务未同步终止
• Tortoise ORM连接池配置与Celery异步任务生命周期不兼容，任务执行时连接池已被回收

解决方案 #

1. 为Celery任务单独初始化数据库连接 #

Celery工作进程与FastAPI主进程完全分离，不能复用主进程的Tortoise连接。需在任务启动时单独初始化，结束后关闭：

from celery import Celery
from tortoise import Tortoise

celery_app = Celery('tasks', broker='redis://localhost:6379/0')

# 复用FastAPI中的数据库配置
DB_CONFIG = {
    "connections": {
        "default": "postgres://user:pass@localhost/dbname"
    },
    "apps": {
        "models": {
            "models": ["your_app.models"],
            "default_connection": "default",
        }
    }
}

@celery_app.task(bind=True)
async def background_data_task(self):
    # 任务初始化阶段启动Tortoise连接
    await Tortoise.init(config=DB_CONFIG)
    try:
        # 执行带行锁的查询与更新
        obj = await YourModel.get(id=1).select_for_update()
        obj.status = "processed"
        await obj.save()
        # ...后续4-5次数据操作
    finally:
        # 任务结束后强制关闭连接
        await Tortoise.close_connections()

2. 调整连接池闲置存活时长 #

修改FastAPI中Tortoise的连接池配置，延长闲置连接的存活时间，避免主进程暂时闲置时池被提前销毁：

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

register_tortoise(
    app,
    db_url="postgres://user:pass@localhost/dbname",
    modules={"models": ["your_app.models"]},
    connection_config={
        "max_inactive_connection_lifetime": 300  # 5分钟，根据任务实际耗时调整
    },
    generate_schemas=True,
)

3. 用Celery信号管理连接生命周期 #

通过Celery的worker初始化/关闭信号，让每个工作进程独立管理Tortoise连接：

@celery_app.on_after_configure.connect
def setup_tortoise(sender, **kwargs):
    async def init_conn():
        await Tortoise.init(config=DB_CONFIG)
    # 仅在worker启动时执行一次初始化
    sender.add_periodic_task(1.0, init_conn, expires=1)

@celery_app.on_worker_shutdown.connect
def shutdown_tortoise(sender, **kwargs):
    async def close_conn():
        await Tortoise.close_connections()
    # worker关闭时清理连接
    sender.add_periodic_task(1.0, close_conn, expires=1)

4. 显式管理单任务内的连接 #

如果任务操作耗时较长，可显式获取/释放连接，避免连接被池自动回收：

async def batch_update():
    conn = await Tortoise.get_connection("default")
    try:
        async with conn.transaction():
            # 绑定连接执行带锁查询
            obj = await YourModel.get(id=1).using_db(conn).select_for_update()
            obj.value += 1
            await obj.save(using_db=conn)
            # ...其他数据操作
    finally:
        await conn.close()

内容的提问来源于stack exchange，提问作者Shruthi Sagar CR