feature: add transaction and session contexts

README.md

@@ -99,6 +99,8 @@ results = (
 - Direct `session` — use your own session, commit manually
 - Context manager — full lifecycle control with auto rollback
 - `with_session()` — override session per query
+- `transaction()` — one transaction shared by commands across models
+- `session_context()` — one shared session with manual commit, rollback, and flush
 - Works with Flask, FastAPI, and any framework
 
 **Async**
@@ -296,6 +298,120 @@ item = qm.get(id=1)
 > Accessing lazy relationship attributes on detached objects raises `DetachedInstanceError`.
 > Use `select_related()` or `prefetch_related()` to load relationships upfront, or use a direct `session`.
 
+##### Multi-model transaction
+
+Use `transaction()` when commands for different models must succeed or fail
+together. Pass a `sessionmaker`; the context creates and closes the session itself.
+Every query manager inside the block automatically uses that session, so individual
+commands do not need a `session=` argument.
+
+```python
+from sqlalchemy_query_manager import transaction
+
+with transaction(Session):
+    owner = Owner.query_manager.create(
+        first_name="John",
+        last_name="Doe",
+    )
+    group = Group.query_manager.create(
+        name="Backend",
+        owner_id=owner.id,
+    )
+    Item.query_manager.create(
+        name="First task",
+        group_id=group.id,
+    )
+```
+
+The outer transaction commits on normal exit and rolls back on an exception. Manual
+`commit()` and `rollback()` are intentionally unavailable because they would break
+the atomic boundary. Explicit `flush()` remains available:
+
+```python
+with transaction(Session) as control:
+    item = Item.query_manager.create(name="Needs an id")
+    control.flush()
+    print(item.id)
+```
+
+Nested transaction blocks reuse the same session and create a database SAVEPOINT.
+The nested transaction does not need the session source again:
+
+```python
+with transaction(Session):
+    Owner.query_manager.create(first_name="John", last_name="Doe")
+
+    try:
+        with transaction():
+            Group.query_manager.create(name="Rolled back")
+            raise ValueError("cancel group")
+    except ValueError:
+        pass
+
+    Item.query_manager.create(name="Still committed")
+```
+
+When the nested block fails, only its SAVEPOINT is rolled back. If the exception
+leaves the outer block, the complete transaction is rolled back.
+
+##### Manual session context
+
+Use `session_context()` for commit-as-you-go workflows. It creates one ambient session
+for all models but leaves transaction boundaries under your control.
+
+```python
+from sqlalchemy_query_manager import session_context
+
+with session_context(Session) as work:
+    owner = Owner.query_manager.create(
+        first_name="John",
+        last_name="Doe",
+    )
+    Group.query_manager.create(name="Committed", owner_id=owner.id)
+
+    work.flush()
+    work.commit()
+
+    Item.query_manager.create(name="Discarded")
+    work.rollback()
+```
+
+After `commit()` or `rollback()`, the next database command starts a new transaction
+in the same session. For a session created by the context, any transaction still
+pending on exit is rolled back; the context never commits implicitly.
+
+Both APIs also accept an existing `Session` or a session context-manager factory.
+An existing session is borrowed and is never committed, rolled back, or closed by
+`session_context()` on exit; its lifecycle remains the caller's responsibility.
+
+##### Async contexts
+
+The same API works with an async sessionmaker. Context and control methods become
+awaitable, while all models still share one `AsyncSession`:
+
+```python
+async with transaction(AsyncSessionMaker) as control:
+    owner = await Owner.query_manager.create(
+        first_name="John",
+        last_name="Doe",
+    )
+    await Group.query_manager.create(name="Backend", owner_id=owner.id)
+    await control.flush()
+
+async with session_context(AsyncSessionMaker) as work:
+    await Item.query_manager.create(name="Committed manually")
+    await work.commit()
+```
+
+An `AsyncSession` cannot be shared by concurrent tasks. Do not run query-manager
+commands from one context through `asyncio.gather()` or child tasks. Open a separate
+context with its own sessionmaker inside each concurrent task instead.
+
+A synchronous `Session` likewise cannot be shared across threads. Each worker thread
+must open its own `transaction(SessionMaker)` or `session_context(SessionMaker)`.
+If a `ContextVar` is explicitly propagated to another thread, the package detects
+the ownership mismatch and raises instead of reusing the ambient session.
+
 ---
 
 #### Filtering

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "sqlalchemy-query-manager"
-version = "0.6.1"
+version = "0.7.0"
 description = "Django-style ORM interface for SQLAlchemy: Q filters, eager loading, bulk operations, and native async support."
 authors = ["ViAchKoN"]
 readme = "README.md"

sqlalchemy_query_manager/__init__.py

@@ -1,6 +1,7 @@
 """Public API for SQLAlchemy Query Manager."""
 
 from sqlalchemy_query_manager.core.async_query_manager import AsyncQueryManager
+from sqlalchemy_query_manager.core.contexts import session_context, transaction
 from sqlalchemy_query_manager.core.exceptions import (
     DoesNotExist,
     MultipleObjectsReturned,
@@ -28,5 +29,7 @@
     "MultipleObjectsReturned",
     "Q",
     "QueryManager",
+    "session_context",
     "Sum",
+    "transaction",
 ]

sqlalchemy_query_manager/core/async_query_manager.py

@@ -3,12 +3,15 @@
 from sqlalchemy import delete, func, inspect, select, text, update
 from sqlalchemy.orm import sessionmaker
 
+from sqlalchemy_query_manager.core.contexts import get_current_context
 from sqlalchemy_query_manager.core.helpers import AggregateFunc
 from sqlalchemy_query_manager.core.sync_query_manager import QueryManager
 from sqlalchemy_query_manager.core.utils import get_async_session
 
 
 class AsyncQueryManager(QueryManager):
+    def _should_commit(self):
+        return isinstance(self.session, sessionmaker) and get_current_context() is None
 
     @get_async_session
     async def first(self, session=None):
@@ -104,7 +107,7 @@ async def create(self, session=None, **kwargs):
         new_obj = self.ConverterConfig.model(**kwargs)
         session.add(new_obj)
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()
@@ -125,7 +128,7 @@ async def bulk_create(
         objects = [self.ConverterConfig.model(**item) for item in data]
         session.add_all(objects)
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()
@@ -193,7 +196,7 @@ async def update(self, session=None, expunge=True, **kwargs):
         )
         returned_pks = [row[0] for row in result]
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()
@@ -217,7 +220,7 @@ async def _update_legacy_path(self, session, expunge=True, **kwargs):
 
         await session.execute(update_query)
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()
@@ -256,7 +259,7 @@ async def update_raw(self, session=None, **kwargs):
 
         result = await session.execute(update_query)
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()
@@ -279,7 +282,7 @@ async def update_or_create(self, session=None, defaults=None, **kwargs):
                     if hasattr(existing, key):
                         setattr(existing, key, value)
 
-            if isinstance(self.session, sessionmaker):
+            if self._should_commit():
                 await session.commit()
             else:
                 await session.flush()
@@ -349,7 +352,7 @@ async def delete(self, session=None, synchronize_session=True):
             delete_query, execution_options={"synchronize_session": False}
         )
 
-        if isinstance(self.session, sessionmaker):
+        if self._should_commit():
             await session.commit()
         else:
             await session.flush()