如何在Django/DRF中无Model动态创建CSV驱动的数据库表

核心思路 #

直接通过Django的数据库连接执行原生SQL完成表创建与数据插入，步骤如下：

• 解析上传的CSV文件，提取文件名作为表名（处理为合法标识符）
• 读取CSV表头作为数据库表的字段
• 动态生成CREATE TABLE语句并执行
• 批量插入CSV中的数据行

完整代码实现 #

import io
import csv
import os
from django.db import connection
from rest_framework import generics
from rest_framework.response import Response
from rest_framework import status
from rest_framework import serializers

class FileUploadSerializer(serializers.Serializer):
    file = serializers.FileField()

class UploadProductApiView(generics.CreateAPIView):
    serializer_class = FileUploadSerializer

    def post(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        file = serializer.validated_data['file']
        
        # 1. 处理表名：从文件名提取，去掉后缀，转合法标识符
        filename = os.path.splitext(file.name)[0]
        table_name = filename.lower().replace(' ', '_').replace('-', '_')
        
        # 2. 读取CSV内容
        decoded_file = file.read().decode('utf-8')
        io_string = io.StringIO(decoded_file)
        reader = csv.reader(io_string)
        
        # 获取表头并处理为合法字段名
        headers = next(reader)
        cleaned_fields = [field.lower().replace(' ', '_').replace('-', '_') for field in headers]
        
        try:
            with connection.cursor() as cursor:
                # 3. 动态生成创建表的SQL
                create_table_sql = f"""
                CREATE TABLE IF NOT EXISTS `{table_name}` (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    {', '.join([f'`{field}` VARCHAR(255)' for field in cleaned_fields])}
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                """
                cursor.execute(create_table_sql)
                
                # 4. 批量插入数据
                if cleaned_fields:
                    insert_sql = f"""
                    INSERT INTO `{table_name}` ({', '.join([f'`{field}`' for field in cleaned_fields])})
                    VALUES ({', '.join(['%s'] * len(cleaned_fields))})
                    """
                    data_rows = []
                    for row in reader:
                        processed_row = [value if value else None for value in row]
                        data_rows.append(processed_row)
                    
                    if data_rows:
                        cursor.executemany(insert_sql, data_rows)
            
            return Response({
                'message': f'表 {table_name} 创建并填充数据成功',
                'table_name': table_name,
                'fields': cleaned_fields,
                'row_count': len(data_rows) if 'data_rows' in locals() else 0
            }, status=status.HTTP_201_CREATED)
        
        except Exception as e:
            return Response({
                'error': str(e)
            }, status=status.HTTP_400_BAD_REQUEST)

关键细节说明 #

• 表名与字段名处理：通过替换空格、横杠为下划线并转为小写，用反引号包裹，避免SQL语法错误和注入风险
• 数据类型：示例默认使用VARCHAR(255)，若需适配不同数据类型，可添加逻辑判断（如识别数字转为INT/DECIMAL，日期转为DATE）
• 批量插入：使用executemany替代逐行插入，大幅提升大数据量CSV的处理效率
• 异常处理：捕获SQL执行错误，返回清晰的错误提示
• IF NOT EXISTS：避免重复创建表引发的错误，可根据业务需求移除该选项

注意事项 #

• 确保Django配置的数据库用户拥有创建表的权限
• 加强文件名校验，避免包含特殊字符导致表名不合法
• 大型CSV文件建议搭配异步任务（如原代码中注释的Celery任务），避免请求超时

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