Celery执行Django PostgreSQL批量插入任务未更新数据库的问题排查

针对你遇到的Celery任务显示成功但PostgreSQL数据库未更新的情况，可按以下步骤排查解决：

1. 确保Celery Worker正确初始化Django环境 #

Celery进程独立于Django应用，必须手动加载Django配置才能让ORM正常工作，这是最常见的问题根源：

• 在项目根目录创建celery.py文件，内容如下：

import os
from celery import Celery

# 设置Django配置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

app = Celery('你的项目名')
# 从Django settings加载Celery配置，自动处理CELERY_前缀的变量
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有app中的任务
app.autodiscover_tasks()

• 在项目根目录的__init__.py中添加：

from .celery import app as celery_app

__all__ = ('celery_app',)

重启Celery Worker，确保它能正确加载Django的数据库配置。

2. 检查文件路径有效性 #

API进程和Celery Worker的工作目录可能不同，相对路径会导致Worker找不到CSV文件，最终读取空数据：

• 调用任务时传递绝对路径：

import os
from django.conf import settings

# 基于项目根目录生成绝对路径
file_path = os.path.join(settings.BASE_DIR, "path_to_file.csv")
db_operation.delay(file_path)

3. 添加日志排查数据流转问题 #

在任务中加入日志，确认每一步的数据处理情况，判断是读取文件失败还是批量插入无数据：

修改任务代码：

import logging
from os import PathLike

import pandas as pd
from app.models import MyModel
from celery import shared_task

logger = logging.getLogger(__name__)

@shared_task()
def db_operation(csv_file: PathLike) -> None:
    logger.info(f"开始处理文件: {csv_file}")
    try:
        dataframe = pd.read_csv(csv_file)
        logger.info(f"原始数据行数: {len(dataframe)}")
        
        dataframe.dropna(subset="code", inplace=True)
        logger.info(f"去除空code后行数: {len(dataframe)}")
        
        dataframe = dataframe.drop_duplicates(subset=["code"], keep="first")
        logger.info(f"去重后行数: {len(dataframe)}")

        dataset = [
            MyModel(code=row.code, title=row.title)
            for row in dataframe.itertuples()
        ]
        logger.info(f"待插入对象数量: {len(dataset)}")

        created_objs = MyModel.objects.bulk_create(dataset)
        logger.info(f"成功插入{len(created_objs)}条数据")

        return "finished"
    except Exception as e:
        logger.error(f"任务执行出错: {str(e)}", exc_info=True)
        raise

查看Celery Worker的日志，就能明确哪一步出现异常或数据为空。

4. 验证数据库连接一致性 #

确保Celery Worker所在环境的数据库配置和Django应用完全一致：

• 检查环境变量中的数据库连接信息（HOST、PORT、USER、PASSWORD）是否正确
• 确认Worker连接的是目标PostgreSQL实例，而非其他测试数据库

5. 手动强制事务提交（可选） #

如果Django配置了特殊事务策略，可在批量插入时手动启用原子事务：

from django.db import transaction

# ... 数据处理代码 ...

with transaction.atomic():
    MyModel.objects.bulk_create(dataset)

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