transactional-boundary

한국어

Spring-style declarative transaction boundaries for SQLAlchemy — enforced at runtime, not by convention.

A write happened outside a transaction boundary (ORM write). Nothing will be saved.
  where: services/orders.py:20 (create_order)
  fix: Add @transactional to the function at the location above.
       If this helper only runs inside an existing boundary, decorate its caller instead.
  See: https://github.com/nohchang/transactional-boundary#boundaries

Most transaction decorators stop at "if you remember the decorator, it commits". This package adds the other half: a runtime guard that makes forgetting the decorator an error instead of a silent data loss. Every write that reaches the database is checked against a declared boundary — no boundary, no write, and a BoundaryError that names the function to fix.

Both session types are covered: Session through txbound, AsyncSession through txbound.asyncio, with the same vocabulary in each — see Async.

The problem

When session.commit() calls are scattered across a codebase, every developer has to keep three things in their head at once:

• A missing commit is a silent loss. session.add() without a later commit doesn't fail — the row just never appears, and nothing points at the function that forgot.
• Catching an exception is not enough. After a database error, the session is left in a failed state; forget the rollback and every later operation dies with PendingRollbackError, far from the actual bug.
• Composing services multiplies commits. If ServiceA.do() commits and ServiceB.do() calls it before doing its own work, a failure in B leaves A's half of the unit permanently committed.

Declarative boundaries collapse those three into one rule: the function that does the work declares the transaction, and the library commits or rolls back.

Why not just decorators

Because a decorator alone doesn't fix the failure mode that matters. If committing only happens where someone remembered @transactional, then a write where nobody remembered it is still a silent loss — you've moved the convention, not enforced it.

txbound installs an event guard on the session (install_boundary_guard). A flush, a Core insert()/update()/delete(), or a textual DML statement that executes outside a declared boundary raises BoundaryError immediately, with the message shown above — and the rejected state is disposed of, so it cannot ride along on a later, unrelated commit.

Install

pip install transactional-boundary

# for txbound.asyncio — adds SQLAlchemy's asyncio extra, which is what pulls
# greenlet in. On a machine SQLAlchemy does not list greenlet unconditionally
# for (an Apple Silicon Mac reports arm64, which is not on that list), the
# plain install has no greenlet and the first awaited operation inside a
# boundary raises about it. Pick your own async driver: asyncpg, aiosqlite, ...
pip install "transactional-boundary[asyncio]"

The import name is txbound (deliberately shorter than the PyPI name):

import txbound

Quick start

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

import txbound
from txbound import transactional


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    label: Mapped[str]


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

# Once, at startup. Only @independent_transaction needs this — it opens its
# own sessions. Register a sessionmaker, not a scoped_session.
txbound.configure(session_factory=SessionLocal)


class OrderService:
    def __init__(self, session):
        self.session = session

    @transactional
    def place(self, label: str) -> int:
        order = Order(label=label)
        self.session.add(order)
        self.session.flush()          # assigns order.id
        return order.id

    @transactional(read_only=True)
    def count(self) -> int:
        return self.session.query(Order).count()


# The entrypoint (HTTP handler, worker, test) creates the session, owns its
# lifetime, and installs the guard.
session = SessionLocal()
txbound.install_boundary_guard(session)

service = OrderService(session)
service.place("first")     # committed when place() returns
assert service.count() == 1
session.close()

The session is found on self.session (a service method) or among the positional arguments, searched left to right — so a classmethod, whose leading positional is cls, works too. Either way it must arrive positionally: keyword arguments are not searched, so place(session=s) is invisible here.

The engine above is SQLite. It needs one extra piece of engine setup, but only if an outer boundary delegates to an inner one before writing anything of its own — which nothing above does, and neither does Composing services below. See SQLite: take BEGIN control away from the driver for the shape that does need it.

The vocabulary

Two decorators × one read_only flag. The axes are orthogonal: the decorator decides where the session comes from, the flag decides whether anything is committed. There are no combinations to memorize.

Async

Install the extra — it is what brings greenlet in, and without it the first awaited operation inside a boundary raises about greenlet rather than anything to do with transactions:

pip install "transactional-boundary[asyncio]"

AsyncSession is then driven by a parallel pair of decorators:

from txbound.asyncio import independent_transaction, transactional

The vocabulary above is unchanged — two decorators × one read_only flag, REQUIRED and REQUIRES_NEW, a joined inner boundary isolated in a savepoint, a write inside a read-only boundary rejected. What differs is mechanical: the decorated function is an async def, and its body is awaited inside the boundary.

import asyncio

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

import txbound
from txbound.asyncio import transactional


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    label: Mapped[str]


engine = create_async_engine("sqlite+aiosqlite://")
AsyncSessionLocal = async_sessionmaker(bind=engine)

# Once, at startup. Only @independent_transaction needs this — it opens its
# own sessions. Register an async_sessionmaker, not an async_scoped_session.
txbound.configure(async_session_factory=AsyncSessionLocal)


class OrderService:
    def __init__(self, session: AsyncSession):
        self.session = session

    @transactional
    async def place(self, label: str) -> int:
        order = Order(label=label)
        self.session.add(order)
        await self.session.flush()          # assigns order.id
        return order.id

    @transactional(read_only=True)
    async def count(self) -> int:
        return await self.session.scalar(select(func.count()).select_from(Order))


async def main() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    # The entrypoint creates the session, owns its lifetime, and installs the
    # guard — on the AsyncSession itself.
    session = AsyncSessionLocal()
    txbound.install_boundary_guard(session)

    service = OrderService(session)
    await service.place("first")     # committed when place() returns
    assert await service.count() == 1
    await session.close()


asyncio.run(main())

install_boundary_guard takes the AsyncSession itself — it attaches to the plain Session that session drives, which is where the ORM events fire. The rest of the package root serves both worlds too — mark_read_only_entrypoint, configure, BoundaryError; only the two decorators are duplicated.

The engine above is SQLite, under the same condition as the synchronous half: the BEGIN-control setup is needed only if an outer boundary delegates to an inner one before writing anything of its own, and for an AsyncEngine it goes on engine.sync_engine. See SQLite: take BEGIN control away from the driver.

configure has one slot per world, both keyword-only, at least one required:

txbound.configure(
    session_factory=SessionLocal,             # txbound's decorators
    async_session_factory=AsyncSessionLocal,  # txbound.asyncio's
)

The slots are independent — a call filling one leaves the other alone, so a codebase migrating to async can register them in two places without losing the first.

What is not supported

A synchronous decorator cannot drive an AsyncSession, and an async decorator cannot drive a Session. Both directions raise rather than degrade, and the message names the module to switch to. Left to run, the first commits nothing behind an unawaited coroutine and the second dies with a TypeError about awaiting None from inside this package — neither is a failure worth inheriting. The function shape is checked the same way, at import time: @transactional on an async def is refused (use txbound.asyncio), and txbound.asyncio's decorators on a plain def are refused (use txbound).

Boundaries

Every runtime error links here. These are the rules the guard enforces:

1. Every write needs a declared boundary. A flush, Core DML, textual DML, or bulk_* call outside @transactional / @independent_transaction raises BoundaryError, and the rejected pending state is rolled away — it will not be committed later by an unrelated boundary.
2. Only the outermost boundary commits. An inner @transactional joins the outer one (REQUIRED propagation) and runs inside a savepoint; see Composing services.
3. A read-only boundary commits nothing. A write inside @transactional(read_only=True) raises — including when the read-only call has joined an enclosing write boundary; the writable flag is dropped for exactly that stretch and restored afterwards. A session.add() that never flushed is caught at the end of the boundary instead of silently vanishing on the rollback.
4. An entrypoint can declare the whole session read-only. mark_read_only_entrypoint(session) (say, for HTTP GET/HEAD/OPTIONS requests, or a report-only worker) makes any later attempt to open a write boundary on that session raise.
5. A boundary refuses to open over leaked state. Uncommitted changes that entered the session before the boundary opened are not this boundary's work; it raises rather than adopting them.
6. @independent_transaction requires a separate session on a separate connection. A scoped_session hands back the caller's own session; an engine using SingletonThreadPool or StaticPool (which is what sqlite:// gives you) hands back the caller's own connection. Either one silently defeats REQUIRES_NEW, so both raise ConfigurationError rather than committing the caller's work by accident — the scoped registry at configure(), before anything can run on it, and the shared pool on the first call.
7. Each world drives its own session type, loudly. The synchronous pair drives a Session, txbound.asyncio's drives an AsyncSession, and a decorator handed the other kind raises instead of degrading — with the message naming the module to switch to. A function whose body does not run when it is called is refused the same way, at import time: a generator or an async generator under either pair, and an async def under the synchronous one. The boundary would commit an empty session and close before the body ever ran, losing the write with nothing raised. See Async.

Under an AsyncSession, none of the above changes. Boundary state lives on session.info, and an AsyncSession's info is not a copy of the Session it drives — it is literally the same dict, so the two objects are one boundary however you reach it. install_boundary_guard, handed the AsyncSession, attaches its listeners to that inner Session, where all four write shapes fire — including the bulk_* one, reached through await session.run_sync(...). A boundary whose body is awaited opens when the coroutine starts and settles when it returns, so it holds its transaction across every await inside it; that is what the static checker's rule 7 (external-await-in-boundary) is about.

Every message carries a location. A guard event names the line that made the write. A refusal raised as a boundary opens — a read-only entrypoint, leaked state, the wrong kind of session — names path:line (Qualname) of where the boundary is declared, and says so on the line, because a decorator has no way to know its caller's line.

SQLite: take BEGIN control away from the driver

pysqlite and aiosqlite do not emit BEGIN until they see DML. An outer @transactional that has issued no write of its own before its first joined call therefore arrives at the inner boundary in autocommit mode: the SAVEPOINT lands outside any transaction, its RELEASE commits outright, and the outer's rollback then has nothing left to undo — the inner's write survives the failure that was supposed to erase it. Rather than let rule 2 above fail silently, that join raises ConfigurationError and hands you the fix:

from sqlalchemy import event


@event.listens_for(engine, "connect")
def _disable_driver_begin(dbapi_connection, record):
    dbapi_connection.isolation_level = None


@event.listens_for(engine, "begin")
def _emit_begin(conn):
    conn.exec_driver_sql("BEGIN")

For an AsyncEngine, listen on engine.sync_engine — Core events live on the synchronous engine underneath. PostgreSQL and every other dialect need none of this; their drivers open the transaction before the first statement.

The refusal is narrower than it sounds, and the scope is worth knowing:

• SQLite only, and a write join only. A read-only join writes nothing, so its own RELEASE loses nothing and it is never checked.
• Only when the outer has written nothing of its own yet. A single session.add(...) is already safe: opening the savepoint flushes first, and that flush is what starts the transaction. So is a Core insert() the outer has already executed. A query is not a write — an outer that ran a SELECT and then delegated is still refused, because a SELECT starts no transaction on this driver, and the refusal says so.

Composing services

An inner @transactional does not commit — it joins the outer boundary, so two services compose into one atomic unit:

class SignupService:
    def __init__(self, session):
        self.session = session
        self.orders = OrderService(session)

    @transactional
    def register(self, name: str) -> None:
        self.session.add(User(name=name))
        try:
            # joins register()'s boundary; runs inside a savepoint
            self.orders.place("welcome gift")
        except GiftUnavailable:
            pass  # optional step — signup must still go through

The joined inner call runs inside a savepoint. If place() fails and register() swallows the exception, only what place() wrote is rolled back — the User row commits normally. "The outer swallowed the inner's failure" never turns into "the inner's partial write got committed".

This example runs as written on SQLite: register() adds the User before it delegates, and that pending write is what starts the driver's transaction. An outer that delegates before writing anything of its own needs the engine to control BEGIN first — see SQLite above.

Coming from Spring? This is the one place txbound deliberately differs. Spring's REQUIRED opens no savepoint (that's NESTED): a swallowed inner failure marks the whole transaction rollback-only, and the outer commit throws UnexpectedRollbackException. txbound isolates the inner failure instead, and the outer commits normally. If you relied on Spring's "any inner failure poisons the unit" behaviour, re-raise instead of swallowing.

@independent_transaction

REQUIRES_NEW: the function always runs in its own session and its own transaction, committed (or rolled back) independently of whatever is happening outside. That is what you record failure reasons with — a write that must survive the rollback caused by the very failure it is recording:

class AuditLog:
    @staticmethod
    @independent_transaction
    def record_failure(session, *, order_id: int, reason: str) -> None:
        session.add(FailureRecord(order_id=order_id, reason=reason))

SQLAlchemy cannot park an ongoing transaction the way JPA can, so a new transaction means a separate session on a separate connection. Three things follow, and the decorator's enforced signature exists to keep you clear of them:

1. The identity map differs — an ORM object loaded outside must not be passed in. Pass primitive ids.
2. The outer transaction's uncommitted rows are invisible in here.
3. Updating a row the outer transaction already locked is a deadlock.

The function must be shaped (session, *, keyword_only_args) — the decorator creates the session and passes it as the first argument, and everything else must be keyword-only. There is no shape in which a caller can hand a session (or share one) — enforced at import time.

Two SQLite notes, because this is where the pattern is usually first tried.

In-memory SQLite is rejected outright. create_engine("sqlite://") uses SingletonThreadPool, and poolclass=StaticPool is the usual recipe for sharing one in-memory database across a test suite. Both hand the same connection to every checkout, so the "separate" session would sit on the caller's open transaction and its commit would commit the caller's uncommitted work too — with the caller's later rollback then finding nothing to undo. That is invisible, so txbound raises ConfigurationError instead. Point tests that exercise @independent_transaction at a file (sqlite:///./test.db) or a real database — and give that file engine the two BEGIN-control listeners (above) if the same suite also composes boundaries.

SQLite locks the whole file, not rows. With a file-backed database, if the outer transaction has flushed any write, an independent transaction's write blocks on it and dies with database is locked — even on unrelated tables. That is SQLite's single-writer model, not a row conflict; on PostgreSQL the same code runs fine. Develop this pattern against the engine you deploy on.

lock_timeout="3s" is the default on PostgreSQL, so consequence 3 fails fast with a lock error instead of hanging until the pool times out. Pass lock_timeout=None for paths that are supposed to wait (a batch serializing behind pg_advisory_xact_lock — PostgreSQL's lock_timeout applies to advisory-lock waits too). On other databases the default is silently skipped; an explicitly passed value warns instead of pretending.

read_only=True works as everywhere else, with one extra rule: return plain values, not ORM entities. The session closes on return and a read-only session's objects expire on rollback, so a returned entity raises DetachedInstanceError at first attribute access. (The write side returns entities fine — expire_on_commit is off.)

In txbound.asyncio it is the same decorator awaited:

from txbound.asyncio import independent_transaction


class AuditLog:
    @staticmethod
    @independent_transaction
    async def record_failure(
        session: AsyncSession, *, order_id: int, reason: str
    ) -> None:
        session.add(FailureRecord(order_id=order_id, reason=reason))

It opens its session from the async_session_factory slot, and refuses the same two setups for the same reasons: an async_scoped_session hands back the caller's own session, and a shared-connection pool hands back the caller's own connection — which is what an in-memory sqlite+aiosqlite:// URL gives you (StaticPool). Both raise ConfigurationError; the scoped registry is refused by configure() itself, so it never reaches a boundary.

Framework integration

txbound deliberately ships no framework glue — the seam is small enough to own. FastAPI:

from fastapi import Depends, Request
from sqlalchemy.orm import Session
from typing import Annotated, Generator
import txbound

READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})


def get_db_session(request: Request) -> Generator[Session, None, None]:
    session = SessionLocal()
    txbound.install_boundary_guard(session)
    if request.method in READ_ONLY_METHODS:
        txbound.mark_read_only_entrypoint(session)
    try:
        yield session
    finally:
        session.close()


DbSession = Annotated[Session, Depends(get_db_session)]

The async dependency is the same four calls, awaited where they have to be:

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession


async def get_async_db_session(
    request: Request,
) -> AsyncGenerator[AsyncSession, None]:
    session = AsyncSessionLocal()
    # The AsyncSession itself — the guard attaches to the Session it drives,
    # and mark_read_only_entrypoint writes to the info dict they share.
    txbound.install_boundary_guard(session)
    if request.method in READ_ONLY_METHODS:
        txbound.mark_read_only_entrypoint(session)
    try:
        yield session
    finally:
        await session.close()


AsyncDbSession = Annotated[AsyncSession, Depends(get_async_db_session)]

Flask is the same three calls in a before_request/teardown_request pair (create + guard on request start, close() on teardown), with mark_read_only_entrypoint keyed on request.method.

A worker or scheduler is even smaller: create the session, install the guard, call the service, close. If a worker only ever reads, mark it read-only at the top and any write that sneaks into it becomes an error instead of a surprise.

Static checks

The runtime guard cannot see code that never runs, and it structurally cannot see writes that bypass the session's event hooks. The checker covers that ground:

python -m txbound.check          # or: txbound-check

Upgrading from 0.2.0? Several rules report where they did not, so a build that was green can go red on unchanged code. The two with the most to say have callouts below; the rest, one line each:

• rule 2 catches two write shapes it used to miss — a chained Core DML call (execute(insert(...).values(...)), and the same for update/delete) and a literal execute(text("INSERT ...")). It also stops reporting a function whose boundary decorator arrived under an import alias.
• rule 3 resolves that same alias, so an @independent_transaction taking an ORM parameter is now reported where from txbound import independent_transaction as sync_it used to hide it.
• rule 7 exempts a call to a boundary declared anywhere in the scanned tree, so composed-service calls and @independent_transaction calls stop being reported as external round trips. The exemption is harvested from write boundaries only, so await self.repo.get(id) against a @transactional(read_only=True) accessor now reports where it did not.
• rule 6 exempts a swallow whose only database call is to a declared write boundary, and correspondingly no longer exempts one around a read-only inner call, or one that also touches the boundary's own session directly.
• rule 8's message for flush() no longer says the write is lost — inside a boundary the row is still saved by the commit, and what breaks is the id that call was supposed to fetch.

A # txbound: ignore[rule-name] comment is the escape for any finding that is wrong on your code. An annotation is not a general one: it helps only where the checker could not tell that a receiver is a session — annotating self.repo: ReadRepo does not silence rule 7, and rules 2 and 3 are answered by a boundary and by a scalar parameter respectively, not by a type.

Rule 8 — a forgotten await. What it reports is a real silent write loss: session.execute(...) without await builds a coroutine and throws it away, so the statement never runs, the boundary commits an empty transaction, and the caller sees success. The runtime guard structurally cannot see it — there is no statement for it to observe — which is why it lives here. # txbound: ignore[unawaited-session-call] is the local escape if a finding is wrong.

Its coverage stops where the checker stops being sure. The receiver has to resolve to an AsyncSession — a parameter or an attribute annotated with it, including through an alias like the AsyncDbSession above: a module-level alias is followed when the module defines it or imports it by name (from .deps import AsyncDbSession), across a chain of aliases and through a re-export in an __init__.py. Three things are not followed: a name the module neither defines nor imports, a name the module binds itself (its own class, def or assignment always wins), and a dotted reference (import app.deps, then db: app.deps.AsyncDbSession). The result must also be discarded inside an async def. That alias resolution is not specific to this rule — rules 1, 2, 5 and 7 read receivers through the same table.

A receiver that cannot be shown to be specifically an AsyncSession — including one that only comes from a factory call (db = AsyncSessionLocal()) — is not reported: a synchronous Session's flush()/delete() return None and have already run, so flagging those would be noise. The rule fails toward silence, so a clean run is not proof that every forgotten await is gone.

Rule 7 — renamed from async-boundary. Before 1.0.0 it flagged any boundary on an async def. That shape is now supported, so the rule was re-aimed at the hazard it had been standing in for, and renamed with it. Some builds that were red go green. But a # txbound: ignore[async-boundary] comment carried over from 0.2.0 now names a rule that no longer exists, and an ignore naming an unknown rule is reported as unused-ignore and exits 1 — on its own, with no other violation in the file. Delete it if it was excusing a boundary on an async def; rename it to external-await-in-boundary if the line really does await something other than its own session.

Rules 1 and 5–8 need no configuration, but they still need files: with no roots configured the checker walks nothing, so it reports "nothing was checked" and exits 1 rather than printing a pass. A gate that scans zero files must not be green.

Rules 2–4 additionally need to know where your code lives and stay off until configured — reported as skipped, never as passed:

[tool.txbound.check]
scan_roots = ["myapp/services"]         # rule 2: code that must declare boundaries
scan_files = ["myapp/tasks.py"]         # single files to scan, same rules
entrypoint_roots = ["myapp/api"]        # rules 1 & 4-8 additionally look here
model_patterns = ["myapp/**/models"]    # rule 3: model directories or .py files (glob)
session_factories = ["SessionLocal"]    # rule 4: what creates sessions
exempt_roots = ["myapp/migrations"]     # suppression — applies to every rule
extra_write_methods = []                # rule 2: custom query-builder write methods
unresolved_allowlist = []               # receivers confirmed not to be DB sessions

Any single line can opt out of a single rule:

session.commit()  # txbound: ignore[manual-session-control] — worker owns this session

The rule name is required — a bare # txbound: ignore is reported, not honoured, because it would also silence rules added in later releases. A trailing comment covers its own line and nothing else; a comment written on a line of its own also covers the line below it, which is what makes a multi-line statement suppressible.

An ignore that suppresses nothing is itself reported as unused-ignore — a misspelled or unknown rule name, a name that can never be suppressed, or a comment left behind after the code it once excused was fixed. When the named rule never ran over that file — it is unconfigured, or the file sits outside the roots it runs over — the report says so and tells you to keep the comment: the line is unchecked because the rule is off, not because the comment is dead.

--list prints one violation per line and exits 1 when violations exist, so both the human format and the machine format fail a build. Pre-commit:

- repo: local
  hooks:
    - id: txbound-check
      name: transaction boundary check
      entry: txbound-check
      language: system
      pass_filenames: false

How this differs

Two other Spring-style transaction packages exist on PyPI. They aim at different things — facts only, judge for yourself:

Compared against sqlalchemy-transactional 1.0.3 and transactional-sqlalchemy 0.1.9 (2026-08).

Everything in this package's column holds in both worlds. The runtime guard, the automatic savepoint isolation of a joined inner boundary and the rejection of a write inside a read_only boundary all work the same under an AsyncSession as under a Session, and two of the eight static rules exist specifically for async code. txbound.asyncio is the same vocabulary awaited, not a reduced subset.

Requirements

• Python 3.11+
• SQLAlchemy 2.0+
• For txbound.asyncio, the asyncio extra (pip install "transactional-boundary[asyncio]"), which adds SQLAlchemy's own asyncio extra and therefore greenlet, plus an async driver of your choice (asyncpg, aiosqlite, …). A sync-only stack needs none of it: nothing in the synchronous half imports sqlalchemy.ext.asyncio.

License

MIT