仅在事务完成后触发 post_save 信号

Posted 2023-02-23

【中文标题】仅在事务完成后触发 post_save 信号

【英文标题】：Trigering post_save signal only after transaction has completed

【发布时间】：2016-01-15 19:28:36

【问题描述】：

我已经编写了一些 API，在事务块中执行相应的函数。我在一个/多个模型/s 的实例/s 上调用save() 方法（经过一些修改），并且还在Elasticsearch 中连续索引该实例/s 的一些JSON 相关信息。我希望数据库回滚，即使出于某种原因，其中一个实例的 save() 或对 Elasticsearch 的索引失败。

现在，问题出现了，即使在事务块内，post_save() 信号也会被调用，这是一个问题，因为某些通知是从这些信号触发的。

有没有办法在事务成功完成后才触发post_save() 信号？

【参考方案1】：

不是真的。这些信号与 db 事务成功或失败无关，但与 save 方法本身有关 - 在调用之前您触发了 pre_save 信号，在调用之后您触发了 post_save 信号。

这里有两种方法：

您将在 post_save 方法中检查实例并确定模型是否保存成功；最简单的方法是：在 save 方法中，在事务成功执行后，使用标志注释您的实例，例如 instance.saved_successfully = True，您将在 post_save 处理程序中对其进行测试。 您将放弃 post_save 信号并为自己创建一个自定义信号，您将在事务成功运行后触发该信号。

有道理吗？

附言

如果您严格需要绑定到事务提交信号，请查看此包：https://django-transaction-hooks.readthedocs.org/en/latest/;看起来该功能已集成在 Django 1.9a 中。

【讨论】：

谢谢，我已经尝试过第一种方法，我设置了instance.in_transaction = True，但这确实造成了很多混乱，因为许多其他开发人员随后错过了对信号的处理。我认为，第二种方法会更容易处理，并且并发症最少。

信号包含在事务***.com/questions/36331753/…

【参考方案2】：

当 django 的管理员在修改内联子对象时不允许对父对象进行 post_save 事务，我遇到了严重问题。

这是我抱怨在原子块中间进行查询的错误的解决方案：

def on_user_post_save_impl(user):
     do_something_to_the_user(user)

def on_user_post_save(sender, instance, **kwargs):
    if not transaction.get_connection().in_atomic_block:
        on_user_post_save_impl(instance)
    else:
        transaction.on_commit(lambda: on_user_post_save_impl(instance))

【讨论】：

这不起作用 - AttributeError: "'module' object has no attribute 'on_commit'"

你从django导入事务了吗？我认为是 django.db.transaction

【参考方案3】：

我认为最简单的方法是使用transaction.on_commit()。这是一个使用 models.Model 子类 Photo 的示例，它只会在当前事务结束后与 Elasticsearch 对话：

from django.db import transaction
from django.db.models.signals import post_save

@receiver(post_save, sender=Photo)
def save_photo(**kwargs):
    transaction.on_commit(lambda: talk_to_elasticsearch(kwargs['instance']))

请注意，如果transaction.on_commit() 在未处于活动事务中时被执行，它将立即运行。

【讨论】：

感谢分享。但是，我想知道您是如何使用 django 的测试 API 来测试信号的。测试用例包装在始终回滚的事务中，导致信号函数永远不会执行并且测试失败。

@Scratch'N'Purr 如果你使用Django的基于unittest的框架，你可以使用TransactionTestCase，如果你使用pytest，你可以使用transactional_db fixture在pytest-django package中找到！

【参考方案4】：

我们正在使用这个小块：

def atomic_post_save(sender, instance, **kwargs):
    if hasattr(instance, "atomic_post_save") and transaction.get_connection().in_atomic_block:
        transaction.on_commit(lambda: instance.atomic_post_save(sender, instance=instance, **kwargs))

post_save.connect(atomic_post_save)

然后我们只需在我们喜欢的任何模型上定义一个atomic_post_save 方法：

class MyModel(Model):
    def atomic_post_save(self, sender, created, **kwargs):
        talk_to_elasticsearch(self)

有两点需要注意：

我们只在事务中调用atomic_post_save。 发送消息并将其包含在来自atomic_post_save 内部的当前请求中为时已晚。