(
self, _capture_exception: bool = False, _to_root: bool = False
)
| 1355 | SessionTransactionState.CLOSED, |
| 1356 | ) |
| 1357 | def rollback( |
| 1358 | self, _capture_exception: bool = False, _to_root: bool = False |
| 1359 | ) -> None: |
| 1360 | stx = self.session._transaction |
| 1361 | assert stx is not None |
| 1362 | if stx is not self: |
| 1363 | for subtransaction in stx._iterate_self_and_parents(upto=self): |
| 1364 | subtransaction.close() |
| 1365 | |
| 1366 | boundary = self |
| 1367 | rollback_err = None |
| 1368 | if self._state in ( |
| 1369 | SessionTransactionState.ACTIVE, |
| 1370 | SessionTransactionState.PREPARED, |
| 1371 | ): |
| 1372 | for transaction in self._iterate_self_and_parents(): |
| 1373 | if transaction._parent is None or transaction.nested: |
| 1374 | try: |
| 1375 | for t in set(transaction._connections.values()): |
| 1376 | t[1].rollback() |
| 1377 | |
| 1378 | transaction._state = SessionTransactionState.DEACTIVE |
| 1379 | self.session.dispatch.after_rollback(self.session) |
| 1380 | except: |
| 1381 | rollback_err = sys.exc_info() |
| 1382 | finally: |
| 1383 | transaction._state = SessionTransactionState.DEACTIVE |
| 1384 | transaction._restore_snapshot( |
| 1385 | dirty_only=transaction.nested |
| 1386 | ) |
| 1387 | boundary = transaction |
| 1388 | break |
| 1389 | else: |
| 1390 | transaction._state = SessionTransactionState.DEACTIVE |
| 1391 | |
| 1392 | sess = self.session |
| 1393 | |
| 1394 | if not rollback_err and not sess._is_clean(): |
| 1395 | # if items were added, deleted, or mutated |
| 1396 | # here, we need to re-restore the snapshot |
| 1397 | util.warn( |
| 1398 | "Session's state has been changed on " |
| 1399 | "a non-active transaction - this state " |
| 1400 | "will be discarded." |
| 1401 | ) |
| 1402 | boundary._restore_snapshot(dirty_only=boundary.nested) |
| 1403 | |
| 1404 | with self._expect_state(SessionTransactionState.CLOSED): |
| 1405 | self.close() |
| 1406 | |
| 1407 | if self._parent and _capture_exception: |
| 1408 | self._parent._rollback_exception = sys.exc_info()[1] |
| 1409 | |
| 1410 | if rollback_err and rollback_err[1]: |
| 1411 | raise rollback_err[1].with_traceback(rollback_err[2]) |
| 1412 | |
| 1413 | sess.dispatch.after_soft_rollback(sess, self) |
| 1414 |
no test coverage detected