我有一个包含以下两个模型的 Django 项目:
# models.py
from django.db import models, transaction
class Person(models.Model):
name = models.TextField()
surname = models.TextField()
class EmployeeInfo(models.Model):
person = models.OneToOneField(Person, on_delete=models.CASCADE)
employee_id = models.TextField(null=True, blank=True)
@transaction.atomic
def provision_employee_id(self):
"""Make atomic to ensure that two employees being provisioned at the same time do not get
the same employee number"""
self.employee_id = new_employee_number()
self.save()
raise Exception("forcing an exception") # Always raise an exception (for testing only)
这是失败的单元测试:
# tests.py
class PersonTestCase(TestCase):
def test_employee_id_atomic(self):
person1 = Person(name="John", surname="Doe")
employee_info = EmployeeInfo(person=person1)
self.assertIsNone(employee_info.employee_id) # No employee_id yet.
with self.assertRaises(Exception) as context:
cluster_updated = employee_info.provision_employee_id()
self.assertIsNone(employee_info.employee_id) # This FAILS
换句话说,即使我将
provision_employee_id()
包装在原子事务中,当引发后续异常时,save()
也不会回滚。为什么?
这与数据库无关。如果数据库发生故障,它不会回滚 Python 函数,它只会回滚事务中的所有数据库查询。 如果您这样从数据库中获取
Employee
对象,它仍然是
NULL
。如果您在函数中进行了两次数据库更改,则这些更改都会回滚。 如果你再次从数据库中获取数据,使用
[Django-doc],你会发现它确实仍然是
NULL
:# tests.py
class PersonTestCase(TestCase):
def test_employee_id_atomic(self):
person1 = Person.objects.create(name='John', surname='Doe')
employee_info = EmployeeInfo.objects.create(person=person1)
self.assertIsNone(employee_info.employee_id) # No employee_id yet.
with self.assertRaises(Exception) as context:
cluster_updated = employee_info.provision_employee_id()
employee_info.refresh_from_db()
self.assertIsNone(employee_info.employee_id)