如何在Django ORM中追踪事务变更及特定事务的修改行？

嘿，这两个问题在业务审计、变更追踪场景里特别常见，我来给你梳理下纯Django运行时就能搞定的方案，完全不需要依赖外部CDC框架：

当然可以，核心思路是结合Django的事务上下文+信号机制+自定义事务ID，把每个事务的操作都绑定到一个唯一标识上：

• 用线程本地存储（threading.local）在当前事务的执行线程中传递唯一事务ID，让事务内的所有ORM操作都能共享这个标识
• 借助Django的post_save、post_delete信号，在每次模型实例变更时，把变更记录和当前事务ID关联起来
• 记得用transaction.on_commit()确保变更日志在事务提交后才保存，避免事务回滚时留下无效记录

基于上面的事务ID机制，我们可以直接把每个变更行的信息（模型名、实例ID、操作类型、字段变化）和事务ID绑定，之后就能通过事务ID精准查询该事务下的所有修改行：

具体实现示例 #

第一步：定义线程本地存储和事务ID生成函数

import threading
from uuid import uuid4
from django.db import transaction

# 线程本地存储，用来在当前线程的事务内传递事务ID
transaction_local = threading.local()

def start_tracked_transaction():
    """启动一个带追踪的事务，返回唯一事务ID"""
    tx_id = str(uuid4())
    transaction_local.transaction_id = tx_id
    return tx_id

第二步：创建变更日志模型（存储事务变更记录）

from django.db import models

class ChangeLog(models.Model):
    transaction_id = models.CharField(max_length=36, db_index=True)
    model_name = models.CharField(max_length=100)
    instance_id = models.CharField(max_length=100)  # 兼容不同类型的主键
    operation = models.CharField(max_length=10, choices=[
        ('CREATE', '新增'),
        ('UPDATE', '修改'),
        ('DELETE', '删除')
    ])
    changed_fields = models.JSONField(null=True)  # 存储修改的字段和新旧值
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        indexes = [
            models.Index(fields=['transaction_id', 'model_name']),
        ]

第三步：编写信号处理器，关联变更和事务ID

如果需要记录具体的字段变化，可以用django-dirtyfields包追踪实例的脏字段：

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from dirtyfields import DirtyFieldsMixin  # 需要先安装django-dirtyfields

# 你的模型需要继承DirtyFieldsMixin来追踪字段变化
# class MyModel(DirtyFieldsMixin, models.Model):
#     ...

@receiver(post_save)
def log_save_change(sender, instance, created, **kwargs):
    # 只追踪带有事务ID的操作
    if not hasattr(transaction_local, 'transaction_id'):
        return
    
    changed_fields = None
    if not created:
        # 获取修改的字段和新旧值
        dirty_fields = instance.get_dirty_fields(check_relationship=True)
        changed_fields = {
            field: {'old': dirty_fields[field], 'new': getattr(instance, field)}
            for field in dirty_fields
        }
    
    # 确保日志在事务提交后再保存，避免回滚时产生无效日志
    transaction.on_commit(lambda: ChangeLog.objects.create(
        transaction_id=transaction_local.transaction_id,
        model_name=sender.__name__,
        instance_id=str(instance.pk),
        operation='CREATE' if created else 'UPDATE',
        changed_fields=changed_fields
    ))

@receiver(post_delete)
def log_delete_change(sender, instance, **kwargs):
    if not hasattr(transaction_local, 'transaction_id'):
        return
    
    transaction.on_commit(lambda: ChangeLog.objects.create(
        transaction_id=transaction_local.transaction_id,
        model_name=sender.__name__,
        instance_id=str(instance.pk),
        operation='DELETE'
    ))

第四步：在业务代码中使用带追踪的事务

def process_order(request):
    with transaction.atomic():
        tx_id = start_tracked_transaction()
        # 执行你的ORM操作
        order = Order.objects.get(pk=request.POST['order_id'])
        order.status = 'completed'
        order.save()
        
        # 关联的操作，比如更新库存
        stock = Stock.objects.get(product=order.product)
        stock.quantity -= order.quantity
        stock.save()
    
    # 之后可以用tx_id查询这个事务的所有变更
    transaction_changes = ChangeLog.objects.filter(transaction_id=tx_id)
    # 输出变更：比如Order的status修改，Stock的quantity修改
    for change in transaction_changes:
        print(f"{change.model_name}({change.instance_id}) - {change.operation}: {change.changed_fields}")

注意事项 #

• 异步环境适配：如果用的是Async Django，需要把threading.local换成asyncio.local()，确保事务ID在异步任务中正确传递
• 性能考量：高并发场景下，要注意ChangeLog表的写入性能，可以考虑批量插入或者用异步任务处理日志写入（但要确保事务提交后再执行）
• 敏感字段过滤：模型中有敏感字段（比如密码）时，要在信号处理器中过滤掉，不要写到日志里

内容的提问来源于stack exchange，提问作者Yaniv Peer