Metadata-Version: 2.4
Name: fastapi-transactional
Version: 0.1.0
Summary: Transactional session management for SQLAlchemy use cases (sync + async).
Project-URL: Homepage, https://github.com/oviladrosa/fastapi-transactional
Project-URL: Repository, https://github.com/oviladrosa/fastapi-transactional
Project-URL: Issues, https://github.com/oviladrosa/fastapi-transactional/issues
Project-URL: Changelog, https://github.com/oviladrosa/fastapi-transactional/blob/main/CHANGELOG.md
Author-email: Oriol Viladrosa <oviladrosa@gmail.com>
License: MIT License
        
        Copyright (c) 2026 uri_vg
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: async,context-manager,decorator,fastapi,hexagonal-architecture,sqlalchemy,transactions
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: sqlalchemy>=2.0
Provides-Extra: dev
Requires-Dist: aiosqlite>=0.19; extra == 'dev'
Requires-Dist: build>=1.2; extra == 'dev'
Requires-Dist: greenlet>=3; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pre-commit>=3.7; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=4; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Requires-Dist: twine>=5; extra == 'dev'
Provides-Extra: test
Requires-Dist: aiosqlite>=0.19; extra == 'test'
Requires-Dist: greenlet>=3; extra == 'test'
Requires-Dist: pytest-asyncio>=0.23; extra == 'test'
Requires-Dist: pytest-cov>=4; extra == 'test'
Requires-Dist: pytest>=8; extra == 'test'
Description-Content-Type: text/markdown

# fastapi-transactional

[![PyPI version](https://img.shields.io/pypi/v/fastapi-transactional.svg)](https://pypi.org/project/fastapi-transactional/)
[![Python versions](https://img.shields.io/pypi/pyversions/fastapi-transactional.svg)](https://pypi.org/project/fastapi-transactional/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![CI](https://github.com/oviladrosa/fastapi-transactional/actions/workflows/ci.yml/badge.svg)](https://github.com/oviladrosa/fastapi-transactional/actions/workflows/ci.yml)

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.

```python
@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:

```python
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

```bash
pip install fastapi-transactional
```

Requires Python 3.10+ and SQLAlchemy 2.0+.

## Quickstart — sync

```python
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

```python
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 `ContextVar`s 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:

```python
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):

```python
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:

```python
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

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

## Contributing

Issues and pull requests welcome. See [CONTRIBUTING.md](CONTRIBUTING.md).

## License

MIT — see [LICENSE](LICENSE).
