def process_response(self, request, response):
"""
Disconnects the signal receiver to prevent it from staying active.
"""
if hasattr(threadlocal, 'actionslog'):
pre_save.disconnect(sender=LogAction, dispatch_uid=threadlocal.actionslog['signal_duid'])
return response
python类disconnect()的实例源码
def process_exception(self, request, exception):
"""
Disconnects the signal receiver to prevent it from staying active in case of an exception.
"""
if hasattr(threadlocal, 'actionslog'):
pre_save.disconnect(sender=LogAction, dispatch_uid=threadlocal.actionslog['signal_duid'])
return None
def unregister(model):
"""Unregister a model to the audit code.
:param model: Model to unregister.
:type model: object
"""
try:
pre_save.disconnect(_pre_save, sender=model, dispatch_uid=str(model))
post_save.disconnect(_post_save, sender=model, dispatch_uid=str(model))
pre_delete.disconnect(_pre_delete, sender=model, dispatch_uid=str(model))
except Exception as e:
logger.error("<Unregister> %s", e.message)
def register_pre_save_on_AUTH_USER_MODER_change(sender, setting, value, enter, **kwargs):
if setting == "AUTH_USER_MODEL" and value != USER_MODEL:
if enter:
pre_save.connect(useraudit.password_expiry.user_pre_save, sender=value)
else:
pre_save.disconnect(useraudit.password_expiry.user_pre_save, sender=value)
def test_signal(self):
def handler(sender, user=None, **kwargs):
self.handler_called = True
self.assertEquals(sender, type(self.user))
self.assertEquals(user, self.user)
login_failure_limit_reached.connect(handler)
self.handler_called = False
_ = authenticate(username=self.username, password="INCORRECT")
_ = authenticate(username=self.username, password="INCORRECT")
login_failure_limit_reached.disconnect(handler)
self.assertTrue(self.handler_called)
def job_model_saved(mocker):
from django.db.models.signals import pre_save
from django_remote_submission.models import Job
mock = mocker.Mock()
pre_save.connect(mock, sender=Job)
yield mock
pre_save.disconnect(mock, sender=Job)
def _disconnect_signals():
""" used in testing """
post_save.disconnect(plan_watchers.on_plan_save, TestPlan)
pre_delete.disconnect(plan_watchers.on_plan_delete, TestPlan)
pre_save.disconnect(plan_watchers.pre_save_clean, TestPlan)
def _disconnect_signals():
# used in testing
post_save.disconnect(case_watchers.on_case_save, TestCase)
post_delete.disconnect(case_watchers.on_case_delete, TestCase)
pre_save.disconnect(case_watchers.pre_save_clean, TestCase)