mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-08-04 15:44:09 +00:00
fix: 'Could not refresh instance' when using EventSession (#341)
This commit is contained in:
@@ -204,6 +204,11 @@ async def _invoke_callback(
|
|||||||
await result
|
await result
|
||||||
|
|
||||||
|
|
||||||
|
async def _reload_if_present(session: AsyncSession, obj: Any, state: Any) -> None:
|
||||||
|
"""Re-populate *obj* from the DB if its row still exists."""
|
||||||
|
await session.get(type(obj), state.key[1], populate_existing=True)
|
||||||
|
|
||||||
|
|
||||||
class EventSession(AsyncSession):
|
class EventSession(AsyncSession):
|
||||||
"""AsyncSession subclass that dispatches lifecycle callbacks after commit."""
|
"""AsyncSession subclass that dispatches lifecycle callbacks after commit."""
|
||||||
|
|
||||||
@@ -253,7 +258,7 @@ class EventSession(AsyncSession):
|
|||||||
state is None or state.detached or state.transient
|
state is None or state.detached or state.transient
|
||||||
): # pragma: no cover
|
): # pragma: no cover
|
||||||
continue
|
continue
|
||||||
await self.refresh(obj)
|
await _reload_if_present(self, obj, state)
|
||||||
for handler in _get_handlers(type(obj), ModelEvent.CREATE):
|
for handler in _get_handlers(type(obj), ModelEvent.CREATE):
|
||||||
await _invoke_callback(handler, obj, ModelEvent.CREATE, None)
|
await _invoke_callback(handler, obj, ModelEvent.CREATE, None)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -277,7 +282,7 @@ class EventSession(AsyncSession):
|
|||||||
state is None or state.detached or state.transient
|
state is None or state.detached or state.transient
|
||||||
): # pragma: no cover
|
): # pragma: no cover
|
||||||
continue
|
continue
|
||||||
await self.refresh(obj)
|
await _reload_if_present(self, obj, state)
|
||||||
for handler in _get_handlers(type(obj), ModelEvent.UPDATE):
|
for handler in _get_handlers(type(obj), ModelEvent.UPDATE):
|
||||||
await _invoke_callback(handler, obj, ModelEvent.UPDATE, changes)
|
await _invoke_callback(handler, obj, ModelEvent.UPDATE, changes)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
|
|||||||
+52
-1
@@ -21,12 +21,12 @@ from fastapi_toolsets.models import (
|
|||||||
listens_for,
|
listens_for,
|
||||||
)
|
)
|
||||||
from fastapi_toolsets.models.watched import (
|
from fastapi_toolsets.models.watched import (
|
||||||
EventSession,
|
|
||||||
_EVENT_HANDLERS,
|
_EVENT_HANDLERS,
|
||||||
_SESSION_CREATES,
|
_SESSION_CREATES,
|
||||||
_SESSION_DELETES,
|
_SESSION_DELETES,
|
||||||
_SESSION_UPDATES,
|
_SESSION_UPDATES,
|
||||||
_WATCHED_MODELS,
|
_WATCHED_MODELS,
|
||||||
|
EventSession,
|
||||||
_after_flush,
|
_after_flush,
|
||||||
_after_rollback,
|
_after_rollback,
|
||||||
_get_watched_fields,
|
_get_watched_fields,
|
||||||
@@ -1001,6 +1001,57 @@ class TestEventCallbacks:
|
|||||||
|
|
||||||
assert _test_events == []
|
assert _test_events == []
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_create_survives_row_deleted_before_reload(self, mixin_session):
|
||||||
|
"""A row deleted by another transaction right after commit still fires CREATE."""
|
||||||
|
keep = WatchedModel(status="active", other="x")
|
||||||
|
doomed = WatchedModel(status="active", other="x")
|
||||||
|
mixin_session.add_all([keep, doomed])
|
||||||
|
await mixin_session.flush()
|
||||||
|
doomed_id = doomed.id
|
||||||
|
|
||||||
|
raced = {"done": False}
|
||||||
|
|
||||||
|
async def kill_doomed_row_once():
|
||||||
|
if raced["done"]:
|
||||||
|
return
|
||||||
|
raced["done"] = True
|
||||||
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
||||||
|
async with async_sessionmaker(engine)() as other:
|
||||||
|
row = await other.get(WatchedModel, doomed_id)
|
||||||
|
await other.delete(row)
|
||||||
|
await other.commit()
|
||||||
|
await engine.dispose()
|
||||||
|
|
||||||
|
real_get = mixin_session.get
|
||||||
|
real_refresh = mixin_session.refresh
|
||||||
|
|
||||||
|
def _matches_doomed(pk):
|
||||||
|
return pk == doomed_id or (isinstance(pk, tuple) and pk[0] == doomed_id)
|
||||||
|
|
||||||
|
async def racing_get(model, pk, *args, **kwargs):
|
||||||
|
if _matches_doomed(pk):
|
||||||
|
await kill_doomed_row_once()
|
||||||
|
return await real_get(model, pk, *args, **kwargs)
|
||||||
|
|
||||||
|
async def racing_refresh(obj, *args, **kwargs):
|
||||||
|
if getattr(obj, "id", None) == doomed_id:
|
||||||
|
await kill_doomed_row_once()
|
||||||
|
return await real_refresh(obj, *args, **kwargs)
|
||||||
|
|
||||||
|
# Patch both possible reload mechanisms (session.get / session.refresh)
|
||||||
|
# so this test still exercises the race regardless of which one
|
||||||
|
# EventSession.commit() uses internally to pick up server defaults.
|
||||||
|
mixin_session.get = racing_get
|
||||||
|
mixin_session.refresh = racing_refresh
|
||||||
|
with patch.object(_watched_module._logger, "error") as mock_error:
|
||||||
|
await mixin_session.commit()
|
||||||
|
mock_error.assert_not_called()
|
||||||
|
|
||||||
|
assert raced["done"]
|
||||||
|
created_ids = {e["obj_id"] for e in _test_events if e["event"] == "create"}
|
||||||
|
assert created_ids == {keep.id, doomed_id}
|
||||||
|
|
||||||
|
|
||||||
class TestTransientObject:
|
class TestTransientObject:
|
||||||
"""Create + delete within the same transaction should fire no events."""
|
"""Create + delete within the same transaction should fire no events."""
|
||||||
|
|||||||
Reference in New Issue
Block a user