Guide to Database Transactions in Python with Mitsuki

What a database transaction is, why a service method that writes twice needs one, and how @Transactional gives you one in Mitsuki.

A database transaction groups several statements into one unit of work that either happens completely or not at all.

It's the typical failsafe that prevents a bank transfer from debiting one account without crediting the other, an order from being saved without its line items, or a sign-up from creating a user without their settings.

Mitsuki has @Transactional just like Spring, Django has transaction.atomic(), and with plain SQLAlchemy (and therefore FastAPI) you open one yourself with session.begin().

This guide explains what a transaction does, and then uses Mitsuki's @Transactional to make a small ledger API safe, from a single transfer up to batches that skip the transfers that fail.

Everything here is the transactions example from the Mitsuki repository.

Mitsuki

Mitsuki is an batteries-included, structure-first web development framework for Python. It's opinionated, lightweight and performant.

What are Transactions?

In SQL, a transaction is everything between BEGIN and COMMIT:

BEGIN;
UPDATE accounts SET balance = balance - 50 WHERE id = 1;
UPDATE accounts SET balance = balance + 50 WHERE id = 2;
COMMIT;    -- both changes become permanent together
-- or
ROLLBACK;  -- neither change is applied

If anything fails between BEGIN and COMMIT (an error, divine intervention, a crash, a dropped connection), the database discards every change made so far.

The guarantees are usually summed up as ACID:

For this, a transaction lives on a single database connection (at least for non-distributed databases).

A BEGIN on one connection has no effect on another, so for several statements to be in one transaction, they all have to run on the same connection.

When are Transactions Useful?

By default, a Mitsuki repository call opens a connection, runs its statement, commits and returns the connection to the pool. That's the right default for a single write, but not for two writes that belong together.

The example's transfer withdraws from one account and deposits into the other:

@Service()
class TransferService:
    def __init__(self, accounts: AccountService, audit: AuditService):
        self.accounts = accounts
        self.audit = audit

    async def transfer(self, transfer: Transfer) -> None:
        ...
        await self.accounts.withdraw(transfer.source_id, transfer.amount)
        await self.accounts.deposit(transfer.target_id, transfer.amount)

Without a transaction, a transfer to an account that doesn't exist withdraws the money, fails on the deposit, and returns an error.

The withdrawal has already been committed:

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":70},{"id":2,"owner":"bob","balance":80}]

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 99, "amount": 50}'
{"error":"Account 99 does not exist"}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":20},{"id":2,"owner":"bob","balance":80}]

Alice is 50 down but nobody received it. Of course, you can catch the exception, and return the money to Alice, but that's another operation that needs to be persisted.

Instead of persisting one operation, then persisting a revert operation, you can simply... rollback the original operation before it's even applied. This is where transactions come into play!

Example Application

The transactions example runs against PostgreSQL.

Pull the repository, then let's run the app:

$ cd examples/transactions
$ docker compose up -d
$ pip install "mitsuki[postgresql]"
$ python app.py

The app's structure:

transactions/
├── app.py
├── application.yml
├── docker-compose.yml
└── src/
    ├── domain/          # Account, AuditEntry, AccountNotFound, InsufficientFunds
    ├── repositories/    # AccountRepository, AuditRepository
    ├── services/        # AccountService, TransferService, AuditService
    └── controllers/     # /api/accounts, /api/transfers, /api/audit

Let's open two accounts in the database to share balance between:

$ curl -X POST http://localhost:8000/api/accounts \
    -H "Content-Type: application/json" \
    -d '{"owner": "alice", "balance": 100}'
{"id":1,"owner":"alice","balance":100}

$ curl -X POST http://localhost:8000/api/accounts \
    -H "Content-Type: application/json" \
    -d '{"owner": "bob", "balance": 50}'
{"id":2,"owner":"bob","balance":50}

Making a Transfer Atomic with @Transactional

Add @Transactional() to the method:

from mitsuki import Service, Transactional

@Service()
class TransferService:
    ...

    @Transactional()
    async def transfer(self, transfer: Transfer) -> None:
        ...
        await self.accounts.withdraw(transfer.source_id, transfer.amount)
        await self.accounts.deposit(transfer.target_id, transfer.amount)

The whole method now runs in one transaction. It commits when the method returns, and rolls back when it raises. The exception is then re-raised, so the controller still turns it into an error response:

@PostMapping("/transfers")
@Consumes(Transfer)
async def transfer(self, body: Transfer = RequestBody()) -> ResponseEntity:
    try:
        await self.transfers.transfer(body)
    except AccountNotFound as error:
        return ResponseEntity.not_found({"error": str(error)})
    except InsufficientFunds as error:
        return ResponseEntity.conflict({"error": str(error)})
    return ResponseEntity.ok({"status": "completed"})

A transfer that goes through, then the same failing transfer as before:

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 2, "amount": 30}'
{"status":"completed"}

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 99, "amount": 50}'
{"error":"Account 99 does not exist"}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":70},{"id":2,"owner":"bob","balance":80}]

The withdrawal from Alice was rolled back with the failed deposit. Neither the repositories nor AccountService changed: built-in repository methods, Query DSL methods, @Query methods and get_connection() all join the transaction in progress.

How Transactions Work in Mitsuki

When transfer starts, Mitsuki opens a connection, runs BEGIN, and binds the connection to the current request through a ContextVar.

Every repository call awaited inside the method, finds that connection and runs on it instead of opening its own.

When the method finishes, Mitsuki commits or rolls back and returns the connection to the pool.

Mitsuki only decides where transactions begin and end. Atomicity, locking and isolation are the database's job, reached through SQLAlchemy.

Propagation: When Transactional Methods Call Each Other

AccountService.withdraw and deposit are transactional as well, because they're also safe to call on their own:

@Service()
class AccountService:
    def __init__(self, accounts: AccountRepository):
        self.accounts = accounts

    @Transactional()
    async def withdraw(self, account_id: int, amount: int) -> Account:
        account = await self.get(account_id)
        if account.balance < amount:
            raise InsufficientFunds(
                f"Account {account_id} holds {account.balance}, not {amount}"
            )
        account.balance -= amount
        return await self.accounts.save(account)

    @Transactional()
    async def deposit(self, account_id: int, amount: int) -> Account:
        account = await self.get(account_id)
        account.balance += amount
        return await self.accounts.save(account)

So what happens when transfer, already in a transaction, calls them?

That's decided by propagation. Mitsuki supports three modes:

REQUIRED: Join the Caller

withdraw() and deposit() use the default, so inside transfer() they join its transaction rather than committing on their own. That's why the failed deposit above could undo the withdrawal.

REQUIRES_NEW: An Audit Log That Survives Rollbacks

In the example app, every transfer request is written to an audit log, including the ones that fail. Though, if the audit entry joined the transfer's transaction, a failed transfer would roll its own audit entry back.

REQUIRES_NEW writes it in a separate transaction on a separate connection, which commits on its own:

@Service()
class AuditService:
    def __init__(self, entries: AuditRepository):
        self.entries = entries

    @Transactional(propagation=Propagation.REQUIRES_NEW)
    async def record(self, message: str) -> None:
        await self.entries.save(AuditEntry(message=message))

transfer() records the request before moving any money:

@Transactional()
async def transfer(self, transfer: Transfer) -> None:
    await self.audit.record(
        f"Transfer of {transfer.amount} from {transfer.source_id} "
        f"to {transfer.target_id} requested"
    )
    await self.accounts.withdraw(transfer.source_id, transfer.amount)
    await self.accounts.deposit(transfer.target_id, transfer.amount)

After the transfers above, let's add one Alice can't cover:

$ curl -X POST http://localhost:8000/api/transfers \
    -H "Content-Type: application/json" \
    -d '{"source_id": 1, "target_id": 2, "amount": 500}'
{"error":"Account 1 holds 70, not 500"}

$ curl http://localhost:8000/api/audit
[{"id":1,"message":"Transfer of 30 from 1 to 2 requested"},
 {"id":2,"message":"Transfer of 50 from 1 to 99 requested"},
 {"id":3,"message":"Transfer of 500 from 1 to 2 requested"}]

Two of the three transfers rolled back, and all three audit entries are kept.

NESTED: A Batch That Skips Failures

In the example app, a batch endpoint runs several transfers at once. If any transfer fails, we want to skip it and still keep the rest.

With REQUIRED, a failed transfer would doom the whole batch. With REQUIRES_NEW, each transfer would commit on its own, so a crash halfway through would leave half a batch committed.

NESTED runs each transfer behind a savepoint, which is a marker inside the transaction that the database can roll back to without abandoning the transaction itself:

@Transactional()
async def transfer_all(self, transfers: List[Transfer]) -> List[bool]:
    """Run every transfer that can be covered and skip the rest."""
    completed = []
    for transfer in transfers:
        try:
            async with transaction(propagation=Propagation.NESTED):
                await self.transfer(transfer)
            completed.append(True)
        except (AccountNotFound, InsufficientFunds):
            completed.append(False)
    return completed

transaction() is the context-manager form of @Transactional, with the same options. Here it wraps a single loop iteration.

A failed transfer rolls back to its savepoint, and the batch carries on:

$ curl -X POST http://localhost:8000/api/transfers/batch \
    -H "Content-Type: application/json" \
    -d '{"transfers": [
          {"source_id": 2, "target_id": 1, "amount": 20},
          {"source_id": 1, "target_id": 2, "amount": 1000},
          {"source_id": 1, "target_id": 2, "amount": 10}
        ]}'
{"completed":[true,false,true]}

$ curl http://localhost:8000/api/accounts
[{"id":1,"owner":"alice","balance":80},{"id":2,"owner":"bob","balance":70}]

The two transfers that could be covered moved the money, and the batch committed them together at the end.

Rollback Rules

Any exception rolls back, including asyncio.CancelledError, so a request cancelled mid-transfer never commits half of it.

To commit despite a particular exception, list it in no_rollback_for:

@Transactional(no_rollback_for=(PaymentDeclined,))
async def checkout(self, cart: Cart):
    ...

The exception is still re-raised so you can log it, handle it depending on your business logic, etc.

Isolation

Atomicity covers failures within one transaction. Isolation covers transactions running at the same time.

Two withdrawals of 80 from a balance of 100, running concurrently, can both read 100 before either writes, both pass the balance check, and both commit.

Whether that happens depends on the isolation level, which you set per method. For example:

@Transactional(isolation=Isolation.SERIALIZABLE)
async def withdraw(self, account_id: int, amount: int) -> Account:
    ...

At SERIALIZABLE, the database aborts one of the two withdrawals, and a retry sees the new balance.

The isolation section of the documentation walks through what each level allows, and how PostgreSQL, MySQL and SQLite differ.

Limitations

• Async methods only. @Transactional on a sync method raises TypeError.
• A transaction holds a connection until it ends. Keep slow work, such as HTTP calls to other services, out of transactional methods, or enough concurrent requests will exhaust the pool.
• No asyncio.gather inside a transaction. A connection runs one statement at a time, so repository calls from concurrent tasks inside a transaction raise TransactionException. Await them one after another.
• SQLite allows one writer at a time. Propagation modes like REQUIRES_NEW can't write while its caller holds the write lock. Note that underlying databases/engines have different behavior.

Next Steps

• Run the transactions example and try the batch endpoint with your own transfers.
• For every option, including class-level @Transactional, get_connection() inside a transaction comparison, read the Transactions documentation.
• Learn how the repositories used here work in Guide to Mitsuki Repositories: From Zero to Full CRUD.