Skip to content
Create a gist now

Instantly share code, notes, and snippets.

Embed URL


Subversion checkout URL

You can clone with
Download ZIP
# post_commit and post_rollback transaction signals for Django with monkey patching
# Author Grégoire Cachet <>
# Usage:
# You have to make sure to load this before you use signals.
# For example, create utils/ and then utils/ contening
# this gist in you project. Then add "import utils.transaction" in your project
# file
# Then, to use the signals, create a function and bind it to the post_commit
# signal:
# from django.db import transaction
# def my_function(**kwargs):
# # do your stuff here
# pass
# transaction.signals.post_commit.connect(my_function)
# If you're using non-local variables in your callback function, make sure to
# use non-weak reference or your variables could be garbarge collected before
# the function gets called. For example, in a model save() method:
# def save(self, *args, **kwargs):
# def my_function(**kwargs):
# # do your stuff here
# # access self variable
# self
# transaction.signals.post_commit.connect(my_function, weak=False)
from django.db import transaction
from django.dispatch import Signal
import thread
except ImportError:
import dummy_thread as thread
class ThreadSignals(object):
def __init__(self):
self.post_commit = Signal()
self.post_rollback = Signal()
class TransactionSignals(object):
signals = {}
def _has_signals(self):
thread_ident = thread.get_ident()
return thread_ident in self.signals
def _init_signals(self):
thread_ident = thread.get_ident()
assert thread_ident not in self.signals
self.signals[thread_ident] = ThreadSignals()
return self.signals[thread_ident]
def _remove_signals(self):
thread_ident = thread.get_ident()
assert thread_ident in self.signals
del self.signals[thread_ident]
def _get_signals(self):
thread_ident = thread.get_ident()
assert thread_ident in self.signals
return self.signals[thread_ident]
def _get_or_init_signals(self):
if self._has_signals():
return self._get_signals()
return self._init_signals()
def _send_post_commit(self):
if self._has_signals():
_signals = self._get_signals()
def _send_post_rollback(self):
if self._has_signals():
_signals = self._get_signals()
def post_commit(self):
return self._get_or_init_signals().post_commit
def post_rollback(self):
return self._get_or_init_signals().post_rollback
transaction.signals = TransactionSignals()
# monkey patching
old_managed = transaction.managed
def managed(flag=True):
to_commit = False
if not flag and transaction.is_dirty():
to_commit = True
if to_commit:
transaction.managed = managed
old_commit_unless_managed = transaction.commit_unless_managed
def commit_unless_managed(*args, **kwargs):
old_commit_unless_managed(*args, **kwargs)
if not transaction.is_managed():
transaction.commit_unless_managed = commit_unless_managed
old_rollback_unless_managed = transaction.rollback_unless_managed
def rollback_unless_managed(*args, **kwargs):
old_rollback_unless_managed(*args, **kwargs)
if not transaction.is_managed():
transaction.rollback_unless_managed = rollback_unless_managed
# If post_commit or post_rollback signals set the transaction to dirty state
# they must commit or rollback by themselves
old_commit = transaction.commit
def commit(*args, **kwargs):
old_commit(*args, **kwargs)
transaction.commit = commit
old_rollback = transaction.rollback
def rollback(*args, **kwargs):
old_rollback(*args, **kwargs)
transaction.rollback = rollback

Hi, I am trying to use you monkey patch and got the following error:

Django Version: 1.2.3
Exception Type: TypeError
Exception Value:

managed() got an unexpected keyword argument 'using'

Exception Location: /opt/local/Library/Frameworks/Python.framework/Versions/2.5/lib/python2.5/site-packages/django/db/ in _commit_on_success, line 297
Python Executable: /opt/local/Library/Frameworks/Python.framework/Versions/2.5/Resources/
Python Version: 2.5.4

any idea?

Thanks in advance


Use Django 1.3

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.