Use db_session retry and more exception handling in upgrade

.ruff.toml

@@ -4,7 +4,6 @@ lint.unfixable = []
 
 lint.ignore = [
     "ANN003",
-    "ANN101",
     "ARG001",
     "ARG002",
     "ARG005",

src/tribler/upgrade_script.py

@@ -18,7 +18,7 @@
 from typing import TYPE_CHECKING
 
 from configobj import ConfigObj
-from pony.orm import Database, db_session
+from pony.orm import Database, OperationalError, db_session
 
 if TYPE_CHECKING:
     from collections.abc import Generator
@@ -28,7 +28,7 @@
 FROM: str = "7.14"
 TO: str = "8.0"
 
-# ruff: noqa: B007,F841,N802,RUF015,W291
+# ruff: noqa: B007,C901,F841,N802,RUF015,W291
 
 
 def batched(results: list, n: int = 1) -> Generator[list]:
@@ -122,13 +122,13 @@ def _inject_StatementOp(abs_src_db: str, abs_dst_db: str) -> None:
     db = Database()
     db.bind(provider="sqlite", filename=abs_dst_db)
     for batch in batched(output, n=20):
-        with db_session:
-            for (subject_name, subject_type,
-                 object_name, object_type,
-                 stmt_added_count, stmt_removed_count, stmt_local_operation,
-                 peer_public_key, peer_added_at,
-                 stmtop_operation, stmtop_clock, stmtop_signature, stmtop_updated_at, stmtop_auto_generated) in batch:
-                try:
+        try:
+            with db_session(retry=3):
+                for (subject_name, subject_type,
+                     object_name, object_type,
+                     stmt_added_count, stmt_removed_count, stmt_local_operation,
+                     peer_public_key, peer_added_at,
+                     stmtop_operation, stmtop_clock, stmtop_signature, stmtop_updated_at, stmtop_auto_generated) in batch:
                     # Insert subject
                     results = list(db.execute("SELECT id FROM Resource WHERE name=$subject_name AND type=$subject_type",
                                               globals(), locals()))
@@ -195,12 +195,12 @@ def _inject_StatementOp(abs_src_db: str, abs_dst_db: str) -> None:
                             "$statement_id, $peer_id, $stmtop_operation, $stmtop_clock, $stmtop_signature, "
                             "$stmtop_updated_at, $stmtop_auto_generated)",
                             globals(), locals())
-
-                    db.commit()
-                except sqlite3.DatabaseError as e:
-                    db.rollback()
-                    logging.exception(e)
+        except OperationalError as e:
+            logging.exception(e)
+    try:
         db.disconnect()
+    except OperationalError as e:
+        logging.exception(e)
 
 
 def _inject_ChannelNode(abs_src_db: str, abs_dst_db: str) -> None:
@@ -221,11 +221,11 @@ def _inject_ChannelNode(abs_src_db: str, abs_dst_db: str) -> None:
     db = Database()
     db.bind(provider="sqlite", filename=abs_dst_db)
     for batch in batched(output, n=20):
-        with db_session:
-            for (infohash, size, torrent_date, tracker_info, title, tags, metadata_type, reserved_flags,
-                 origin_id, public_key, id_, timestamp, signature, added_on, status, xxx, tag_processor_version,
-                 seeders, leechers, last_check, self_checked, has_data) in batch:
-                try:
+        try:
+            with db_session(retry=3):
+                for (infohash, size, torrent_date, tracker_info, title, tags, metadata_type, reserved_flags,
+                     origin_id, public_key, id_, timestamp, signature, added_on, status, xxx, tag_processor_version,
+                     seeders, leechers, last_check, self_checked, has_data) in batch:
                     # Insert subject
                     results = list(db.execute("SELECT rowid FROM TorrentState WHERE infohash=$infohash",
                                               globals(), locals()))
@@ -247,12 +247,12 @@ def _inject_ChannelNode(abs_src_db: str, abs_dst_db: str) -> None:
                             "$reserved_flags, $origin_id, $public_key, $id_, $timestamp, $signature, $added_on, "
                             "$status, $xxx, $health_id, $tag_processor_version)",
                             globals(), locals())
-
-                    db.commit()
-                except sqlite3.DatabaseError as e:
-                    db.rollback()
-                    logging.exception(e)
+        except OperationalError as e:
+            logging.exception(e)
+    try:
         db.disconnect()
+    except OperationalError as e:
+        logging.exception(e)
 
 
 def _inject_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
@@ -266,9 +266,9 @@ def _inject_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
     db = Database()
     db.bind(provider="sqlite", filename=abs_dst_db)
     for batch in batched(output, n=20):
-        with db_session:
-            for (url, last_check, alive, failures) in batch:
-                try:
+        try:
+            with db_session(retry=3):
+                for (url, last_check, alive, failures) in batch:
                     results = list(db.execute("SELECT rowid, last_check, alive, failures FROM TrackerState WHERE "
                                               "url=$url",
                                               globals(), locals()))
@@ -286,12 +286,12 @@ def _inject_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
                             "INSERT INTO TrackerState VALUES ((SELECT COALESCE(MAX(rowid),0)+1 FROM TrackerState), "
                             "$url, $last_check, $alive, $failures)",
                             globals(), locals())
-
-                    db.commit()
-                except sqlite3.DatabaseError as e:
-                    db.rollback()
-                    logging.exception(e)
+        except OperationalError as e:
+            logging.exception(e)
+    try:
         db.disconnect()
+    except OperationalError as e:
+        logging.exception(e)
 
 
 def _inject_TorrentState_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
@@ -309,9 +309,9 @@ def _inject_TorrentState_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
     db = Database()
     db.bind(provider="sqlite", filename=abs_dst_db)
     for batch in batched(output, n=20):
-        with db_session:
-            for (infohash, url) in batch:
-                try:
+        try:
+            with db_session(retry=3):
+                for (infohash, url) in batch:
                     results = list(db.execute("""SELECT TorrentState.infohash, TrackerState.url
 FROM TorrentState_TrackerState
 INNER JOIN TorrentState ON TorrentState_TrackerState.torrentstate=TorrentState.rowid
@@ -334,12 +334,12 @@ def _inject_TorrentState_TrackerState(abs_src_db: str, abs_dst_db: str) -> None:
                         tracker_state, = tracker_states[0]
                         db.execute("INSERT INTO TorrentState_TrackerState VALUES ($torrent_state, $tracker_state)",
                                     globals(), locals())
-
-                    db.commit()
-                except sqlite3.DatabaseError as e:
-                    db.rollback()
-                    logging.exception(e)
+        except OperationalError as e:
+            logging.exception(e)
+    try:
         db.disconnect()
+    except OperationalError as e:
+        logging.exception(e)
 
 
 def _inject_7_14_tables(src_db: str, dst_db: str, db_format: str) -> None: