# SQLAlchemy Transaction Context
An extension for sqlalchemy to store the session object in context and simplify the retrieval of the query result.
## Create instance
```python
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy_tx_context import SQLAlchemyTransactionContext
engine = create_async_engine(...)
db = SQLAlchemyTransactionContext(engine)
```
## Execute query
`SQLAlchemyTransactionContext` contains the following methods for creating queries with execution
through a context session:
- `select`
- `insert`
- `update`
- `delete`
- `union`
- `union_all`
- `exists`
Query will use a session from the context.
```python
async def some_repository_method():
await db.insert(...).values(...).execute()
async def some_function():
async with db.transaction() as tx:
await some_repository_method()
await tx.rollback() # Record will not be inserted
```
If there is no session in the context, a default session will be created to execute a single query.
```python
async def some_repository_method():
await db.insert(...).values(...).execute()
async def some_function():
await some_repository_method()
```
Calling the transaction method inside the session will create a nested transaction.
```python
async def some_function():
async with db.transaction() as tx1:
isinstance(tx1, AsyncSession) # True
async with db.transaction() as tx2:
isinstance(tx2, AsyncSessionTransaction) # True
```
## Execute query with proxy methods
List of proxy methods added to the request object
Proxy methods for the `AsyncSession` properties:
- `execute`
- `scalar`
- `scalars`
Example:
```python
value = await db.select(...).execute()
```
The same as:
```python
async with async_sessionmaker(engine).begin() as tx:
result = await tx.execute(select(...))
```
Proxy methods for the `Result` properties:
- `first`
- `all`
Example:
```python
value = await db.select(...).first()
```
The same as:
```python
async with async_sessionmaker(engine).begin() as tx:
result = await tx.execute(select(...))
value = result.first()
```
Proxy methods for the `MappingResult` properties:
- `mapped_first`
- `mapped_one`
- `mapped_all`
Example:
```python
value = await db.select(...).mapped_first()
```
The same as:
```python
async with async_sessionmaker(engine).begin() as tx:
result = await tx.execute(select(...))
value = result.mappings().first()
```
Proxy method for the `CursorResult`:
- `rowcount`
Raw data
{
"_id": null,
"home_page": "https://github.com/QuisEgoSum/sqlalchemy-tx-context",
"name": "sqlalchemy-tx-context",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "sqlalchemy",
"author": "QuisEgoSum",
"author_email": "subbotin.evdokim@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/72/c1/a506e3157b5382688273b9d93737b157499834b74849d2eb65bb7b4313e9/sqlalchemy_tx_context-0.4.3.tar.gz",
"platform": null,
"description": "\n# SQLAlchemy Transaction Context\n\nAn extension for sqlalchemy to store the session object in context and simplify the retrieval of the query result.\n\n\n## Create instance \n```python\nfrom sqlalchemy.ext.asyncio import create_async_engine\nfrom sqlalchemy_tx_context import SQLAlchemyTransactionContext\n\n\nengine = create_async_engine(...)\ndb = SQLAlchemyTransactionContext(engine)\n```\n\n## Execute query\n\n`SQLAlchemyTransactionContext` contains the following methods for creating queries with execution\nthrough a context session:\n- `select`\n- `insert`\n- `update`\n- `delete`\n- `union`\n- `union_all`\n- `exists`\n\nQuery will use a session from the context.\n\n```python\nasync def some_repository_method():\n await db.insert(...).values(...).execute()\n\n\nasync def some_function():\n async with db.transaction() as tx:\n await some_repository_method()\n await tx.rollback() # Record will not be inserted\n```\n\n\nIf there is no session in the context, a default session will be created to execute a single query.\n\n```python\nasync def some_repository_method():\n await db.insert(...).values(...).execute()\n\nasync def some_function():\n await some_repository_method()\n```\n\n\nCalling the transaction method inside the session will create a nested transaction.\n\n```python\nasync def some_function():\n async with db.transaction() as tx1:\n isinstance(tx1, AsyncSession) # True\n async with db.transaction() as tx2:\n isinstance(tx2, AsyncSessionTransaction) # True\n```\n\n## Execute query with proxy methods\n\nList of proxy methods added to the request object\n\nProxy methods for the `AsyncSession` properties:\n- `execute`\n- `scalar`\n- `scalars`\n\nExample:\n\n```python\nvalue = await db.select(...).execute()\n```\n\nThe same as:\n\n```python\nasync with async_sessionmaker(engine).begin() as tx:\n result = await tx.execute(select(...))\n```\n\nProxy methods for the `Result` properties:\n- `first`\n- `all`\n\nExample:\n```python\nvalue = await db.select(...).first()\n```\n\nThe same as:\n\n```python\nasync with async_sessionmaker(engine).begin() as tx:\n result = await tx.execute(select(...))\n value = result.first()\n```\n\nProxy methods for the `MappingResult` properties:\n\n- `mapped_first`\n- `mapped_one`\n- `mapped_all`\n\nExample:\n```python\nvalue = await db.select(...).mapped_first()\n```\n\nThe same as:\n\n```python\nasync with async_sessionmaker(engine).begin() as tx:\n result = await tx.execute(select(...))\n value = result.mappings().first()\n```\n\nProxy method for the `CursorResult`:\n\n- `rowcount`\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "An extension for sqlalchemy",
"version": "0.4.3",
"project_urls": {
"Homepage": "https://github.com/QuisEgoSum/sqlalchemy-tx-context"
},
"split_keywords": [
"sqlalchemy"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0fd707f4e9f4082eb952e3f831b1eddeac485e2bc9e5d9c65a2c7773197ceaf9",
"md5": "eebf755e36e9a9e60d93a9599ce80ba7",
"sha256": "4f3c0d92fbe5741b03d2d9256c90dfd8ff95407baa82dd3990e56e114be321f7"
},
"downloads": -1,
"filename": "sqlalchemy_tx_context-0.4.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "eebf755e36e9a9e60d93a9599ce80ba7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 13632,
"upload_time": "2025-03-09T08:09:05",
"upload_time_iso_8601": "2025-03-09T08:09:05.317930Z",
"url": "https://files.pythonhosted.org/packages/0f/d7/07f4e9f4082eb952e3f831b1eddeac485e2bc9e5d9c65a2c7773197ceaf9/sqlalchemy_tx_context-0.4.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "72c1a506e3157b5382688273b9d93737b157499834b74849d2eb65bb7b4313e9",
"md5": "e3842d17f10f39f6f0ffdffb470de36f",
"sha256": "53f1bde8c82bea058614dd0d28ad7d145d90cfecf8972c589c03788632e98430"
},
"downloads": -1,
"filename": "sqlalchemy_tx_context-0.4.3.tar.gz",
"has_sig": false,
"md5_digest": "e3842d17f10f39f6f0ffdffb470de36f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 8000,
"upload_time": "2025-03-09T08:09:06",
"upload_time_iso_8601": "2025-03-09T08:09:06.827395Z",
"url": "https://files.pythonhosted.org/packages/72/c1/a506e3157b5382688273b9d93737b157499834b74849d2eb65bb7b4313e9/sqlalchemy_tx_context-0.4.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-03-09 08:09:06",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "QuisEgoSum",
"github_project": "sqlalchemy-tx-context",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "asyncpg",
"specs": [
[
"==",
"0.29.0"
]
]
},
{
"name": "SQLAlchemy",
"specs": [
[
"==",
"2.0.29"
]
]
},
{
"name": "contextvars",
"specs": [
[
"==",
"2.4"
]
]
}
],
"lcname": "sqlalchemy-tx-context"
}
JSON
