Django 数据系统设计原则

背景#

在把 Excel + Streamlit 的财务系统重写成 Django 全栈平台的过程中，踩了不少坑，也沉淀了一些判断。

这篇不是 Django 教程，而是在真实项目里反复验证的设计决策清单——哪些地方值得多想一步，哪些地方偷懒会还债。

原则一：分层要彻底，每层只做自己的事#

Django 项目很容易退化成"把所有逻辑塞进 View"。一旦这样，代码就很难测试和复用。

按职责分四层：

Model     →  管数据结构与字段定义
Manager   →  管查询逻辑（聚合、过滤、排序）
Service   →  管业务聚合（跨 Model 的复杂逻辑）
View      →  只管 HTTP 请求/响应，调用 Service

反模式：View 里直接写 .annotate().filter().values() 的长链查询，下次改需求时找都找不到。

正确做法：把分析查询封装成 Manager 方法，语义化命名：

# ❌ 不要这样
def dashboard(request):
    data = Invoice.objects.filter(year=2024)\
        .values('month').annotate(total=Sum('total_amount'))...

# ✅ 应该这样
def dashboard(request):
    data = Invoice.analytics.monthly_summary(year=2024)

原则二：ETL 必须有状态，每次导入都要留痕#

数据分析系统的上游是数据导入，而数据导入是最容易出错的环节。

每次 ETL 任务都应该持久化一条记录，至少包含：

任务状态（pending / running / success / partial / failed）
总条数、成功条数、失败条数
完整的错误日志（JSONField，记录哪行、什么错）
开始时间和结束时间

没有任务日志的后果：某天数据对不上，你根本不知道是哪批导入出了问题，只能重跑全量——这在财务场景下很危险。

class ETLTask(BaseModel):
    status = models.CharField(max_length=20, choices=STATUS_CHOICES)
    total_records = models.IntegerField(default=0)
    success_count = models.IntegerField(default=0)
    failed_count = models.IntegerField(default=0)
    error_log = models.JSONField(default=list)   # 每条错误都记下来
    started_at = models.DateTimeField(null=True)
    completed_at = models.DateTimeField(null=True)

原则三：缓存要主动失效，不能只靠 TTL#

分析查询结果缓存在 Redis 可以极大提升响应速度，但缓存管理不当会导致数据不一致。

常见的错误做法：设一个 1 小时的 TTL，然后祈祷数据不会在这 1 小时内变化。

正确做法是主动失效：当发票数据更新时，通过 Django Signal 精确清除相关缓存。

# 发票变更 → 清除该年份所有分析缓存
@receiver([post_save, post_delete], sender=Invoice)
def invalidate_analytics_cache(sender, instance, **kwargs):
    pattern = f"dashboard:{instance.year}:*"
    cache.delete_pattern(pattern)  # django-redis 支持通配符

TTL 作为兜底机制保留，主动失效作为主要机制——两者配合才可靠。

原则四：索引和查询对齐，建模时就要想好#

索引不是建了就行，复合索引的字段顺序和你实际查询的过滤条件顺序必须一致，否则不走索引。

分析型系统的典型查询模式：

# 最常见：按年+月+方向过滤
Invoice.objects.filter(year=2024, month=6, direction='outbound')

# 按公司+时间范围
Invoice.objects.filter(seller=company, invoice_date__range=(start, end))

对应的索引设计：

class Meta:
    indexes = [
        # 字段顺序和查询过滤顺序一致
        models.Index(fields=['year', 'month', 'direction']),
        models.Index(fields=['seller', 'invoice_date']),
    ]

事后加索引的代价：生产环境大表加索引会锁表，加之前还要 EXPLAIN 验证。建模时就规划好，成本最低。

原则五：财务金额永远用 DecimalField#

这条没有商量余地。

FloatField 在 Python 底层是 IEEE 754 双精度浮点，做乘法和加法会有精度误差：

>>> 0.1 + 0.2
0.30000000000000004

财务对账时，哪怕差 0.01 元也要查原因。用 FloatField 存金额，等于给自己埋了定时炸弹。

# ❌
amount = models.FloatField()

# ✅
amount = models.DecimalField(max_digits=14, decimal_places=2)

连带地，Pandas 处理后写回数据库前也要先转 Decimal，不要直接用 float 赋值给 DecimalField。

原则六：时间维度做冗余字段，别在查询里算#

分析系统高频的操作是"按年/按月/按季度聚合"。如果每次都在 SQL 里用 YEAR(invoice_date) 这类函数提取，有两个问题：

函数列无法走普通索引
每次聚合都要重新计算

更好的做法是在模型的 save() 里预计算，以空间换时间：

def save(self, *args, **kwargs):
    if self.invoice_date:
        self.year = self.invoice_date.year
        self.month = self.invoice_date.month
        self.quarter = (self.invoice_date.month - 1) // 3 + 1
    super().save(*args, **kwargs)

这样 year、month、quarter 都是普通整数列，可以建索引，聚合查询性能提升明显。

决策速查表#

场景	推荐做法	避免做法
复杂聚合查询	封装到自定义 Manager	写在 View 里
数据导入	ETLTask 记录任务状态	直接跑脚本不留日志
分析结果缓存	Signal 主动失效 + TTL 兜底	只靠 TTL
大表索引规划	建模时对齐查询模式	等慢了再加
金额字段	DecimalField	FloatField
时间维度聚合	预计算冗余字段	查询时用日期函数
耗时操作	Celery 异步任务	在请求里同步执行

背景#