Python的Django框架中如何实现按月自动分表的MySQL架构？

Django ORM本身不支持每月自动制作表格，必须由应用层控制；可行的方案是继承抽象基础类别和动态结构子类别并注册，并在写入之前将手动路由写入相应的月度表格。跨月查询需要使用本地SQL UNION或逐月合并。

在Django的ORM设计中，假设每个模型对应一个固定表名的物理表，db_table 它是一个静态字符串，在运行过程中不支持动态拼接。试着在运行过程中 Meta.db_table 里写 f"logs_{timezone.now().strftime('%Y%m')}" 由于迁移系统、反射机制、缓存键都依赖于确定的表名，因此会导致迁移失败、admin报错、查询混乱。

只有一条真正可行的路径:放弃“一个Model对应多个物理表”的幻想，改用“逻辑表” + 多个继承子类 + 手动路由模式。核心是在操作前将分表逻辑从ORM元数据中剥离出来，交给业务代码和数据库。

• 所有按月分表的模型必须继承同一抽象基类(带) abstract = True）
• 每个具体月份的子类显式声明 db_table，如 logs_202409，而且物理表需要提前创建(不能依赖) makemigrations）
• 读写操作前，根据时间字段计算目标表名，然后决定使用哪个子类
• 跨月查询必须手动进行 union 或用原生 SQL，Django QuerySet 不能跨模型合并

每个月手动写一个模型是不现实的。需要使用 Python 的 type() 注入动态结构类，并注入动态结构类 Django 在模型注册系统中。关键是类别必须在 Django 启动完成、AppConfig 加载后才能注册，否则加载后才能注册 apps.get_model() 找不到。

推荐放在 apps.py 的 ready() 初始化的方法，如：

立即学习“Python免费学习笔记(深入)；

def ready(self, *args, **kwargs):
    from django.apps import apps
    from datetime import datetime, timedelta
    # 取最近12个月 + 在接下来的三个月里，它将覆盖常用范围
    base_date = datetime.now()
    for i in range(-12, 3):
        month_date = base_date + timedelta(days=i*30)
        table_name = f"logs_{month_date.strftime('%Y%m')}"
        class_name = f"LogEntry_{month_date.strftime('%Y%m')}"
        # 动态创建模型类
        model_class = type(
            class_name,
            (LogEntryBase,),  # 继承抽象基类
            {
                "Meta": type("Meta", (), {"db_table": table_name, "managed": False}),
                "__module__": "myapp.models",
            }
        )
        # 注册进 Django 模型系统
        apps.register_model("myapp", model_class)

注意：managed = False 表示不参与迁移；表结构必须与基类完全一致，否则字段映射错误。

Python数据分析助手

为业务和科研数据的快速处理提供Python数据清理、统计分析和可视化建议。

• 每次部署或重启都要重新注册，不能只注册一次
• 类名必须是全局唯一的，否则， get_model() 会回到错误类
• 如果用了 select_related 或者反向关系，这条路基本上是不可能的——跨表外键 MySQL 不能在里面约束，Django 也不支持

不能依赖信号或重写 save()，因为 save() 调用时可能会错过时间计算机(例如，包含历史时间的数据被批量导入)。最安全的方法是包装一个写入函数，强制业务层调用它:

def write_log_entry(entry_data: dict) -> None:
    dt = entry_data.get("created_at") or timezone.now()
    table_suffix = dt.strftime("%Y%m")
    model_class = apps.get_model("myapp", f"LogEntry_{table_suffix}")
    instance = model_class(**entry_data)
    instance.save()

这个函数需要处理两个常见的坑：

• 当目标表不存在时，save() 报 OperationalError: (1146, "Table 'db.logs_202413' doesn't exist") —— 必须提前用 SHOW TABLES LIKE 'logs_202413' 检查并自动建表(通过原始检查) SQL 或 django.db.connection.cursor()）
• 并发写入同一个新表时，可能会同时触发多个请求，导致建表 1050 Table already exists 错误，需加 try/except 忽略该错误
• 如果 entry_data 包含外键 ID，确保存在关联表 ID 有效；分表后外键实际失效，只能靠业务逻辑保证

你不能写 LogEntry_202408.objects.filter(...) | LogEntry_202409.objects.filter(...)，因为 | 是 QuerySet 的 OR 操作，不是 UNION，它会生成 LEFT JOIN 或子查询，结果混乱而缓慢。真正的需求是 UNION ALL。

只有两种正确的方法：

• 用 django.db.connection.cursor() 写原生 SQL，手动拼接 SELECT ... FROM logs_202408 UNION ALL SELECT ... FROM logs_202409，再用 dictfetchall() 解析结果
• 逐月查询再 Python 层合并(适用于数据量少、月份少的场景)，但要注意排序和分页不能下推，容易 OOM

另外，count() 在 UNION 必须在场景中包裹一层 SELECT COUNT(*) FROM (UNION ...) AS t，Django 这层包装不会帮你做。

最容易被忽视的一点:MySQL 的 UNION 每个句子的段名、类型和顺序都必须严格一致。如果您在一个月表中添加了一个新的字段，而其他手表没有同步，则查询将直接失败。表格不是一劳永逸的，表格结构的变化必须完全同步到所有现有的月表。