如何在Django 5.0.3中后台运行MySQL原子事务完成5万行CSV导入？

问题背景 #

我需要在Django中将包含5万行数据的CSV文件导入MySQL数据表，已编写对应模型，通过模板的Ajax异步POST请求启动原子事务，但事务被StateReloader中断，无法完成。想知道能否将该事务置于后台运行直至完成？

代码示例 #

models.py

from django.db import models, transaction

class MyManager(models.Manager):
    @transaction.atomic
    def create_models(self, file, callback=None):
        if callback is not None:
            transaction.on_commit(callback)
        with open(file, "r") as fp:
            lines = map(str.strip, fp.readlines())
            lines = map(lambda x: x.split(";"), lines)
            lines = list(lines)
        headers = lines[0]
        print("Writing table with {} lines...".format(len(lines) - 1)) # 50k lines
        for i, line in enumerate(lines[1:]):
            for j in range(len(headers)):
                print("Saving line #{}".format(i), end="\r")          # ends at line 800 before the StateReloader
                self.create(row=i + 1, header=headers[j], value=line[j])
        print()
        return True


class MyModel(models.Model):
    row = models.IntegerField(null=False, blank=False)
    header = models.CharField(max_length=255, null=False, blank=False)
    value = models.CharField(max_length=255, null=False, blank=False)
    objects = MyManager()
    
    def __str__(self) -> str:
        return f"{self.header} = {self.value} on row {self.row}"

views.py

import os
from django.http import JsonResponse
from django.views import generic
from django.contrib.auth.mixins import LoginRequiredMixin


from .models import MyModel


def end_transaction():
    print("Done!")


class MyView(LoginRequiredMixin, generic.TemplateView):
    template_name = "edit_csv.html"

    def get(self, request, *args, **kwargs):
        context = self.get_context_data(*args, **kwargs)
        return self.render_to_response(context)

    def post(self, request, *args, **kwargs):
        context = self.get_context_data(*args, **kwargs)
        filename = request.POST.get("filename", "").strip()
        if filename != "":
            filename = os.path.basename(filename)
            routine = MyModel.objects.create_models(filename, end_transaction)
        return JsonResponse(self.render_to_response(context), safe=False)

    def get_context_data(self, *args, **kwargs):
        context = super(MyView, self).get_context_data(*args, **kwargs)
        return context

解决方案 #

直接在请求响应周期内处理5万行数据，必然会因超时或开发环境的StateReloader自动重载机制中断，必须将导入逻辑移至后台异步执行。

临时方案：关闭开发环境自动重载 #

如果仅在开发环境测试，可以启动Django时关闭自动重载，避免进程中断：

python manage.py runserver --noreload

但这只是临时手段，生产环境无需依赖该机制，核心解决方案是异步任务。

推荐方案：使用后台任务框架处理 #

通过Celery或django-background-tasks等工具，将导入逻辑封装为后台任务，脱离请求周期执行。以下是django-background-tasks的实现示例：

1. 安装依赖并配置

pip install django-background-tasks

在settings.py中添加应用：

INSTALLED_APPS = [
    # ... 其他已安装应用
    'background_task',
]

2. 修改模型层，优化导入逻辑+封装后台任务

# models.py
from django.db import models, transaction
from background_task import background

class MyManager(models.Manager):
    def create_models(self, file):
        with open(file, "r") as fp:
            lines = map(str.strip, fp.readlines())
            lines = map(lambda x: x.split(";"), lines)
            lines = list(lines)
        headers = lines[0]
        print("开始导入，共{}行数据...".format(len(lines) - 1))
        
        # 批量插入优化：避免循环调用create，降低数据库开销
        objs_batch = []
        for i, line in enumerate(lines[1:]):
            for j in range(len(headers)):
                objs_batch.append(MyModel(
                    row=i + 1,
                    header=headers[j],
                    value=line[j]
                ))
                # 每1000条批量插入一次，控制内存占用
                if len(objs_batch) >= 1000:
                    with transaction.atomic():
                        MyModel.objects.bulk_create(objs_batch)
                    objs_batch = []
        # 插入剩余数据
        if objs_batch:
            with transaction.atomic():
                MyModel.objects.bulk_create(objs_batch)
        print("导入完成")
        return True

# 定义后台任务，schedule=0表示立即执行
@background(schedule=0)
def import_csv_task(filename):
    # 注意：确保文件为绝对路径，后台进程需能访问到该文件
    MyModel.objects.create_models(filename)

class MyModel(models.Model):
    row = models.IntegerField(null=False, blank=False)
    header = models.CharField(max_length=255, null=False, blank=False)
    value = models.CharField(max_length=255, null=False, blank=False)
    objects = MyManager()
    
    def __str__(self) -> str:
        return f"{self.header} = {self.value} on row {self.row}"

3. 修改视图，触发后台任务

# views.py
import os
from django.http import JsonResponse
from django.views import generic
from django.contrib.auth.mixins import LoginRequiredMixin
from .models import import_csv_task

class MyView(LoginRequiredMixin, generic.TemplateView):
    template_name = "edit_csv.html"

    def post(self, request, *args, **kwargs):
        filename = request.POST.get("filename", "").strip()
        if filename:
            # 建议拼接绝对路径，避免后台进程找不到文件
            full_path = os.path.join("/path/to/your/csv/files", os.path.basename(filename))
            # 触发后台任务
            import_csv_task(full_path)
            return JsonResponse({"status": "success", "msg": "导入任务已启动，后台执行中"})
        return JsonResponse({"status": "error", "msg": "未指定有效文件名"})

4. 启动后台任务进程

python manage.py process_tasks

额外优化建议 #

• 批量插入：用bulk_create替代循环create，能将插入速度提升数倍
• 事务拆分：大事务会占用数据库资源，可按批次拆分事务
• 进度反馈：若需前端展示导入进度，可新增任务状态表，记录已导入行数和状态，前端定时Ajax查询

内容的提问来源于stack exchange，提问作者G.F