"""Transaction support for Gaphor."""from__future__importannotationsimportloggingfromgaphor.eventimportTransactionBegin,TransactionCommit,TransactionRollbacklog=logging.getLogger(__name__)classTransactionError(Exception):"""Errors related to the transaction module."""
[docs]classTransaction:"""The transaction. On start and end of a transaction an event is emitted. Transactions can be nested. Events are only emitted when the outermost transaction begins and finishes. Note that transactions are a global construct. >>> import gaphor.core.eventmanager >>> event_manager = gaphor.core.eventmanager.EventManager() Transactions can be nested. If the outermost transaction is committed or rolled back, an event is emitted. It's most convenient to use ``Transaction`` as a context manager: >>> with Transaction(event_manager) as ctx: ... ... # do actions ... # in case the transaction should be rolled back: ... ctx.rollback() Events can be handled programmatically, although this is discouraged: >>> tx = Transaction(event_manager) >>> tx.commit() """_stack:list[Transaction]=[]def__init__(self,event_manager,context=None):"""Initialize the transaction. If this is the first transaction in the stack, a :obj:`~gaphor.event.TransactionBegin` event is emitted. """self.event_manager=event_managerself.context=contextself._need_rollback=Falseifnotself._stack:self._handle(TransactionBegin(self.context))self._stack.append(self)
[docs]defcommit(self):"""Commit the transaction. The transaction is closed. A :obj:`~gaphor.event.TransactionCommit` event is emitted. If the transaction needs to be rolled back, a :obj:`~gaphor.event.TransactionRollback` event is emitted instead. """self._close()ifnotself._stack:ifself._need_rollback:self._handle(TransactionRollback(self.context))else:self._handle(TransactionCommit(self.context))
[docs]defrollback(self):"""Roll-back the transaction. First, the transaction is closed. A :obj:`~gaphor.event.TransactionRollback` event is emitted. """self.mark_rollback()self.commit()
[docs]@classmethoddefmark_rollback(cls):"""Mark the transaction for rollback. This operation itself will not close the transaction, instead it will allow you to elegantly revert changes. """fortxincls._stack:tx._need_rollback=True# noqa: SLF001
[docs]@classmethoddefin_transaction(cls)->bool:"""Are you running inside a transaction?"""returnbool(cls._stack)
def_close(self):try:last=self._stack.pop()exceptIndexError:raiseTransactionError("No Transaction on stack.")fromNoneiflastisnotself:self._stack.append(last)raiseTransactionError("Transaction on stack is not the transaction being closed.")def_handle(self,event):self.event_manager.handle(event)def__enter__(self)->TransactionContext:"""Provide ``with``-statement transaction support."""returnTransactionContext(self)def__exit__(self,exc_type,exc_val,exc_tb):"""Provide ``with``-statement transaction support. If an error occurred, the transaction is rolled back. Otherwise, it is committed. """ifexc_typeandnotself._need_rollback:log.error("Transaction terminated due to an exception, performing a rollback",)self.mark_rollback()self.commit()
classTransactionContext:"""A simple context for a transaction. Can only perform a rollback. """def__init__(self,tx:Transaction)->None:self._tx=txdefrollback(self)->None:self._tx.mark_rollback()