Transactions & Factories

Transactions & Factories

Two features that share a page because they are both about controlling writes: transactions decide whether a group of writes happens at all, and factories generate the writes your tests need.

Transactions

A transaction makes several statements atomic — either every write lands or none does. Any operation that touches more than one row and would leave the database inconsistent if interrupted halfway belongs in one.

the basic form

from sillo.record import transaction

async with transaction():
    user = await User.create(email="a@b.com")
    await Profile.create(user=user, bio="Hello")
    await Order.create(user=user, total=99.99)

All three rows commit together on clean exit. If anything raises, the whole block rolls back and the exception propagates:

rollback on exception

from tortoise.expressions import F

try:
    async with transaction():
        await Account.filter(id=1).update(balance=F("balance") - 100)
        await Account.filter(id=2).update(balance=F("balance") + 100)
        raise RuntimeError("boom")
except RuntimeError:
    pass

# Neither update survives.

That is the whole contract, and it works. transaction() delegates to Tortoise’s connection._in_transaction(), which handles the driver-level protocol for each backend.

When to reach for this

Wrap a transaction around any write that must be all-or-nothing: a purchase that debits one account and credits another, a signup that creates a user plus a profile plus an audit row, a bulk import where a partial result is worse than no result.

Do not wrap read-only work in a transaction. It holds a connection and, on some isolation levels, takes locks, for no benefit.

Do not wrap slow external calls in one. A transaction that stays open while an HTTP request to a payment provider completes holds locks for the duration of that request, and a provider timeout becomes a database incident. Do the external call first, then open a short transaction to record the result.

Multiple connections

Pass connection_name to run against a non-default connection:

async with transaction(connection_name="analytics"):
    await Event.create(name="signup")

A transaction covers one connection. Writes to two different databases inside one async with are not atomic across both — that requires two-phase commit, which neither Tortoise nor sillo implements. If you need cross-database consistency, use an outbox table in the primary database and a worker that replays it.

Nesting

transaction() blocks nest as far as Tortoise supports them, and Tortoise’s _in_transaction() reuses the existing transaction when one is already open on that connection rather than starting a second. The practical implication is that an inner block committing does not make its writes durable — the outer block still decides. Do not write code that depends on an inner transaction being independently committed.

Savepoints are broken

TransactionContext.savepoint() raises TypeError on every backend

The implementation calls self._conn.execute_insert("SAVEPOINT sp"). execute_insert requires two arguments — a query and its values — so the call fails immediately:

TypeError: SqliteClient.execute_insert() missing 1 required positional
argument: 'values'

The exception handler then calls execute_insert("ROLLBACK TO SAVEPOINT sp"), which fails the same way, so the original error is replaced by a second TypeError during cleanup. There is no backend on which this works.

There is a second, independent design problem. The savepoint name is the literal string sp for every savepoint, so even with the call fixed, nesting two savepoints would make the inner RELEASE and ROLLBACK TO ambiguous.

And the intended semantics do not hold either: savepoint() re-raises after rolling back, so the exception propagates to the enclosing transaction() and rolls everything back — unless the caller catches it. The docstring’s promise that a failed inner block leaves the outer transaction intact requires a try/except the examples do not show.

If you need partial rollback, issue the SQL yourself with unique names:

from tortoise import connections

conn = connections.get("default")

async with transaction():
    await User.create(email="a@b.com")
    await conn.execute_query("SAVEPOINT sp_risky")
    try:
        await risky_operation()
        await conn.execute_query("RELEASE SAVEPOINT sp_risky")
    except Exception:
        await conn.execute_query("ROLLBACK TO SAVEPOINT sp_risky")
        # swallow deliberately: the outer transaction continues
    await Order.create(...)

Savepoints are supported by PostgreSQL, MySQL 5.0+, MariaDB, and — contrary to what is sometimes assumed — SQLite, which has had SAVEPOINT since 3.6.8.

Do not use begin / commit / rollback

The manual API does not compose with connection pooling

begin(), commit(), and rollback() issue raw execute_query("BEGIN"), "COMMIT", and "ROLLBACK" against connections.get(name).

They do not participate in Tortoise’s transaction context. Nothing records that a transaction is open, so a transaction() block entered afterwards has no idea, and Tortoise’s own bookkeeping is out of sync with the database’s.

On a pooled backend the problem is worse: each execute_query may be served by a different pooled connection. BEGIN on connection A followed by an INSERT on connection B followed by COMMIT on connection C commits nothing and leaves connection A holding an idle open transaction until it is recycled.

If a context manager genuinely does not fit your control flow, use AsyncExitStack to hold the transaction() context open across functions rather than reaching for these:

from contextlib import AsyncExitStack

async with AsyncExitStack() as stack:
    tx = await stack.enter_async_context(transaction())
    await step_one()
    await step_two()

Logging

transaction() catches, logs with logger.exception("Transaction rolled back"), and re-raises. Every rolled-back transaction therefore produces a full traceback in the sillo.record.transactions logger and whatever your application logs when it handles the exception. Expect duplicate tracebacks; if rollbacks are an expected part of your flow — optimistic concurrency retries, for instance — raise the level of that logger to avoid drowning your log aggregator.

A worked example

An idempotent order placement that keeps the external call outside the transaction.

ordering with an external charge

@app.post("/orders")
async def place_order(request, response):
    payload = request.validated_data

    # Outside the transaction: slow, and safe to retry.
    charge = await payments.charge(payload.token, payload.amount_cents)

    async with transaction():
        order = await Order.create(
            user_id=request.user.id,
            amount_cents=payload.amount_cents,
            charge_id=charge.id,
        )
        await OrderLine.bulk_create(
            [{"order_id": order.id, **line} for line in payload.lines]
        )
        await AuditEntry.create(action="order.created", subject_id=order.id)

    return response.json(order.to_dict(), status_code=201)

If the database write fails after a successful charge, the transaction rolls back and you are left with a charge and no order — which is why charge_id is stored and why a reconciliation job matching charges against orders is not optional in this design.

Factories

A factory generates model instances with sensible defaults so a test can say “give me a user” without restating ten required fields.

defining a factory

from uuid import uuid4
from sillo.record import Factory


class UserFactory(Factory):
    model = User
    definition = staticmethod(lambda: {
        "email": f"user-{uuid4().hex[:8]}@test.com",
        "name": "Test User",
        "plan": "free",
    })

using it

user = await UserFactory.create()                    # saved
admin = await UserFactory.create({"name": "Admin"})  # saved, with overrides
users = await UserFactory.create_many(5)             # five saved rows
draft = UserFactory.make()                           # not saved; draft.id is None

make() builds the instance without touching the database, which is what you want for a unit test of serialization or validation logic. create() calls instance.save(), so it goes through the normal write path — including ValidatesBeforeSaveMixin and any casts.

definition must not be a normal method

def definition(self) raises TypeError

Factory.make() calls cls.definition() with no arguments. A plain method declared on the class expects self:

class BadFactory(Factory):
    model = User
    def definition(self):          # WRONG
        return {"email": "a@b.com"}

BadFactory.make()
# TypeError: BadFactory.definition() missing 1 required positional argument: 'self'

Use a lambda or an explicit @staticmethod:

class GoodFactory(Factory):
    model = User

    @staticmethod
    def definition():
        return {"email": f"user-{uuid4().hex[:8]}@test.com"}

The staticmethod form is the one to prefer — it works, it reads normally, and it lets the body span more than one expression.

Uniqueness and create_many

create_many(count, overrides) applies the same overrides dict to every instance. That is fine for shared attributes and fatal for unique ones:

await UserFactory.create_many(5, {"email": "same@test.com"})
# IntegrityError on the second row

Generate unique values inside definition — a uuid4() fragment or a counter — so that every call produces a distinct row. If you need per-instance overrides, loop:

users = [await UserFactory.create({"email": f"u{i}@test.com"}) for i in range(5)]

create_many also inserts one row at a time, which is fine for a handful of test rows and slow for thousands. For bulk fixtures use Model.bulk_create with a list comprehension over Factory.make().

state() returns a function, not a chainable factory

Factory.state(**kwargs) returns a plain callable that produces {**cls.definition(), **kwargs}. There is no .state(...).create() chain — the returned function has to be installed as a definition:

what state() actually does

modifier = UserFactory.state(plan="enterprise")
modifier()          # {'email': '...', 'name': 'Test User', 'plan': 'enterprise'}

A subclass is clearer than trying to make state() into something it is not:

factory variants as subclasses

class AdminUserFactory(UserFactory):
    definition = staticmethod(lambda: {
        **UserFactory.definition(),
        "plan": "enterprise",
        "is_admin": True,
    })


admin = await AdminUserFactory.create()

FactoryBuilder

A dictionary with a nicer error message. register(name, factory) stores a factory class, get(name) retrieves it and raises KeyError: Factory 'x' not registered when absent.

builder = FactoryBuilder()
builder.register("user", UserFactory)
builder.register("post", PostFactory)

user = await builder.get("user").create()

It is worth using when fixtures are looked up by string — a seed-from-YAML script, say. For ordinary tests, importing the factory class is simpler.

Factories in tests

Use a transaction per test and roll it back, so tests do not have to clean up after themselves:

conftest.py

import pytest
from tortoise import Tortoise, connections


@pytest.fixture(scope="session", autouse=True)
async def db():
    await Tortoise.init(db_url="sqlite://:memory:",
                        modules={"models": ["myapp.models"]})
    await Tortoise.generate_schemas()
    yield
    await connections.close_all()


@pytest.fixture
async def user():
    return await UserFactory.create()

An in-memory SQLite database gives every session a clean schema in milliseconds. If your application depends on Postgres-specific behaviour — ON CONFLICT targets, array columns, ILIKE — test against Postgres instead; SQLite will pass tests that production fails.

What not to do

Do not use TransactionContext.savepoint(). It raises TypeError before doing anything.

Do not use begin() / commit() / rollback(). They bypass Tortoise’s transaction tracking and break under connection pooling.

Do not make external calls inside a transaction. Locks are held for the duration.

Do not assume an inner transaction() commits independently. The outermost block decides.

Do not expect atomicity across two connections. There is no two-phase commit.

Do not declare definition as def definition(self). Use a staticmethod or a lambda.

Do not pass a unique value in create_many overrides. Every row gets the same one.

Do not use create_many for thousands of rows. It is one insert per row.

Performance notes

A transaction holds a connection for its entire lifetime. With a pool of five, six concurrent long transactions starve every other request. Keep them short and measure the slowest statement inside them.

Row locks taken inside a transaction are held until commit. Two transactions updating the same rows in different orders deadlock; the database picks a victim and raises. Always touch rows in a consistent order — sorting ids before a bulk update is the usual fix.

create_many(1000) issues a thousand INSERT statements plus whatever each save() triggers. Model.bulk_create([...]) issues roughly 1000 / batch_size. For test fixtures the difference is seconds versus minutes.

API reference

Name	Signature	Notes
transaction	async with transaction(connection_name="default")	Commits on exit, rolls back on exception
TransactionContext.savepoint	async with tx.savepoint()	Raises TypeError; do not use
begin / commit / rollback	(connection_name="default")	Unsafe with pooling; do not use
Factory.make	(overrides=None) -> instance	Not saved
Factory.create	(overrides=None) -> instance	Saved via save()
Factory.create_many	(count, overrides=None) -> list	Same overrides for every row
Factory.state	(**kwargs) -> Callable	Returns a function, not a factory
FactoryBuilder	.register(name, factory), .get(name)	get raises KeyError

Related

• Record Overview — setup and connection lifecycle
• Models & Mixins — bulk_create, upsert, and validation hooks
• Scopes & Events — why lifecycle events do not fire inside create()
• Migrations & Seeding — seeders and fixtures for non-test data
• Exception Handlers & Pydantic — turning IntegrityError into a 409
• Concurrency — how sillo schedules the tasks these run in