如何在Celery任务中通过Tortoise-ORM操作数据库？

核心问题分析 #

你的Celery worker运行在独立容器中，和FastAPI应用完全隔离，因此无法复用FastAPI里的Tortoise-ORM初始化配置；同时Tortoise-ORM是异步ORM，而Celery默认执行同步任务，需要额外处理异步代码的运行逻辑。

具体解决方案 #

1. 为Celery单独初始化Tortoise-ORM

创建专门用于Celery的数据库初始化函数，在Celery worker启动时执行：

# 示例路径：project/celery_utils.py
from tortoise import Tortoise, run_async
from project.core.config import settings

async def init_tortoise_for_celery():
    await Tortoise.init(
        db_url=settings.DATABASE_URL,
        modules={"models": ["project.users.models"]}
    )
    # 若需要自动生成表结构，可取消以下注释
    # await Tortoise.generate_schemas()

def init_celery_db():
    run_async(init_tortoise_for_celery())

在Celery实例初始化文件（比如project/celery.py）中调用该初始化函数：

from celery import Celery
from project.celery_utils import init_celery_db
from project.core.config import settings

app = Celery('tasks', broker=settings.CELERY_BROKER_URL)
app.config_from_object(settings, namespace='CELERY')

# 启动Celery worker时初始化数据库连接
init_celery_db()

2. 修改Celery任务适配异步ORM操作

Tortoise-ORM的查询为异步方法，需在Celery的同步任务中通过asyncio.run()执行异步逻辑：

import asyncio
from celery import shared_task
from project.users.models import User
import logging

logger = logging.getLogger(__name__)

@shared_task()
def task_send_welcome_email(user_pk):
    async def get_user_and_log():
        user = await User.filter(id=user_pk).first()
        if user:
            logger.info(f'send email to {user.email} (ID: {user.id})')
        else:
            logger.warning(f'User with ID {user_pk} not found')
    
    asyncio.run(get_user_and_log())

3. 关键配置注意事项

• 确保settings.DATABASE_URL中的数据库地址使用Docker服务名（比如PostgreSQL容器服务名为db，则URL应为postgres://用户名:密码@db:5432/数据库名），而非localhost，否则Celery容器无法访问数据库。
• 保证Celery容器的环境变量与FastAPI容器一致，能正确读取数据库配置。

额外优化建议 #

可通过Celery的worker_init信号触发初始化，确保每次worker启动都能正确建立连接：

from celery.signals import worker_init

@worker_init.connect
def initialize_tortoise(sender=None, conf=None, **kwargs):
    init_celery_db()

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