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.

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

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 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.

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.
7. Sync only, loudly. An AsyncSession is rejected with a clear error instead of committing nothing behind an unawaited coroutine. So is any function whose body does not run when it is called — async def, a generator, an async generator — because the boundary would commit an empty session and close before the body ever ran, losing the write with nothing raised. All three fail at import time. See Async support.

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".

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.

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.)

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)]

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

Rules 1 and 5–7 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-7 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

--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).

Async support

This release is synchronous only. AsyncSession is not supported yet — it raises rather than failing silently. Async support is planned for a following release. Until then, if your stack is async end-to-end, the two packages above both handle AsyncSession today and are the better fit; if you only have an async edge, a synchronous session run off the event loop (asyncio.to_thread, or a sync worker) works with txbound as-is.

Requirements

• Python 3.11+
• SQLAlchemy 2.0+

License

MIT