fastapi-transactional 0.1.0

Transactional session management for SQLAlchemy use cases (sync + async).

fastapi-transactional

Transactional session management for SQLAlchemy use cases — sync and async. Built for FastAPI and other apps that follow hexagonal / clean architecture, where a single use case orchestrates multiple repositories and they all need to share one transaction.

@with_transaction
def execute(self, policy_id: str) -> None:
    policy = self.policy_repo.find_by_id(policy_id)
    policy.status = "ACTIVE"
    self.policy_repo.save(policy)
    self.notification_repo.save(Notification(policy_id=policy_id))
    # Commit on success. Rollback on any exception. Both repos share the same session.

Why

In a hexagonal architecture, a use case typically depends on multiple repositories:

class ActivatePolicyUseCase:
    def __init__(self):
        self.policy_repo = PolicyRepository()
        self.notification_repo = NotificationRepository()

If each repository opens its own session, you lose transactional guarantees — a failure in notification_repo.save() won't undo the change in policy_repo.save(). The usual fix is to pass a session around explicitly, which leaks infrastructure into the domain layer.

fastapi-transactional solves this with a ContextVar-based session and a one-line decorator. Every repository inheriting from DatabaseRepository automatically receives the active session before the use case runs.

Install

pip install fastapi-transactional

Requires Python 3.10+ and SQLAlchemy 2.0+.

Quickstart — sync

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi_transactional import (
    DatabaseRepository, configure, with_transaction,
)

# 1. Configure once at startup.
engine = create_engine("postgresql+psycopg://user:pass@host/db")
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
configure(session_factory=SessionLocal)


# 2. Inherit from DatabaseRepository — `self.db` is injected automatically.
class PolicyRepository(DatabaseRepository):
    def find_by_id(self, policy_id):
        return self.db.get(Policy, policy_id)

    def save(self, policy):
        self.db.add(policy)


# 3. Decorate the use case method.
class ActivatePolicyUseCase:
    def __init__(self):
        self.policy_repo = PolicyRepository()

    @with_transaction
    def execute(self, policy_id: str) -> None:
        policy = self.policy_repo.find_by_id(policy_id)
        policy.status = "ACTIVE"
        self.policy_repo.save(policy)

Quickstart — async

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from fastapi_transactional import (
    AsyncDatabaseRepository, configure, with_async_transaction,
)

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
AsyncSessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)
configure(async_session_factory=AsyncSessionLocal)


class PolicyRepository(AsyncDatabaseRepository):
    async def find_by_id(self, policy_id):
        return await self.db.get(Policy, policy_id)

    async def save(self, policy):
        self.db.add(policy)


class ActivatePolicyUseCase:
    def __init__(self):
        self.policy_repo = PolicyRepository()

    @with_async_transaction
    async def execute(self, policy_id: str) -> None:
        policy = await self.policy_repo.find_by_id(policy_id)
        policy.status = "ACTIVE"
        await self.policy_repo.save(policy)

You can configure both sync and async factories at the same time — they live in separate ContextVars and don't interfere.

How it works

1. The decorator opens a session_manager() (sync) or async_session_manager() (async) context.
2. The manager creates a session from the factory you registered with configure(...) and stores it in a ContextVar.
3. The decorator scans self for attributes that are DatabaseRepository / AsyncDatabaseRepository instances and sets their .db to the active session.
4. Your method runs. On clean exit the session commits; on any exception it rolls back.
5. The session is always closed and removed from the ContextVar.

ContextVar is thread-safe and asyncio-safe — each request handler gets its own slot, so concurrent requests never see each other's sessions.

Nested use cases

A use case can call another use case without worrying about double transactions:

class Outer:
    def __init__(self):
        self.inner = Inner()

    @with_transaction
    def execute(self):
        self.inner.execute()      # Reuses the outer session.
        self.repo.save(thing)     # Same transaction.
        # Single commit when execute() returns.

When session_manager() sees an existing session bound to the context, it yields that session instead of opening a new one — so the outermost call owns the transaction lifecycle.

Advanced — accessing the session directly

For code that isn't a repository (e.g. a service that needs to run a raw query inside the active transaction):

from fastapi_transactional import get_current_session

def some_service():
    session = get_current_session()
    if session is None:
        raise RuntimeError("Must be called inside a transactional scope")
    session.execute(...)

The async equivalents are get_current_async_session, set_current_async_session, clear_current_async_session.

You can also use the context manager directly without the decorator:

with session_manager() as session:
    session.execute(...)
# Commit happens here. Rollback on exception.

API reference

Symbol	Purpose
configure(session_factory=..., async_session_factory=...)	Register session factories (call once at startup).
reset()	Clear the registered factories (useful in tests).
session_manager() / async_session_manager()	Context manager that opens a transactional scope.
with_transaction / with_async_transaction	Decorator wrapping a method in a transactional scope and injecting sessions into repositories.
DatabaseRepository / AsyncDatabaseRepository	Base class — gives the repo a self.db slot.
get_current_session / set_current_session / clear_current_session	Direct access to the sync ContextVar.
get_current_async_session / set_current_async_session / clear_current_async_session	Direct access to the async ContextVar.

Tests

pip install -e ".[test]"
pytest --cov=fastapi_transactional

Contributing

Issues and pull requests welcome. See CONTRIBUTING.md.

License

MIT — see LICENSE.