Transactions and sessions¶

This guide shows direct transactions, savepoints, and the session unit-of-work helper.

Use it when several writes must succeed or fail together.

What the example covers¶

async with db.transaction()
async with db.savepoint("name")
async with db.session()
staged inserts and session savepoints
commit-on-success and rollback-on-error behavior

"""Transactions, savepoints, and sessions example."""

import asyncio

from pydantic import BaseModel

from ormdantic import Ormdantic

db = Ormdantic("sqlite:///examples_transactions_sessions.sqlite3")


@db.table(pk="id")
class Flavor(BaseModel):
    id: str
    name: str


async def main() -> None:
    await db.init()
    await db.drop_all()
    await db.create_all()

    async with db.transaction():
        await db[Flavor].insert(Flavor(id="1", name="mocha"))
        try:
            async with db.savepoint("optional_insert"):
                await db[Flavor].insert(Flavor(id="2", name="rollback"))
                raise RuntimeError("rollback only the savepoint")
        except RuntimeError:
            pass

    async with db.session() as session:
        try:
            async with session.savepoint("optional_session_insert"):
                session.add(Flavor(id="2", name="discarded"))
                await session.flush()
                raise RuntimeError("rollback only the session savepoint")
        except RuntimeError:
            pass
        session.add(Flavor(id="3", name="vanilla"))

    assert await db[Flavor].count() == 2


if __name__ == "__main__":
    asyncio.run(main())

Run it locally:

python examples/transactions_sessions.py

Choose the right API¶

Use direct transactions when you already know each operation. Use sessions when you want to stage model objects and let Ormdantic flush them in dependency order.