All or Nothing: Transactions

A library that writes two rows has to answer a question a library that writes one doesn’t: what happens if the second one fails?

“Half the rows are in” is never an acceptable answer, and it’s the default one.

The shape

@contextmanager
def transaction(connection):
    connection.execute("begin")
    try:
        yield connection
        connection.execute("commit")     # inside the try: a commit can fail too
    except BaseException:
        connection.execute("rollback")
        raise

Three things in that sketch are deliberate.

except BaseException, not except Exception. KeyboardInterrupt and SystemExit don’t inherit from Exception, and a half-written transaction is exactly as bad when someone pressed Ctrl-C. Catching BaseException is usually wrong; in cleanup code that re-raises immediately, it’s right.

raise at the end of the handler. The context manager decides what the database does, not what the caller sees. Swallowing the exception here would turn a failed write into a silent no-op.

The commit is inside the try, not after it. It’s tempting to put it in an else, so it only runs when the body succeeded. But a commit is when a database checks deferred constraints, so commit itself can raise — and when it does, the except has to roll back, exactly as for a failure in the body. In an else a failed commit would escape with the transaction still open. The checkpoint has a test that forces this, and it’s why the finished library’s commit lives inside the handler.

import sqlite3
from contextlib import contextmanager

connection = sqlite3.connect(":memory:", isolation_level=None)
connection.execute("create table post (id integer primary key, title text unique)")


@contextmanager
def transaction(connection):
    connection.execute("begin")
    try:
        yield connection
        connection.execute("commit")
    except BaseException:
        connection.execute("rollback")
        raise


def titles():
    return [row[0] for row in connection.execute("select title from post")]


with transaction(connection):
    connection.execute("insert into post (title) values ('first')")
print("after a successful transaction:", titles())

try:
    with transaction(connection):
        connection.execute("insert into post (title) values ('second')")
        connection.execute("insert into post (title) values ('first')")   # duplicate
except sqlite3.IntegrityError as error:
    print("IntegrityError:", error)

print("after a failed transaction:    ", titles())

'second' was inserted successfully and is gone anyway. That’s the guarantee: the unit of work is the with block, not the statement.

Nesting, and why it can’t just be ignored

save() wraps its own writes in a transaction. save_all() wants to wrap several saves in one. Naively, the inner begin fails — SQLite has no nested transactions.

What it does have is savepoints: named markers you can roll back to without abandoning the outer transaction.

    depth = _depth.get(connection, 0)
    savepoint = f"orm_{depth}"
    connection.execute("begin" if depth == 0 else f"savepoint {savepoint}")

On the way out, the outermost block commits and an inner one releases its savepoint — which doesn’t commit anything, it just says “these changes now belong to the enclosing block”.

import sqlite3
from contextlib import contextmanager

connection = sqlite3.connect(":memory:", isolation_level=None)
connection.execute("create table post (id integer primary key, title text)")
depths = {}


@contextmanager
def transaction(connection):
    depth = depths.get(connection, 0)
    savepoint = f"orm_{depth}"
    connection.execute("begin" if depth == 0 else f"savepoint {savepoint}")
    depths[connection] = depth + 1
    try:
        yield connection
        connection.execute("commit" if depth == 0 else f"release {savepoint}")
    except BaseException:
        connection.execute("rollback" if depth == 0 else f"rollback to {savepoint}")
        if depth:
            connection.execute(f"release {savepoint}")
        raise
    finally:
        if depth:
            depths[connection] = depth
        else:
            del depths[connection]


with transaction(connection):
    connection.execute("insert into post (title) values ('outer')")
    try:
        with transaction(connection):
            connection.execute("insert into post (title) values ('inner')")
            raise RuntimeError("the inner work failed")
    except RuntimeError as error:
        print("handled:", error)

print("kept:", [row[0] for row in connection.execute("select title from post")])
print("nothing left behind:", depths)

The outer row survives, the inner one doesn’t, and the outer transaction committed normally. That’s the behaviour a caller expects from nesting, and getting it takes about six lines.

Where the depth lives

The obvious place to keep “how deep are we on this connection” is on the connection:

connection._orm_depth = depth + 1

It doesn’t work. sqlite3.Connection is a C type with no __dict__, and the usual fallback fails too:

import sqlite3
import weakref

connection = sqlite3.connect(":memory:")

try:
    connection._orm_depth = 1
except AttributeError as error:
    print("attribute: AttributeError:", error)

try:
    weakref.WeakKeyDictionary()[connection] = 1
except TypeError as error:
    print("weakref:   TypeError:", error)

print("as a dict key:", {connection: 1}[connection])

So the library keeps a plain dict keyed by the connection — and because a plain dict holds a strong reference, the entry is deleted when the outermost transaction ends:

    finally:
        if depth:
            _depth[connection] = depth
        else:
            del _depth[connection]        # nothing is kept between transactions

Without that del, every connection the library has ever seen stays alive for the life of the process. It’s the same leak as the identity map, in the place you’d least look for it, and a test pins it:

    def test_nothing_is_left_behind_between_transactions(self) -> None:
        Post(title="first", views=1).save(self.connection)
        self.assertEqual(transactions._depth, {}, "a connection stayed in the depth map")

import os
import tempfile
import unittest
from contextlib import contextmanager
from pathlib import Path


@contextmanager
def working_copy(path):
    """Copy `path` to `path.orig`, yield the original path to be edited, and:

    - on success, leave the edited file in place and remove the backup
    - on any exception, restore the file from the backup, then remove it

    TODO: write it.
    """
    raise NotImplementedError("working_copy")
    yield path


class WorkingCopy(unittest.TestCase):
    def setUp(self):
        self.folder = tempfile.TemporaryDirectory()
        self.addCleanup(self.folder.cleanup)
        self.path = Path(self.folder.name) / "notes.txt"
        self.path.write_text("original\n")

    def leftovers(self):
        return sorted(p.name for p in Path(self.folder.name).iterdir())

    def test_edits_are_kept_on_success(self):
        with working_copy(self.path) as path:
            path.write_text("edited\n")
        self.assertEqual(self.path.read_text(), "edited\n")
        self.assertEqual(self.leftovers(), ["notes.txt"])

    def test_edits_are_undone_on_failure(self):
        with self.assertRaises(RuntimeError):
            with working_copy(self.path) as path:
                path.write_text("half an edit\n")
                raise RuntimeError("no good")
        self.assertEqual(self.path.read_text(), "original\n")
        self.assertEqual(self.leftovers(), ["notes.txt"])

    def test_the_exception_still_reaches_the_caller(self):
        with self.assertRaises(KeyboardInterrupt):
            with working_copy(self.path) as path:
                path.write_text("interrupted\n")
                raise KeyboardInterrupt
        self.assertEqual(self.path.read_text(), "original\n")


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(WorkingCopy)
)

@contextmanager
def working_copy(path):
    backup = path.with_suffix(path.suffix + ".orig")
    backup.write_bytes(path.read_bytes())
    try:
        yield path
    except BaseException:
        path.write_bytes(backup.read_bytes())     # put the original back
        raise
    finally:
        backup.unlink(missing_ok=True)            # either way, no backup is left

What you learned

• A transaction is a context manager: commit in else, roll back on BaseException, always re-raise.
• Savepoints make nesting work, so an inner block can fail without discarding the outer one’s work.
• sqlite3 needs isolation_level=None (or autocommit=True) before the library can control transactions.
• State keyed by a caller’s object must be removed when you’re done with it, and that deserves its own test.