Django中如何关联三张表并自动填充汇总计算结果的新表

第一步：定义模型类 #

首先按照你给出的表结构定义对应的Django模型，通过外键关联三张表（若为存量表无法修改结构可调整为手动关联逻辑，以下为Django标准实现）：

from django.db import models

# 表A模型
class TableA(models.Model):
    code = models.CharField(max_length=32, unique=True, verbose_name="A表编码")
    name = models.CharField(max_length=32, verbose_name="名称")
    class Meta:
        db_table = "table_a"

# 表B模型
class TableB(models.Model):
    code_a = models.ForeignKey(TableA, to_field="code", on_delete=models.CASCADE, db_column="code_A", related_name="b_items")
    code = models.CharField(max_length=32, unique=True, verbose_name="B表编码")
    class Meta:
        db_table = "table_b"

# 表C模型
class TableC(models.Model):
    code_b = models.ForeignKey(TableB, to_field="code", on_delete=models.CASCADE, db_column="code_B", related_name="c_items")
    number = models.IntegerField(verbose_name="数值")
    class Meta:
        db_table = "table_c"

# 物理存储的汇总表模型（不需要落库可以不定义）
class SummaryTable(models.Model):
    name = models.CharField(max_length=32, unique=True, verbose_name="名称")
    number = models.IntegerField(default=0, verbose_name="汇总数值")
    class Meta:
        db_table = "summary_table"

第二步：直接查询获取汇总结果（无需物理存储汇总表） #

如果不需要把汇总结果持久化到数据库，直接用Django ORM的聚合查询即可得到你要的结果：

from django.db.models import Sum, F

# 直接查询得到汇总数据，结构和你预期的结果完全一致
summary_data = TableA.objects.annotate(
    total_number=Sum("b_items__c_items__number")
).values(
    "id", "name", number=F("total_number")
)

第三步：自动填充物理汇总表（需要落库时使用） #

如果需要把汇总结果存入单独的汇总表，可根据业务需求选择以下两种自动更新方案：

方案1：定时任务批量更新（适合实时性要求不高的场景） #

可以用Django自定义命令+系统定时任务，或者Celery定时执行以下同步逻辑：

def sync_summary_data():
    # 拉取最新汇总数据
    summary_list = TableA.objects.annotate(
        total=Sum("b_items__c_items__number")
    ).values("name", "total")
    # 批量更新或创建汇总记录
    for item in summary_list:
        SummaryTable.objects.update_or_create(
            name=item["name"],
            defaults={"number": item["total"] or 0}
        )

方案2：信号触发实时更新（适合实时性要求高的场景） #

通过Django信号监听表C的增删改操作，只要数据变动就自动更新对应汇总记录：

# 信号定义（可单独放在signals.py文件中）
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.db.models import Sum

@receiver([post_save, post_delete], sender=TableC)
def update_summary_when_c_change(sender, instance, **kwargs):
    # 获取当前变动记录关联的表A实例
    a_instance = instance.code_b.code_a
    # 重新计算该维度的总和
    total = TableC.objects.filter(
        code_b__code_a=a_instance
    ).aggregate(total=Sum("number"))["total"] or 0
    # 更新汇总表
    SummaryTable.objects.update_or_create(
        name=a_instance.name,
        defaults={"number": total}
    )

信号需要在应用的apps.py中注册才能生效：

from django.apps import AppConfig

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = '你的应用名称'

    def ready(self):
        import 你的应用名称.signals # 导入存放信号的文件

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