How to use UnitOfWork for atomic transactions

Coordinate multi-repository writes and post-commit domain events within an atomic all-or-nothing boundary.

When multiple services or repositories must modify the database within a single transaction, instantiate a UnitOfWork block to guarantee atomicity.

1. Instantiate `UnitOfWork` Inside the Service Method

Inject the system EventDispatcher and your repositories into the service constructor. Then, wrap multi-step database mutations inside async with UnitOfWork(...):

# services.py
from zcore import BaseService, UnitOfWork, EventDispatcher
from .models import Order
from .repositories import OrderRepo, InventoryRepo
from .schemas import OrderCreate

class CheckoutService(BaseService[Order]):
    def __init__(
        self, 
        repository: OrderRepo,
        inventory_repo: InventoryRepo,
        dispatcher: EventDispatcher
    ):
        super().__init__(model=Order, repository=repository)
        self.inventory_repo = inventory_repo
        self.dispatcher = dispatcher

    async def process_checkout(self, order_data: OrderCreate, product_id: str, quantity: int):
        # 1. Open an isolated transaction boundary using the active session and dispatcher
        async with UnitOfWork(self.repository.db, self.dispatcher) as uow:
            # Both operations share the exact same database transaction
            order = await self.repository.create(order_data)
            await self.inventory_repo.decrement_stock(product_id, quantity)

            # 2. Buffer domain events to dispatch ONLY after successful commit
            uow.register_event("order.completed", {"order_id": str(order.id)})
            
        # If any step fails, BOTH operations roll back and the event is never dispatched.

2. Nested & Re-entrant Transactions

In modular applications, domain services frequently invoke one another (e.g. CheckoutService calling an InventoryService). Each service can safely wrap its persistence operations within async with UnitOfWork(...):

Inner Scopes (depth > 1): Calls to uow.commit() safely execute session.flush() to populate database IDs and apply constraints without committing the physical transaction.
Root Scope (depth == 0): The outermost context manager boundary executes the final physical database commit.
Deferred Events Buffer: Events registered across all nested levels (session.info['uow_events']) are accumulated in the shared session buffer and dispatched together only after the root transaction commits.
Chain Rollback: An unhandled error at any depth immediately resets uow_depth, cancels all pending events, and rolls back the entire transaction chain.

3. Post-Commit Domain Events

UnitOfWork.register_event(event_name, payload) prevents side-effects on transaction failures:

Events are queued in an internal buffer.
If the database commit succeeds, all buffered events are dispatched concurrently via EventDispatcher.
If a database error occurs or an exception is raised, the transaction rolls back and all pending events are discarded immediately.

Automatic Coordination with BaseService: Entering async with UnitOfWork(...) automatically marks the session as uow_managed = True. Any BaseService calls (such as self.repository.create) inside the block will automatically skip their individual commits and let UnitOfWork finalize the transaction on block exit.

No Manual Commits: Never call db.commit() manually when working inside a UnitOfWork block. Commit and rollback transitions are orchestrated automatically upon context exit (__aexit__).

How to use UnitOfWork for atomic transactions

1. Instantiate UnitOfWork Inside the Service Method

2. Nested & Re-entrant Transactions

3. Post-Commit Domain Events

On this page

1. Instantiate `UnitOfWork` Inside the Service Method