Session¶
The session API provides a small async unit-of-work wrapper around an Ormdantic database instance.
Use async with db.session() for commit-on-success and rollback-on-error flows. The session tracks pending inserts, dirty loaded models, deleted models, relationship additions, and an identity map for repeated loads.
Session.savepoint(name=None) opens a nested savepoint. If the block raises, Ormdantic rolls back to the database savepoint and restores the session's pending, dirty, deleted, identity, and snapshot state.
If flush() fails, the session restores the pre-flush unit-of-work state and requires rollback() before more work is accepted.
Session(database, *, transaction_options=None)
Minimal async unit-of-work session for Ormdantic models.
Create a session bound to an Ormdantic database instance.
Source code in ormdantic/session.py
def __init__(
self, database: Any, *, transaction_options: Any | None = None
) -> None:
"""Create a session bound to an `Ormdantic` database instance."""
self._database = database
self._transaction_options = transaction_options
self._new: list[BaseModel] = []
self._dirty: list[BaseModel] = []
self._deleted: list[BaseModel] = []
self._identity_map: dict[tuple[type[BaseModel], Any], BaseModel] = {}
self._snapshots: dict[tuple[type[BaseModel], Any], dict[str, Any]] = {}
self._failed_flush_error: Exception | None = None
self._savepoint_sequence = 0
self._closed = False
add(model)
Stage a new model for insertion on flush.
Source code in ormdantic/session.py
def add(self, model: BaseModel) -> None:
"""Stage a new model for insertion on flush."""
self._ensure_usable()
self._cascade_add(model, set())
mark_dirty(model)
Stage an existing model for update on flush.
Source code in ormdantic/session.py
def mark_dirty(self, model: BaseModel) -> None:
"""Stage an existing model for update on flush."""
self._ensure_usable()
if model not in self._new and model not in self._dirty:
self._dirty.append(model)
delete(model)
Stage an existing model for deletion on flush.
Source code in ormdantic/session.py
def delete(self, model: BaseModel) -> None:
"""Stage an existing model for deletion on flush."""
self._ensure_usable()
self._cascade_delete(model, set())
merge(model)
Merge a detached model into the identity map and stage it as dirty.
Source code in ormdantic/session.py
def merge(self, model: BaseModel) -> BaseModel:
"""Merge a detached model into the identity map and stage it as dirty."""
self._ensure_usable()
table = self._database._table_map.model_to_data[type(model)]
key = (type(model), getattr(model, table.pk))
if cached := self._identity_map.get(key):
for field, value in model.__dict__.items():
setattr(cached, field, value)
self.mark_dirty(cached)
return cached
if staged := self._staged_model_for_key(key):
for field, value in model.__dict__.items():
setattr(staged, field, value)
if staged not in self._new:
self.mark_dirty(staged)
return staged
self._remember(model)
self.mark_dirty(model)
return model
expire(model)
Remove a model from the identity map.
Source code in ormdantic/session.py
def expire(self, model: BaseModel) -> None:
"""Remove a model from the identity map."""
self._ensure_usable()
key = self._identity_key(model)
self._identity_map.pop(key, None)
self._snapshots.pop(key, None)
async
¶
flush()
Write staged inserts and updates without ending the transaction.
Source code in ormdantic/session.py
async def flush(self) -> None:
"""Write staged inserts and updates without ending the transaction."""
self._ensure_usable()
state = self._capture_state()
try:
await self._database._events.dispatch("before_flush", session=self)
self._detect_relationship_changes()
for model in self._detect_dirty_models():
self.mark_dirty(model)
inserted = self._dependency_ordered(list(self._new))
for batch in self._model_batches(inserted):
stored_models = await self._database[type(batch[0])].insert_many(batch)
for stored in stored_models:
self._remember(stored)
self._new.clear()
updated = list(self._dirty)
for batch in self._model_batches(updated):
for model in batch:
stored = await self._database[type(model)].update(model)
self._remember(stored)
self._dirty.clear()
deleted = list(reversed(self._dependency_ordered(list(self._deleted))))
for batch in self._model_batches(deleted):
for model in batch:
key = self._identity_key(model)
pk = key[1]
await self._database[type(model)].delete(pk)
self._identity_map.pop(key, None)
self._snapshots.pop(key, None)
self._deleted.clear()
except Exception as exc:
self._restore_state(state)
self._failed_flush_error = exc
raise
await self._database._events.dispatch("after_flush", session=self)
async
¶
commit()
Flush changes and commit the active transaction.
Source code in ormdantic/session.py
async def commit(self) -> None:
"""Flush changes and commit the active transaction."""
if self._closed:
return
await self.flush()
await self._database._commit()
self._closed = True
async
¶
rollback()
Discard staged changes and roll back the active transaction.
Source code in ormdantic/session.py
async def rollback(self) -> None:
"""Discard staged changes and roll back the active transaction."""
if self._closed:
return
try:
await self._database._rollback()
finally:
self._new.clear()
self._dirty.clear()
self._deleted.clear()
self._identity_map.clear()
self._snapshots.clear()
self._failed_flush_error = None
self._closed = True
savepoint(name=None)
Open a nested session savepoint that restores unit-of-work state.
Source code in ormdantic/session.py
def savepoint(self, name: str | None = None) -> Any:
"""Open a nested session savepoint that restores unit-of-work state."""
self._ensure_usable()
if name is None:
self._savepoint_sequence += 1
name = f"session_sp_{self._savepoint_sequence}"
return _SessionSavepoint(self, name)
async
¶
refresh(model, *, depth=0)
Reload a model by primary key and remember the refreshed instance.
Source code in ormdantic/session.py
async def refresh(self, model: BaseModel, *, depth: int = 0) -> BaseModel | None:
"""Reload a model by primary key and remember the refreshed instance."""
self._ensure_usable()
table = self._database._table_map.model_to_data[type(model)]
refreshed = await self._database[type(model)].find_one(
getattr(model, table.pk), depth=depth
)
if refreshed is not None:
self._remember(refreshed)
return refreshed
get_cached(model_type, pk)
Return a model from the identity map if it has been remembered.
Source code in ormdantic/session.py
def get_cached(self, model_type: type[BaseModel], pk: Any) -> BaseModel | None:
"""Return a model from the identity map if it has been remembered."""
return self._identity_map.get((model_type, pk))
async
¶
get(model_type, pk, *, depth=0)
Return a cached model or load it by primary key.
Source code in ormdantic/session.py
async def get(
self, model_type: type[BaseModel], pk: Any, *, depth: int = 0
) -> BaseModel | None:
"""Return a cached model or load it by primary key."""
self._ensure_usable()
if cached := self.get_cached(model_type, pk):
return cached
loaded = await self._database[model_type].find_one(pk, depth=depth)
if loaded is not None:
self._remember(loaded)
return loaded