UnitOfWork
API reference for the UnitOfWork class managing atomic transactions, lifecycle boundaries, and deferred domain events.
The UnitOfWork (UoW) coordinates transactional boundaries and buffers domain events, guaranteeing that events are dispatched only after database modifications have successfully committed. It implements the asynchronous context manager protocol and supports depth-aware nested execution.
Class Definition
from sqlalchemy.ext.asyncio import AsyncSession
from zcore import UnitOfWork
from zcore.kernel.events import EventDispatcher
class UnitOfWork:
def __init__(self, session: AsyncSession, dispatcher: EventDispatcher):
...Properties
Prop
Type
Methods
register_event
Queues a domain event for post-commit dispatch. Events registered across nested unit-of-work scopes are buffered into the shared session store (session.info["uow_events"]) and dispatched collectively upon the successful root commit.
def register_event(self, event_name: str, payload: Any) -> None: ...Prop
Type
commit
Commits the database session and sequentially dispatches all buffered domain events. In nested transaction scopes, this method flushes pending operations to the database without committing the physical transaction until the outermost scope concludes.
async def commit(self) -> None: ...Nested Scope Flushing:
When called inside a nested UnitOfWork scope (session.info["uow_depth"] > 1), commit() executes await self.session.flush() instead of a physical database commit. The actual database commit and post-commit event dispatching are deferred until the outermost root context exits.
If the database commit fails, the session is rolled back immediately, all pending events are purged, and the original database exception is re-raised.
Prop
Type
rollback
Rolls back the database session, clears all pending domain events, and resets the transaction depth.
async def rollback(self) -> None: ...Prop
Type
Context Manager Protocol
UnitOfWork is designed to be used via async with. Entering the context increments session.info["uow_depth"] and marks the session with session.info["uow_managed"] = True to coordinate with BaseService._safe_commit():
from zcore import UnitOfWork, EventDispatcher
async def checkout(order_repo, inventory_repo, dispatcher: EventDispatcher, order_data):
# Automatically manages transaction commit/rollback and event queue on context exit
async with UnitOfWork(order_repo.db, dispatcher) as uow:
order = await order_repo.create(order_data)
await inventory_repo.decrement_stock(order.product_id, 1)
# Buffered in memory; dispatched only if the database commit succeeds!
uow.register_event("order.completed", {"order_id": str(order.id)})Upon block exit (__aexit__), UnitOfWork decrements uow_depth. If an exception occurred, it rolls back the entire session and discards all buffered events. If exiting the root boundary (depth == 0), it executes the final database commit and dispatches all accumulated events concurrently.