面试题：Python Django ORM自定义数据库操作与事务管理深度剖析

Question

在Django项目中，由于业务需求，需要对ORM进行扩展，实现一种自定义的数据库操作，用于在更新某条记录时，同时记录更新日志到另一个日志表中，并且这两个操作要在一个事务中保证原子性。请阐述实现这个功能的详细思路，并给出核心代码示例。说明在高并发场景下，如何确保事务的正确性和数据一致性。

Answer 1

实现思路

1. 定义日志模型：创建一个用于记录更新日志的模型，包含与被更新记录相关联的字段以及记录更新时间、更新内容等信息。
2. 自定义ORM方法：在Django的模型管理器（Manager）中自定义一个方法，该方法负责执行更新操作并同时记录日志。
3. 使用事务：利用Django提供的事务管理机制，确保更新操作和日志记录操作在同一个事务中，保证原子性。
4. 高并发处理：使用数据库锁机制，例如悲观锁或乐观锁，来确保在高并发场景下事务的正确性和数据一致性。

核心代码示例

1. 定义日志模型

from django.db import models

class UpdateLog(models.Model):
    related_model_id = models.IntegerField()
    update_time = models.DateTimeField(auto_now_add=True)
    update_content = models.TextField()

    def __str__(self):
        return f"Log for {self.related_model_id} at {self.update_time}"

2. 自定义ORM方法

from django.db import transaction
from django.db.models import Manager

class CustomModelManager(Manager):
    @transaction.atomic
    def custom_update_with_log(self, pk, **kwargs):
        instance = self.get(pk=pk)
        old_values = {field.name: getattr(instance, field.name) for field in instance._meta.fields}
        for key, value in kwargs.items():
            setattr(instance, key, value)
        instance.save()

        new_values = {field.name: getattr(instance, field.name) for field in instance._meta.fields}
        update_content = f"Old: {old_values}, New: {new_values}"
        UpdateLog.objects.create(
            related_model_id=pk,
            update_content=update_content
        )
        return instance

3. 在模型中使用自定义管理器

class YourMainModel(models.Model):
    # 定义你的模型字段
    objects = CustomModelManager()

4. 调用自定义方法

# 在视图或其他业务逻辑中调用
your_model = YourMainModel.objects.custom_update_with_log(pk=1, some_field='new value')

高并发场景下的处理

1. 悲观锁：在获取需要更新的记录时，使用select_for_update方法获取锁。

class CustomModelManager(Manager):
    @transaction.atomic
    def custom_update_with_log(self, pk, **kwargs):
        with transaction.atomic():
            instance = self.select_for_update().get(pk=pk)
            # 后续操作同之前的代码

2. 乐观锁：在模型中添加一个版本字段，每次更新时检查版本号，更新成功后增加版本号。

class YourMainModel(models.Model):
    version = models.PositiveIntegerField(default=0)
    # 其他字段

    objects = CustomModelManager()

class CustomModelManager(Manager):
    @transaction.atomic
    def custom_update_with_log(self, pk, **kwargs):
        instance = self.get(pk=pk)
        original_version = instance.version
        # 检查版本号
        if not self.filter(pk=pk, version=original_version).update(**kwargs, version=F('version') + 1):
            raise Exception("Concurrent update detected")
        new_instance = self.get(pk=pk)
        # 记录日志操作

这样可以在高并发场景下确保事务的正确性和数据一致性。