diff --git a/src/fastapi_toolsets/db/watch.py b/src/fastapi_toolsets/db/watch.py index b7ab518..1941d36 100644 --- a/src/fastapi_toolsets/db/watch.py +++ b/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: - await session.rollback() - return await session.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" - ) + async def _reload() -> _M | None: + await watcher.rollback() + 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} was deleted") - - current = {col: getattr(instance, col) for col in watch_cols} - if current != initial: - return instance + 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" + ) + + current = {col: getattr(instance, col) for col in watch_cols} + if current != initial: + return instance + finally: + await watcher.close() diff --git a/tests/test_db.py b/tests/test_db.py index 609c8a1..c6bb789 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -689,6 +689,13 @@ async def test_nonexistent_row_raises(self, db_session: AsyncSession): 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 @@ async def delete_later(): 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."""