Transaction boundary for persistence operations.
Groups one or more repository operations into an atomic unit. Use as a
context manager to ensure that all changes within the block are committed
together or rolled back on error::
with UnitOfWork():
repo = domain.repository_for(Order)
order = repo.get(order_id)
order.confirm()
repo.add(order)
Command handlers and the @use_case decorator wrap their execution in a
UnitOfWork automatically, so explicit usage is typically only needed in
application services or scripts.
Nesting joins the outermost transaction. A UnitOfWork started while
another is already active on the same context does not open its own
transaction. There are no savepoints, so it joins the outermost one: every
write, read, and event routes to the outermost UnitOfWork, and only that one
commits or rolls back. A nested rollback rolls back the whole transaction.
This is why an application service called from within a command handler's
UnitOfWork composes into a single transaction rather than committing
independently. Independent inner transactions (a savepoint, or a
sub-transaction that commits on its own) are not supported, by design: the
aggregate is the consistency boundary and one UnitOfWork maps to one use
case. For cross-aggregate coordination use domain events; for a durable
side-effect that must survive a rollback use the outbox.
The UnitOfWork maintains an identity map to track loaded aggregates and
collects domain events raised during the transaction. On commit, events
are persisted to the outbox and dispatched to brokers/event store.
Source code in src/protean/core/unit_of_work.py
86
87
88
89
90
91
92
93
94
95
96
97
98
99 | def __init__(self) -> None:
self.domain = current_domain
self._in_progress = False
self._sessions: dict[str, SessionProtocol] = {}
self._messages_to_dispatch: list[tuple[str, dict[str, Any], str | None]] = []
self._identity_map: defaultdict[str, dict[Any, Any]] = defaultdict(dict)
# A UnitOfWork started while another is already active on this context is
# a participant in the outer transaction (see ``start``). ``_nested`` marks
# it; ``_rollback_only`` is set on the outermost UoW when a participant
# rolls back, dooming the whole transaction.
self._nested = False
self._rollback_only = False
|
start
Begin the transaction and push this UnitOfWork onto the context stack.
Opens no session, deliberately. The session, and on SQLAlchemy the
real BEGIN, appears at the first repository access through
:meth:_initialize_session. ADR-0031 turns that into a contract: a
handler method reaching no repository runs no transaction and holds no
pooled connection, which is what lets a handler talk to an external
system by putting the call in its own method. Opening a session here
would silently pin a connection for the length of every such call, so
this method must stay free of any eager connection or session.
A UnitOfWork started while another is already active on this context does
not open its own transaction. There are no savepoints, so it joins the
active (outermost) UnitOfWork: it does not push onto the stack, so
repository operations keep resolving current_uow to the outermost UoW
and route every write, session, and event there. Its own commit and
rollback then defer to the outermost UoW (see commit/rollback).
Source code in src/protean/core/unit_of_work.py
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202 | def start(self) -> None:
"""Begin the transaction and push this UnitOfWork onto the context stack.
**Opens no session, deliberately.** The session, and on SQLAlchemy the
real ``BEGIN``, appears at the first repository access through
:meth:`_initialize_session`. ADR-0031 turns that into a contract: a
handler method reaching no repository runs no transaction and holds no
pooled connection, which is what lets a handler talk to an external
system by putting the call in its own method. Opening a session here
would silently pin a connection for the length of every such call, so
this method must stay free of any eager connection or session.
A UnitOfWork started while another is already active on this context does
not open its own transaction. There are no savepoints, so it joins the
active (outermost) UnitOfWork: it does not push onto the stack, so
repository operations keep resolving ``current_uow`` to the outermost UoW
and route every write, session, and event there. Its own commit and
rollback then defer to the outermost UoW (see ``commit``/``rollback``).
"""
# Recompute nesting on every start: a UnitOfWork instance can be reused, so
# a stale ``_nested``/``_rollback_only`` from a prior use must not leak into
# this one. (A UoW previously used as nested, then reused as the outermost,
# would otherwise keep ``_nested=True`` and no-op its commit, silently
# losing the transaction and leaving itself on the context stack.)
self._nested = _uow_stack.top is not None
self._rollback_only = False
self._in_progress = True
if self._nested:
return
# Log transaction capability warnings for each configured provider
for provider_name, provider in self.domain.providers.items():
if not provider.has_capability(DatabaseCapabilities.TRANSACTIONS):
if provider.has_capability(DatabaseCapabilities.SIMULATED_TRANSACTIONS):
logger.debug(
"Provider '%s' uses simulated transactions. "
"Rollback will not undo persisted changes.",
provider_name,
)
else:
logger.warning(
"Provider '%s' does not support transactions. "
"UoW will manage identity map and events "
"but commit/rollback are not atomic.",
provider_name,
)
self._in_progress = True
_uow_stack.push(self)
|
commit
Commit all changes, persist outbox messages, and dispatch events.
| RAISES |
DESCRIPTION |
InvalidOperationError
|
If the UnitOfWork is not in progress.
|
ExpectedVersionError
|
On optimistic concurrency conflict.
|
TransactionError
|
If the underlying database commit fails.
|
Source code in src/protean/core/unit_of_work.py
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255 | def commit(self) -> None:
"""Commit all changes, persist outbox messages, and dispatch events.
Raises:
InvalidOperationError: If the UnitOfWork is not in progress.
ExpectedVersionError: On optimistic concurrency conflict.
TransactionError: If the underlying database commit fails.
"""
# Raise error if there the Unit Of Work is not active
logger.debug("uow.committing", extra={"uow_id": id(self)})
if not self._in_progress:
raise InvalidOperationError("UnitOfWork is not in progress")
if self._nested:
# A participant in an outer UnitOfWork: the outermost UoW owns the real
# commit, so there is nothing to commit or dispatch here. Reset our own
# state (it is empty, since a nested UoW routes everything to the
# outermost) for symmetry with the non-nested path and safe reuse.
self._reset()
return
if self._rollback_only:
# A nested participant rolled back, dooming the whole transaction. Roll
# back instead of committing; nothing is persisted or dispatched.
logger.warning(
"A nested UnitOfWork was rolled back; the whole transaction is "
"rolled back."
)
self.rollback()
return
tracer = self.domain.tracer
with tracer.start_as_current_span(
"protean.uow.commit",
record_exception=False,
set_status_on_exception=False,
) as span:
# Propagate correlation and causation IDs from the message being processed
msg = g.get("message_in_context")
if msg is not None and hasattr(msg, "metadata") and msg.metadata:
domain_meta = getattr(msg.metadata, "domain", None)
if domain_meta is not None:
correlation_id = getattr(domain_meta, "correlation_id", None)
if correlation_id:
span.set_attribute("protean.correlation_id", correlation_id)
causation_id = getattr(domain_meta, "causation_id", None)
if causation_id:
span.set_attribute("protean.causation_id", causation_id)
self._do_commit(span)
|
rollback
Roll back all changes and close sessions.
| RAISES |
DESCRIPTION |
InvalidOperationError
|
If the UnitOfWork is not in progress.
|
Source code in src/protean/core/unit_of_work.py
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616 | def rollback(self) -> None:
"""Roll back all changes and close sessions.
Raises:
InvalidOperationError: If the UnitOfWork is not in progress.
"""
# Raise error if the Unit Of Work is not active
if not self._in_progress:
raise InvalidOperationError("UnitOfWork is not in progress")
if self._nested:
# A participant cannot roll back only its own work (there are no
# savepoints), so it dooms the whole transaction: mark the outermost
# UoW rollback-only. The real session rollback happens when that
# outermost UoW exits (its commit sees the flag, or an exception
# propagating out of the block drives its rollback directly). Reset our
# own (empty) state for symmetry with the non-nested path.
outermost = _uow_stack.top
if outermost is not None:
outermost._rollback_only = True
self._reset()
return
# Record UoW outcome for the access log wide event
from protean.domain.context import has_domain_context # noqa: PLC0415
if has_domain_context():
g._access_log_uow_outcome = "rolled_back"
# Exit from Unit of Work. Guarded on identity so a double-pop (when the
# relational commit failed after _do_commit already popped this UoW)
# cannot pop a parent UnitOfWork off the stack.
if _uow_stack.top is self:
_uow_stack.pop()
try:
for session in self._sessions.values():
session.rollback()
logger.debug("uow.rollback_successful")
except Exception: # a failed rollback still resets the UoW; logged
logger.exception("uow.rollback_failed")
self._reset()
|
get_session
get_session(provider_name: str) -> SessionProtocol
Get session for provider, initializing one if it doesn't exist
Source code in src/protean/core/unit_of_work.py
| def get_session(self, provider_name: str) -> "SessionProtocol":
"""Get session for provider, initializing one if it doesn't exist"""
if provider_name in self._sessions:
return self._sessions[provider_name]
else:
return self._initialize_session(provider_name)
|