# SQLAlchemy Transaction Context
Minimalistic context-local session and transaction controller for SQLAlchemy AsyncSession.
`sqlalchemy-tx-context` provides context-aware session and transaction management using Python’s contextvars,
eliminating the need to pass AsyncSession objects explicitly. Especially useful in code where database access should
be decoupled from explicit session passing - such as service layers or background jobs.
---
## Features
- Context-local session management via `contextvars`, without relying on thread-locals or global session objects.
- Clean `async with` API for managing session and transaction scopes.
- Supports safe nesting of transactions.
- `.execute(...)` automatically creates a session or transaction if none is active (optional fallback).
- Explicit control over how sessions are scoped and reused.
---
## Installation
```shell
pip install sqlalchemy-tx-context
```
---
## Quick Example
```python
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy_tx_context import SQLAlchemyTransactionContext
engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")
db = SQLAlchemyTransactionContext(engine)
async def create_user():
async with db.transaction():
await db.execute(insert(User).values(name="John"))
async def get_users():
async with db.session():
result = await db.execute(select(User))
return result.scalars().all()
```
## API Reference
---
### Constructor
```python
SQLAlchemyTransactionContext(
engine: AsyncEngine,
*,
default_session_maker: Optional[async_sessionmaker[AsyncSession]] = None,
auto_context_on_execute: bool = False,
auto_context_force_transaction: bool = False,
)
```
---
### Parameters:
- `engine` - SQLAlchemy `AsyncEngine` instance.
- `default_session_maker` - Optional async session factory. If omitted, uses `async_sessionmaker(engine)`.
- `auto_context_on_execute` - If `True`, allows `.execute()` to run even without an active session
by creating a temporary one.
- `auto_context_force_transaction` - If `True`, `.execute()` always runs inside a transaction when auto context is used.
If `False`, it uses `.session()` for read-only queries (like `Select` or `CompoundSelect`),
and `.transaction()` for everything else, including `Insert`, `Update`, and raw SQL.
---
### Session Methods:
- `session(...) -> AsyncIterator[AsyncSession]` - Enter a new session context, or reuse an existing
one if `reuse_if_exists=True`.
- `transaction(...) -> AsyncIterator[AsyncSession]` - Enter a transactional context.
Will nest if a transaction is already active.
- `new_session(...) -> AsyncIterator[AsyncSession]` - Create a new isolated session, even if another is already active.
Overrides the context for the duration.
- `new_transaction(...) -> AsyncIterator[AsyncSession]` - Create a new transaction in an isolated session.
- `get_session(strict: bool = True) -> AsyncSession | None` - Return the current session from context.
Raises `NoSessionError` if `strict=True` and no session exists.
- `execute(...) -> Result` - Execute a SQLAlchemy `Executable` using the current or temporary context.
Uses the current session if one is active. Otherwise, behavior depends on `auto_context_on_execute` -
a new session or transaction context may be created automatically.
---
## Auto-context Example
```python
# Opens a temporary session or transaction depending on statement type
db = SQLAlchemyTransactionContext(engine, auto_context_on_execute=True)
await db.execute(insert(User).values(name="Alice"))
await db.execute(select(User))
```
---
## Full Example
For a complete working example using PostgreSQL, see
[`example/`](https://github.com/QuisEgoSum/sqlalchemy-tx-context/tree/main/example).
It demonstrates table creation, data insertion, transactional rollback, and querying.
---
## Exceptions
This library defines a few custom exceptions to help catch context-related mistakes:
- `NoSessionError`: Raised when attempting to access or use a session when none is active and fallback is disabled.
- `SessionAlreadyActiveError`: Raised when entering `.session()` while a session is already active
(unless `reuse_if_exists=True`).
- `TransactionAlreadyActiveError`: Raised when entering `.transaction()` while a transaction is already active
and nesting is disabled.
---
## Motivation
SQLAlchemy does not provide an out-of-the-box solution for context-local session tracking when working with
`AsyncSession`. Passing sessions around explicitly can make service-layer code verbose and harder to maintain.
The library introduces a lightweight and predictable abstraction that:
- Stores current session in a `ContextVar`.
- Provides safe transactional/session boundaries.
- Exposes a unified `execute(...)` interface.
- Integrates with standard SQLAlchemy models, statements, and engines.
---
## When should you use this?
- You're writing service-layer logic and want to avoid passing `session` explicitly.
- You need nested transactional logic with clean context boundaries.
- You prefer explicit context management over dependency-injected sessions.
- You work with context-local transaction boundaries in background tasks or microservices.
- You need precise control over session lifecycle and scope.
This library is best suited for functional or script-style code where sessions are not injected via DI frameworks.
---
## Best Practices
- Use `.transaction()` or `.session()` explicitly in your service-layer or job code
to clearly define execution boundaries.
- `.execute(...)` is ideal for small projects, functional-style code, or early-stage prototypes - it avoids
boilerplate and makes code fast to write.
- As your project grows, you can gradually migrate to DI-based session and transaction management:
just replace `db.execute(...)` with `self._session.execute(...)`
and `db.transaction()` with your own context (e.g., `UnitOfWork`, `session.begin()`) - the query logic remains the same.
- This makes the library especially useful for bootstrapping or background scripts,
where a full-blown DI setup would be overkill.
---
## Tests
```shell
pytest --cov=sqlalchemy_tx_context --cov-report=term-missing
```
---
## Compatibility
Tested on Python `3.9` - `3.12`.
---
## License
MIT License
Raw data
{
"_id": null,
"home_page": null,
"name": "sqlalchemy-tx-context",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "sqlalchemy, async, contextvars, transaction, session",
"author": null,
"author_email": "QuisEgoSum <subbotin.evdokim@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/78/25/ad45257d084012fe10127e9bc0cb56eb30230585bd8eaa638dbfd1916ba7/sqlalchemy_tx_context-1.0.1.tar.gz",
"platform": null,
"description": "# SQLAlchemy Transaction Context\n\nMinimalistic context-local session and transaction controller for SQLAlchemy AsyncSession.\n\n`sqlalchemy-tx-context` provides context-aware session and transaction management using Python\u2019s contextvars,\neliminating the need to pass AsyncSession objects explicitly. Especially useful in code where database access should\nbe decoupled from explicit session passing - such as service layers or background jobs.\n\n---\n\n## Features\n\n- Context-local session management via `contextvars`, without relying on thread-locals or global session objects.\n- Clean `async with` API for managing session and transaction scopes.\n- Supports safe nesting of transactions.\n- `.execute(...)` automatically creates a session or transaction if none is active (optional fallback).\n- Explicit control over how sessions are scoped and reused.\n\n---\n\n## Installation\n\n```shell\npip install sqlalchemy-tx-context\n```\n\n---\n\n## Quick Example\n\n```python\nfrom sqlalchemy.ext.asyncio import create_async_engine\nfrom sqlalchemy_tx_context import SQLAlchemyTransactionContext\n\nengine = create_async_engine(\"postgresql+asyncpg://user:pass@host/db\")\n\ndb = SQLAlchemyTransactionContext(engine)\n\nasync def create_user():\n async with db.transaction():\n await db.execute(insert(User).values(name=\"John\"))\n\nasync def get_users():\n async with db.session():\n result = await db.execute(select(User))\n return result.scalars().all()\n```\n\n\n## API Reference\n\n---\n\n### Constructor\n\n```python\nSQLAlchemyTransactionContext(\n engine: AsyncEngine,\n *,\n default_session_maker: Optional[async_sessionmaker[AsyncSession]] = None,\n auto_context_on_execute: bool = False,\n auto_context_force_transaction: bool = False,\n)\n```\n\n---\n\n### Parameters:\n\n- `engine` - SQLAlchemy `AsyncEngine` instance.\n- `default_session_maker` - Optional async session factory. If omitted, uses `async_sessionmaker(engine)`.\n- `auto_context_on_execute` - If `True`, allows `.execute()` to run even without an active session\nby creating a temporary one.\n- `auto_context_force_transaction` - If `True`, `.execute()` always runs inside a transaction when auto context is used.\nIf `False`, it uses `.session()` for read-only queries (like `Select` or `CompoundSelect`),\nand `.transaction()` for everything else, including `Insert`, `Update`, and raw SQL.\n\n---\n\n### Session Methods:\n\n- `session(...) -> AsyncIterator[AsyncSession]` - Enter a new session context, or reuse an existing\none if `reuse_if_exists=True`.\n- `transaction(...) -> AsyncIterator[AsyncSession]` - Enter a transactional context.\nWill nest if a transaction is already active.\n- `new_session(...) -> AsyncIterator[AsyncSession]` - Create a new isolated session, even if another is already active.\nOverrides the context for the duration.\n- `new_transaction(...) -> AsyncIterator[AsyncSession]` - Create a new transaction in an isolated session.\n- `get_session(strict: bool = True) -> AsyncSession | None` - Return the current session from context.\nRaises `NoSessionError` if `strict=True` and no session exists.\n- `execute(...) -> Result` - Execute a SQLAlchemy `Executable` using the current or temporary context.\nUses the current session if one is active. Otherwise, behavior depends on `auto_context_on_execute` -\na new session or transaction context may be created automatically.\n\n---\n\n## Auto-context Example\n\n```python\n# Opens a temporary session or transaction depending on statement type\ndb = SQLAlchemyTransactionContext(engine, auto_context_on_execute=True)\n\nawait db.execute(insert(User).values(name=\"Alice\"))\nawait db.execute(select(User))\n```\n\n---\n\n## Full Example\n\nFor a complete working example using PostgreSQL, see\n[`example/`](https://github.com/QuisEgoSum/sqlalchemy-tx-context/tree/main/example).\nIt demonstrates table creation, data insertion, transactional rollback, and querying.\n\n---\n\n## Exceptions\n\nThis library defines a few custom exceptions to help catch context-related mistakes:\n\n- `NoSessionError`: Raised when attempting to access or use a session when none is active and fallback is disabled.\n- `SessionAlreadyActiveError`: Raised when entering `.session()` while a session is already active \n(unless `reuse_if_exists=True`).\n- `TransactionAlreadyActiveError`: Raised when entering `.transaction()` while a transaction is already active \nand nesting is disabled.\n\n---\n\n## Motivation\n\nSQLAlchemy does not provide an out-of-the-box solution for context-local session tracking when working with\n`AsyncSession`. Passing sessions around explicitly can make service-layer code verbose and harder to maintain.\n\nThe library introduces a lightweight and predictable abstraction that:\n\n- Stores current session in a `ContextVar`.\n- Provides safe transactional/session boundaries.\n- Exposes a unified `execute(...)` interface.\n- Integrates with standard SQLAlchemy models, statements, and engines.\n\n---\n\n## When should you use this?\n\n- You're writing service-layer logic and want to avoid passing `session` explicitly.\n- You need nested transactional logic with clean context boundaries.\n- You prefer explicit context management over dependency-injected sessions.\n- You work with context-local transaction boundaries in background tasks or microservices.\n- You need precise control over session lifecycle and scope.\n\nThis library is best suited for functional or script-style code where sessions are not injected via DI frameworks.\n\n---\n\n## Best Practices\n\n- Use `.transaction()` or `.session()` explicitly in your service-layer or job code\nto clearly define execution boundaries.\n- `.execute(...)` is ideal for small projects, functional-style code, or early-stage prototypes - it avoids\nboilerplate and makes code fast to write.\n- As your project grows, you can gradually migrate to DI-based session and transaction management:\njust replace `db.execute(...)` with `self._session.execute(...)`\nand `db.transaction()` with your own context (e.g., `UnitOfWork`, `session.begin()`) - the query logic remains the same.\n- This makes the library especially useful for bootstrapping or background scripts,\nwhere a full-blown DI setup would be overkill.\n\n---\n\n## Tests\n\n```shell\npytest --cov=sqlalchemy_tx_context --cov-report=term-missing\n```\n\n---\n\n## Compatibility\n\nTested on Python `3.9` - `3.12`.\n\n---\n\n## License\n\nMIT License\n",
"bugtrack_url": null,
"license": null,
"summary": "Context-local session manager for SQLAlchemy AsyncSession",
"version": "1.0.1",
"project_urls": {
"Documentation": "https://github.com/QuisEgoSum/sqlalchemy-tx-context#readme",
"Homepage": "https://github.com/QuisEgoSum/sqlalchemy-tx-context",
"Repository": "https://github.com/QuisEgoSum/sqlalchemy-tx-context"
},
"split_keywords": [
"sqlalchemy",
" async",
" contextvars",
" transaction",
" session"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "7e6d0031d11d1e69c0573afdd79b7b65cc3a784d7a00b0dcf89c330cac735d32",
"md5": "7325a51a5813b18726b608a9d35d8729",
"sha256": "13bcdec1fac24643ac0cbe30909c385dfaa9144809cc5e483c048cd36d72a8fc"
},
"downloads": -1,
"filename": "sqlalchemy_tx_context-1.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7325a51a5813b18726b608a9d35d8729",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 9193,
"upload_time": "2025-06-17T06:06:45",
"upload_time_iso_8601": "2025-06-17T06:06:45.744411Z",
"url": "https://files.pythonhosted.org/packages/7e/6d/0031d11d1e69c0573afdd79b7b65cc3a784d7a00b0dcf89c330cac735d32/sqlalchemy_tx_context-1.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "7825ad45257d084012fe10127e9bc0cb56eb30230585bd8eaa638dbfd1916ba7",
"md5": "430b947596232757a8a18c90d3758b99",
"sha256": "dc990efda9555035db5c9949823cf50e5b89a9de9a5355c046180d7de15c245b"
},
"downloads": -1,
"filename": "sqlalchemy_tx_context-1.0.1.tar.gz",
"has_sig": false,
"md5_digest": "430b947596232757a8a18c90d3758b99",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 10969,
"upload_time": "2025-06-17T06:06:47",
"upload_time_iso_8601": "2025-06-17T06:06:47.171689Z",
"url": "https://files.pythonhosted.org/packages/78/25/ad45257d084012fe10127e9bc0cb56eb30230585bd8eaa638dbfd1916ba7/sqlalchemy_tx_context-1.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-06-17 06:06:47",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "QuisEgoSum",
"github_project": "sqlalchemy-tx-context#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "sqlalchemy-tx-context"
}
JSON
