Transactions¶

Group multiple operations into a single atomic unit with the ferro.transaction() async context manager: commit on clean exit, rollback if the block raises.

Basic Usage¶

        async with transaction():
            checking.balance -= 25
            await checking.save()

            savings.balance += 25
            await savings.save()
        # Both writes commit together when the block exits cleanly

Every ORM operation inside the block — create, save, delete, batch update, bulk_create, queries — runs on the same database connection (connection affinity is tracked through contextvars, so it is safe under concurrent asyncio tasks: each task gets its own transaction).

Rollback on Error¶

If any exception escapes the block, everything performed inside it is rolled back and the exception re-raises:

        try:
            async with transaction():
                checking.balance -= 1000
                await checking.save()
                raise RuntimeError("insufficient funds")
        except RuntimeError:
            pass

        await checking.refresh()
        # The failed write was rolled back
        assert checking.balance == 75

Using Named Connections¶

A transaction is pinned to one connection. Open it against a named connection with using=, and everything inside inherits that connection — unqualified operations included. Trying to route to a different connection inside the block raises:

    async with transaction(using="analytics"):
        await Metric.using("analytics").create(name="clicks", value=42)

Sessions + Transactions¶

transaction() composes with sessions. If a session is active, transactions inherit that session route by default:

import ferro

async with ferro.engines.session("analytics"):
    async with ferro.transaction():
        await Metric.create(name="requests", value=1)

You can also pass an explicit session handle (transaction(session=my_session)) when ambient context is different and you need a deterministic override.

Session.close() raises RuntimeError while session-scoped transactions are still open. Exit all transaction() blocks before closing the session.

Raw SQL Inside a Transaction¶

transaction() yields a Transaction handle exposing execute / fetch_all / fetch_one for raw SQL on the transaction's own connection — useful for Postgres GUCs (set_config, SET LOCAL), advisory locks, or any one-off statement that doesn't fit a model:

        async with transaction() as tx:
            await Account.create(owner="bob")
            # Raw SQL on the same connection, inside the same transaction
            rows = await tx.fetch_all("SELECT COUNT(*) AS n FROM account")

The handle becomes invalid when the block exits — calling it afterwards raises RuntimeError. If you don't need it, simply write async with transaction(): and the handle is discarded. The top-level ferro.execute / fetch_all / fetch_one functions also automatically join the active transaction — see Raw SQL.

Nesting Behavior¶

Nested transaction() blocks map to savepoints on the outer transaction's connection:

If an inner block raises, only its work is rolled back (to the savepoint); the outer transaction can continue and commit.
If the outer block raises, everything rolls back — including inner blocks that completed "successfully".

from ferro import transaction


async def import_rows(rows: list[dict]) -> int:
    imported = 0
    async with transaction():
        await AuditLog.create(event="import-started")
        for row in rows:
            try:
                async with transaction():  # savepoint per row
                    await Record.create(**row)
                    imported += 1
            except ValueError:
                continue  # this row rolled back; the import continues
    return imported

A nested block never commits independently of its parent — only the outermost block's clean exit commits to the database.

Patterns¶

Keep transactions short¶

A transaction holds a pooled connection (and database locks) for its entire duration. Do slow work — HTTP calls, file I/O, expensive computation — outside the block, and keep only the database writes inside:

async def publish(post_id: int) -> None:
    rendered = await render_markdown(post_id)  # slow work first, no tx held

    async with transaction():
        post = await Post.get(post_id)
        post.body_html = rendered
        post.published = True
        await post.save()

Error handling¶

Catch exceptions outside the block when the whole unit should roll back, and inside it only for work you genuinely want to keep partial (paired with a nested block, as above). Catching an exception inside the block and continuing means the surviving operations will commit:

async def transfer(src: Account, dst: Account, amount: int) -> bool:
    try:
        async with transaction():
            src.balance -= amount
            await src.save()
            if src.balance < 0:
                raise ValueError("insufficient funds")
            dst.balance += amount
            await dst.save()
    except ValueError:
        return False  # nothing was committed
    return True