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