fix: internal rollback() corrupts caller's ambient transaction and discards uncommitted work (#339)

src/fastapi_toolsets/db/watch.py

@@ -57,38 +57,50 @@ async def wait_for_row_change(
         )
         ```
     """
+    bind = getattr(session, "bind", None)
+    if bind is None:
+        raise TypeError(
+            "wait_for_row_change requires a session bound to an engine "
+            "(session.bind is None)"
+        )
+    watcher = AsyncSession(bind=bind)
+    try:
 
-    async def _reload() -> _M | None:
+        async def _reload() -> _M | None:
-        await session.rollback()
+            await watcher.rollback()
-        return await session.get(model, pk_value, populate_existing=True)
+            return await watcher.get(model, pk_value, populate_existing=True)
-
-    instance = await _reload()
-    if instance is None:
-        raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
-
-    if columns is not None:
-        watch_cols = columns
-    else:
-        watch_cols = [attr.key for attr in model.__mapper__.column_attrs]
-
-    initial = {col: getattr(instance, col) for col in watch_cols}
-
-    elapsed = 0.0
-    while True:
-        await asyncio.sleep(interval)
-        elapsed += interval
-
-        if timeout is not None and elapsed >= timeout:
-            raise TimeoutError(
-                f"No change detected on {model.__name__} "
-                f"with pk={pk_value!r} within {timeout}s"
-            )
 
         instance = await _reload()
-
         if instance is None:
-            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} was deleted")
+            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
 
-        current = {col: getattr(instance, col) for col in watch_cols}
+        if columns is not None:
-        if current != initial:
+            watch_cols = columns
-            return instance
+        else:
+            watch_cols = [attr.key for attr in model.__mapper__.column_attrs]
+
+        initial = {col: getattr(instance, col) for col in watch_cols}
+
+        elapsed = 0.0
+        while True:
+            await asyncio.sleep(interval)
+            elapsed += interval
+
+            if timeout is not None and elapsed >= timeout:
+                raise TimeoutError(
+                    f"No change detected on {model.__name__} "
+                    f"with pk={pk_value!r} within {timeout}s"
+                )
+
+            instance = await _reload()
+
+            if instance is None:
+                raise NotFoundError(
+                    f"{model.__name__} with pk={pk_value!r} was deleted"
+                )
+
+            current = {col: getattr(instance, col) for col in watch_cols}
+            if current != initial:
+                return instance
+    finally:
+        await watcher.close()

tests/test_db.py

@@ -689,6 +689,13 @@ class TestWaitForRowChange:
         with pytest.raises(NotFoundError, match="not found"):
             await wait_for_row_change(db_session, Role, fake_id, interval=0.05)
 
+    @pytest.mark.anyio
+    async def test_unbound_session_raises_type_error(self):
+        """Raises TypeError when the session has no bind to open a watcher on."""
+        unbound = AsyncSession()
+        with pytest.raises(TypeError, match="requires a session bound to an engine"):
+            await wait_for_row_change(unbound, Role, uuid.uuid4())
+
     @pytest.mark.anyio
     async def test_timeout_raises(self, db_session: AsyncSession):
         """Raises TimeoutError when no change is detected within timeout."""
@@ -781,6 +788,43 @@ class TestWaitForRowChange:
             await wait_for_row_change(db_session, Role, role.id, interval=0.05)
         await delete_task
 
+    @pytest.mark.anyio
+    async def test_does_not_disturb_ambient_transaction(
+        self, db_session: AsyncSession, engine
+    ):
+        """A read-only ambient transaction around the call survives untouched."""
+        role = Role(name="ambient_role")
+        db_session.add(role)
+        await db_session.commit()
+
+        async def update_later():
+            await asyncio.sleep(0.15)
+            factory = async_sessionmaker(engine, expire_on_commit=False)
+            async with factory() as other:
+                r = await other.get(Role, role.id)
+                assert r is not None
+                r.name = "ambient_updated"
+                await other.commit()
+
+        update_task = asyncio.create_task(update_later())
+        async with transaction(db_session):
+            # A read before the watch, establishing an ambient transaction
+            # that must remain usable once wait_for_row_change returns.
+            await db_session.get(Role, role.id)
+            result = await wait_for_row_change(
+                db_session, Role, role.id, interval=0.05, timeout=2.0
+            )
+            await update_task
+            assert result.name == "ambient_updated"
+            # The ambient transaction must still be open and usable here.
+            assert db_session.in_transaction()
+            other_role = Role(name="added_within_ambient_tx")
+            db_session.add(other_role)
+
+        # transaction() committed cleanly on exit; the write above landed.
+        check = await db_session.get(Role, other_role.id)
+        assert check is not None
+
 
 class TestCreateDatabase:
     """Tests for create_database."""