sqlalchemy-transactional

Spring-style transactional boundaries for SQLAlchemy async sessions. Inspired by Spring Framework's @Transactional model.

Table of Contents

Quick Start
Why Use This
Propagation Modes
Isolation Level
Spring-to-FastAPI Mapping
Public API
Contributing

Quick Start

This quick start uses a FastAPI integration example to show transaction boundaries in a real application flow, not as isolated snippets. It assumes you already run FastAPI with an async_sessionmaker and want Spring-like declarative transaction boundaries in service methods.

from fastapi import FastAPI, Request
from pydantic import BaseModel
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from sqlalchemy_transactional.asyncio import (
    current_session,
    sessionmaker_context,
    transactional,
)
from sqlalchemy_transactional.common import Propagation

app = FastAPI()
engine = create_async_engine("sqlite+aiosqlite:///app.db")
sessionmaker = async_sessionmaker(engine, expire_on_commit=False)


# 1) Bind async_sessionmaker once per request boundary
@app.middleware("http")
async def transactional_context_middleware(request: Request, call_next):
    async with sessionmaker_context(sessionmaker):
        return await call_next(request)


class CreateUserRequest(BaseModel):
    name: str


# 2) Put transaction boundaries on service methods
class UserService:
    @transactional  # default propagation = REQUIRED
    async def create_user(self, name: str) -> None:
        await self._validate_name(name)
        await self._insert_user(name)

    @transactional(Propagation.MANDATORY)
    async def _validate_name(self, name: str) -> None:
        result = await current_session().execute(
            text("SELECT 1 FROM users WHERE name = :name"),
            {"name": name},
        )
        if result.first() is not None:
            raise ValueError("name already exists")

    @transactional(Propagation.MANDATORY)
    async def _insert_user(self, name: str) -> None:
        await current_session().execute(
            text("INSERT INTO users (name) VALUES (:name)"),
            {"name": name},
        )


user_service = UserService()


# 3) Route/controller stays thin
@app.post("/users")
async def create_user(payload: CreateUserRequest) -> dict[str, str]:
    await user_service.create_user(payload.name)
    return {"status": "ok"}

Why Use This

Keep transaction plumbing out of service code.
Declare transaction scope at the function boundary.
Use familiar propagation semantics from Spring (REQUIRED, MANDATORY, etc.).
Apply isolation level declaratively when needed.
Make service methods easier to read, test, and review.

Propagation Modes

REQUIRED (default): Join an active transaction, or create one if none exists.
MANDATORY: Require an active transaction; raise an error if missing.
REQUIRES_NEW: Always execute in a new transaction.
NESTED: Use savepoint semantics inside an active transaction; otherwise behave like REQUIRED.

Isolation Level

Set isolation level at the decorator boundary:

@transactional(isolation_level="SERIALIZABLE")
async def run_settlement() -> None:
    await current_session().execute(...)

If isolation_level is omitted (None, the default), SQLAlchemy uses the engine/dialect default isolation behavior from your database driver.

Combine with propagation when needed:

@transactional(Propagation.REQUIRES_NEW, isolation_level="SERIALIZABLE")
async def write_critical_audit_log() -> None:
    await current_session().execute(...)

When isolation_level is applied:

The decorator creates a new transaction (REQUIRES_NEW, or REQUIRED / NESTED when no active transaction exists).

When isolation_level is not applied by this decorator:

The function joins an already active transaction (REQUIRED with an active transaction, MANDATORY, or NESTED with an active transaction).

Spring-to-FastAPI Mapping

Spring concept	FastAPI + sqlalchemy-transactional
`@Transactional` on service methods	`@transactional` on async service methods
`@Transactional(propagation = REQUIRED)`	`@transactional` (default = `Propagation.REQUIRED`)
`@Transactional(propagation = MANDATORY)`	`@transactional(Propagation.MANDATORY)`
Request filter/interceptor binds tx resources	`@app.middleware("http")` + `sessionmaker_context(sessionmaker)`
Current tx-bound resource lookup	`current_session()`

Public API

The primary integration surface is:

sqlalchemy_transactional.asyncio
sqlalchemy_transactional.common