How-To: Undo a transaction with the ZODB

Suppose you’ve written a script to “fix something real quick” and unleashed it upon your live database. Five minutes later, you discover your script had a bug, and now you’ve wrecked quite a bit of production data. Ouch.

You might be lucky, though, since the ZODB offers transaction-level undo. This comes with a lot of caveats, though, the biggest being that if something else was changed in the meantime that causes the undo to conflict, it won’t work. (Before transaction X, some value was A which X changed to B, but later something changes it to C. If I now want to undo transaction X to get back to A, it will conflict. Catalogs and other shared state are prime candidates).

But you still might be lucky and there won’t be a conflict. So, how do you undo a transaction? First, you need to find the transaction. In my case, I knew an object that had been changed by my script. So I asked the ZODB for the history of that object, i.e. the last transaction(s) that changed it:

>>> db = root._p_jar.db()
>>> hist = db.history(my_changed_object._p_oid)
>>> hist
[{'tid': '\x03...', 'size': 123, 'user_name': '', 'description': '', 'time': 1304493667.320477}]

Now I have the offending transaction’s ID. However, the undo() API does not work with transaction ids but needs a special (storage-specific) identifier. And, since as far as I can tell there is no way to map a transaction id to an “undo id”, I had to make to by matching the time stamp:

>>> info = db.undoInfo(specification=dict(time=hist['time']))
>>> info
[{'id': 'A44XmR876bs=', 'time': 1304493667.320477, 'user_name': '', 'description': '', 'size': 315893}]

Finally, call undo and hope you don’t get a conflict upon committing:

>>> db.undo(info['id'])
>>> import transaction
>>> transaction.commit()

Testing pyramid applications with zope.testbrowser

The Pyramid documentation recommends using WebTest for functional tests, but coming from the Zope world, zope.testbrowser is the more familiar tool for this job (and it seems a little more high-level than WebTest at first glance).

With the 3.11 release,  zope.testbrowser gained out-of-the-box support for talking to WSGI applications (courtesy of wsgi_intercept), so herewith some corresponding example code to show the basic usage and setup. (This assumes you are using zope.testrunner or something else that supports its layers).

import unittest
import zope.testbrowser.wsgi

class MyAppLayer(zope.testbrowser.wsgi.Layer):

    def make_wsgi_app(self):
        from myapp import main
        return main({})

MY_APP_LAYER = MyAppLayer()

class FunctionalTests(unittest.TestCase):

    layer = MY_APP_LAYER

    def test_root(self):
        browser = zope.testbrowser.wsgi.Browser()
        browser.open('http://localhost/')
        self.failUnless('Pyramid' in browser.contents)