Merge pull request #73 from qstokkink/add_lt_type_check

Added extended type checking with the libtorrent API

.github/workflows/mypy.yml

@@ -12,4 +12,6 @@ jobs:
       - name: Install mypy
         run: pip install mypy
       - name: Run mypy
-        run: mypy -p src.tribler.core
+        run: |
+          wget -O libtorrent.pyi https://github.com/arvidn/libtorrent/raw/master/bindings/python/install_data/libtorrent/__init__.pyi
+          mypy -p src.tribler.core

src/tribler/core/libtorrent/download_manager/download.py

@@ -26,7 +26,7 @@
 from tribler.core.libtorrent.download_manager.download_state import DownloadState, DownloadStatus
 from tribler.core.libtorrent.download_manager.stream import Stream
 from tribler.core.libtorrent.torrent_file_tree import TorrentFileTree
-from tribler.core.libtorrent.torrentdef import TorrentDef, TorrentDefNoMetainfo
+from tribler.core.libtorrent.torrentdef import MetainfoDict, TorrentDef, TorrentDefNoMetainfo
 from tribler.core.libtorrent.torrents import check_handle, get_info_from_handle, require_handle
 from tribler.core.notifier import Notification, Notifier
 from tribler.tribler_config import TriblerConfigManager
@@ -83,7 +83,7 @@ class PeerDictHave(PeerDict):
     Extended peer info that includes the "have" field.
     """
 
-    have: Any  # Bitfield object for this peer if not completed
+    have: list[bool]  # Bitfield object for this peer if not completed
 
 
 class Download(TaskManager):
@@ -177,7 +177,7 @@ def add_stream(self) -> None:
         assert self.stream is None
         self.stream = Stream(self)
 
-    def get_torrent_data(self) -> bytes | None:
+    def get_torrent_data(self) -> dict[bytes, Any] | None:
         """
         Return torrent data, if the handle is valid and metadata is available.
         """
@@ -243,9 +243,9 @@ def get_atp(self) -> dict:
                         | lt.add_torrent_params_flags_t.flag_update_subscribe}
 
         if self.config.get_share_mode():
-            atp["flags"] = atp["flags"] | lt.add_torrent_params_flags_t.flag_share_mode
+            atp["flags"] = cast(int, atp["flags"]) | lt.add_torrent_params_flags_t.flag_share_mode
         if self.config.get_upload_mode():
-            atp["flags"] = atp["flags"] | lt.add_torrent_params_flags_t.flag_upload_mode
+            atp["flags"] = cast(int, atp["flags"]) | lt.add_torrent_params_flags_t.flag_upload_mode
 
         resume_data = self.config.get_engineresumedata()
         if not isinstance(self.tdef, TorrentDefNoMetainfo):
@@ -387,7 +387,9 @@ def on_save_resume_data_alert(self, alert: lt.save_resume_data_alert) -> None:
         if self.checkpoint_disabled:
             return
 
-        resume_data = alert.resume_data
+        resume_data = (cast(dict[bytes, Any], lt.bdecode(alert.resume_data))
+                       if isinstance(alert.resume_data, bytes)  # Libtorrent 2.X
+                       else alert.resume_data)  # Libtorrent 1.X
         # Make save_path relative if the torrent is saved in the Tribler state directory
         if self.state_dir and b"save_path" in resume_data:
             save_path = Path(resume_data[b"save_path"].decode()).absolute()
@@ -469,7 +471,7 @@ def on_metadata_received_alert(self, alert: lt.metadata_received_alert) -> None:
             return
 
         try:
-            metadata = {b"info": lt.bdecode(torrent_info.metadata()), b"leechers": 0, b"seeders": 0}
+            metadata = cast(MetainfoDict, {b"info": lt.bdecode(torrent_info.metadata()), b"leechers": 0, b"seeders": 0})
         except (RuntimeError, ValueError) as e:
             self._logger.warning(e)
             return
@@ -678,7 +680,7 @@ def get_peer_list(self, include_have: bool = True) -> List[PeerDict | PeerDictHa
             try:
                 extended_version = peer_info.client
             except UnicodeDecodeError:
-                extended_version = "unknown"
+                extended_version = b"unknown"
             peer_dict: PeerDict | PeerDictHave = cast(PeerDict, {
                 "id": hexlify(peer_info.pid.to_bytes()).decode(),
                 "extended_version": extended_version,
@@ -700,7 +702,7 @@ def get_peer_list(self, include_have: bool = True) -> List[PeerDict | PeerDictHa
                 "dtotal": peer_info.total_download,
                 "completed": peer_info.progress,
                 "speed": peer_info.remote_dl_rate,
-                "connection_type": peer_info.connection_type,
+                "connection_type": peer_info.connection_type,  # type: ignore[attr-defined] # shortcoming of stubs
                 "seed": bool(peer_info.flags & peer_info.seed),
                 "upload_only": bool(peer_info.flags & peer_info.upload_only)
             })
@@ -726,7 +728,7 @@ def get_num_connected_seeds_peers(self) -> Tuple[int, int]:
 
         return num_seeds, num_peers
 
-    def get_torrent(self) -> bytes | None:
+    def get_torrent(self) -> dict[bytes, Any] | None:
         """
         Create the raw torrent data from this download.
         """
@@ -874,7 +876,7 @@ def get_magnet_link(self) -> str:
         """
         Generate a magnet link for our download.
         """
-        return lt.make_magnet_uri(self.handle)
+        return lt.make_magnet_uri(cast(lt.torrent_handle, self.handle))  # Ensured by ``check_handle``
 
     @require_handle
     def add_peer(self, addr: tuple[str, int]) -> None:

src/tribler/core/libtorrent/download_manager/download_config.py

@@ -89,11 +89,16 @@ def _from_dict(value: Dict) -> str:
     return base64_bytes.decode()
 
 
-def _to_dict(value: str) -> dict | None:
+def _to_dict(value: str) -> dict[bytes, Any]:
+    """
+    Convert a string value to a libtorrent dict.
+
+    :raises RuntimeError: if the value could not be converted.
+    """
     binary = value.encode()
     # b'==' is added to avoid incorrect padding
     base64_bytes = base64.b64decode(binary + b"==")
-    return lt.bdecode(base64_bytes)
+    return cast(dict[bytes, Any], lt.bdecode(base64_bytes))
 
 
 class DownloadConfig:

src/tribler/core/libtorrent/download_manager/download_manager.py

@@ -10,14 +10,13 @@
 import logging
 import os
 import time
-import time as timemod
 from asyncio import CancelledError, gather, iscoroutine, shield, sleep, wait_for
 from binascii import hexlify, unhexlify
 from collections import defaultdict
 from copy import deepcopy
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Iterable, List
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Iterable, List, cast
 
 import libtorrent as lt
 from configobj import ConfigObj
@@ -291,8 +290,7 @@ def create_session(self, hops: int = 0) -> lt.session:  # noqa: PLR0915
         logger.info("Hops: %d.", hops)
 
         # Elric: Strip out the -rcX, -beta, -whatever tail on the version string.
-        fingerprint = ["TL", 0, 0, 0, 0]
-        ltsession = lt.session(lt.fingerprint(*fingerprint), flags=0) if hops == 0 else lt.session(flags=0)
+        ltsession = lt.session(lt.fingerprint("TL", 0, 0, 0, 0), flags=0) if hops == 0 else lt.session(flags=0)
 
         libtorrent_port = self.config.get("libtorrent/port")
         logger.info("Libtorrent port: %d", libtorrent_port)
@@ -440,19 +438,20 @@ def process_alert(self, alert: lt.alert, hops: int = 0) -> None:  # noqa: C901,
         # Periodically, libtorrent will send us a state_update_alert, which contains the torrent status of
         # all torrents changed since the last time we received this alert.
         if alert_type == "state_update_alert":
-            for status in alert.status:
+            for status in cast(lt.state_update_alert, alert).status:
                 infohash = status.info_hash.to_bytes()
                 if infohash not in self.downloads:
                     logger.debug("Got state_update for unknown torrent %s", hexlify(infohash))
                     continue
                 self.downloads[infohash].update_lt_status(status)
 
         if alert_type == "state_changed_alert":
-            infohash = alert.handle.info_hash().to_bytes()
+            handle = cast(lt.state_changed_alert, alert).handle
+            infohash = handle.info_hash().to_bytes()
             if infohash not in self.downloads:
                 logger.debug("Got state_change for unknown torrent %s", hexlify(infohash))
             else:
-                self.downloads[infohash].update_lt_status(alert.handle.status())
+                self.downloads[infohash].update_lt_status(handle.status())
 
         infohash = (alert.handle.info_hash().to_bytes() if hasattr(alert, "handle") and alert.handle.is_valid()
                     else getattr(alert, "info_hash", b""))
@@ -462,33 +461,36 @@ def process_alert(self, alert: lt.alert, hops: int = 0) -> None:  # noqa: C901,
                                or (not download.handle and alert_type == "add_torrent_alert") \
                                or (download.handle and alert_type == "torrent_removed_alert")
             if is_process_alert:
-                download.process_alert(alert, alert_type)
+                download.process_alert(cast(lt.torrent_alert, alert), alert_type)
             else:
                 logger.debug("Got alert for download without handle %s: %s", infohash, alert)
         elif infohash:
             logger.debug("Got alert for unknown download %s: %s", infohash, alert)
 
         if alert_type == "listen_succeeded_alert":
-            self.listen_ports[hops][alert.address] = alert.port
+            ls_alert = cast(lt.listen_succeeded_alert, alert)
+            self.listen_ports[hops][ls_alert.address] = ls_alert.port
 
         elif alert_type == "peer_disconnected_alert":
-            self.notifier.notify(Notification.peer_disconnected, peer_id=alert.pid.to_bytes())
+            self.notifier.notify(Notification.peer_disconnected,
+                                 peer_id=cast(lt.peer_disconnected_alert, alert).pid.to_bytes())
 
         elif alert_type == "session_stats_alert":
-            queued_disk_jobs = alert.values["disk.queued_disk_jobs"]
-            self.queued_write_bytes = alert.values["disk.queued_write_bytes"]
-            num_write_jobs = alert.values["disk.num_write_jobs"]
+            ss_alert = cast(lt.session_stats_alert, alert)
+            queued_disk_jobs = ss_alert.values["disk.queued_disk_jobs"]
+            self.queued_write_bytes = ss_alert.values["disk.queued_write_bytes"]
+            num_write_jobs = ss_alert.values["disk.num_write_jobs"]
             if queued_disk_jobs == self.queued_write_bytes == num_write_jobs == 0:
                 self.lt_session_shutdown_ready[hops] = True
 
             if self.session_stats_callback:
-                self.session_stats_callback(alert)
+                self.session_stats_callback(ss_alert)
 
         elif alert_type == "dht_pkt_alert" and self.dht_health_manager is not None:
             # Unfortunately, the Python bindings don't have a direction attribute.
             # So, we'll have to resort to using the string representation of the alert instead.
             incoming = str(alert).startswith("<==")
-            decoded = lt.bdecode(alert.pkt_buf)
+            decoded = cast(dict[bytes, Any], lt.bdecode(cast(lt.dht_pkt_alert, alert).pkt_buf))
             if not decoded:
                 return
 
@@ -564,7 +566,7 @@ async def get_metainfo(self, infohash: bytes, timeout: float = 7, hops: int | No
             return None
 
         logger.info("Successfully retrieved metainfo for %s", infohash_hex)
-        self.metainfo_cache[infohash] = {"time": timemod.time(), "meta_info": metainfo}
+        self.metainfo_cache[infohash] = {"time": time.time(), "meta_info": metainfo}
         self.notifier.notify(Notification.torrent_metadata_added, metadata={
             "infohash": infohash,
             "size": download.tdef.get_length(),
@@ -582,7 +584,7 @@ async def get_metainfo(self, infohash: bytes, timeout: float = 7, hops: int | No
         return metainfo
 
     def _task_cleanup_metainfo_cache(self) -> None:
-        oldest_time = timemod.time() - METAINFO_CACHE_PERIOD
+        oldest_time = time.time() - METAINFO_CACHE_PERIOD
 
         for info_hash, cache_entry in list(self.metainfo_cache.items()):
             last_time = cache_entry["time"]
@@ -625,7 +627,7 @@ async def start_download_from_uri(self, uri: str, config: DownloadConfig | None
             params = lt.parse_magnet_uri(uri)
             try:
                 # libtorrent 1.2.19
-                name, infohash = params["name"].encode(), params["info_hash"]
+                name, infohash = params["name"].encode(), params["info_hash"]  # type: ignore[index] # (checker is 2.X)
             except TypeError:
                 # libtorrent 2.0.9
                 name = params.name.encode()
@@ -689,7 +691,7 @@ async def start_download(self, torrent_file: str | None = None, tdef: TorrentDef
             logger.exception("Unable to create the download destination directory.")
 
         if config.get_time_added() == 0:
-            config.set_time_added(int(timemod.time()))
+            config.set_time_added(int(time.time()))
 
         # Create the download
         download = Download(tdef=tdef,
@@ -786,7 +788,7 @@ def set_session_settings(self, lt_session: lt.session, new_settings: dict) -> No
             if hasattr(lt_session, "apply_settings"):
                 lt_session.apply_settings(new_settings)
             else:
-                lt_session.set_settings(new_settings)
+                lt_session.set_settings(new_settings)  # type: ignore[attr-defined] # (checker uses 2.X)
         except OverflowError as e:
             msg = f"Overflow error when setting libtorrent sessions with settings: {new_settings}"
             raise OverflowError(msg) from e

src/tribler/core/libtorrent/download_manager/stream.py

@@ -22,7 +22,7 @@
 from asyncio import sleep
 from io import BufferedReader
 from pathlib import Path
-from typing import TYPE_CHECKING, Generator, cast
+from typing import TYPE_CHECKING, Callable, Generator, cast
 
 import libtorrent
 from typing_extensions import Self
@@ -94,7 +94,7 @@ def __init__(self, download: Download) -> None:
         self.destdir: Path | None = None
         self.piecelen: int | None = None
         self.files: list[tuple[Path, int]] | None = None
-        self.mapfile: libtorrent.peer_request | None = None
+        self.mapfile: Callable[[int, int, int], libtorrent.peer_request] | None = None
         self.prebuffpieces: list[int] = []
         self.headerpieces: list[int] = []
         self.footerpieces: list[int] = []
@@ -151,7 +151,7 @@ async def enable(self, fileindex: int = 0, prebufpos: int | None = None) -> None
         self.piecelen = cast(int, self.piecelen)
         self.files = cast(list[tuple[Path, int]], self.files)
         self.infohash = cast(bytes, self.infohash)
-        self.mapfile = cast(libtorrent.peer_request, self.mapfile)
+        self.mapfile = cast(Callable[[int, int, int], libtorrent.peer_request], self.mapfile)
 
         # if fileindex not available for torrent raise exception
         if fileindex >= len(self.files):
@@ -290,10 +290,12 @@ def bytestopieces(self, bytes_begin: int, bytes_end: int) -> list[int]:
     def bytetopiece(self, byte_begin: int) -> int:
         """
         Finds the piece position that begin_bytes is mapped to.
+
+        ``check_vod`` ensures the types of ``self.mapfile`` and ``self.fileindex``.
         """
-        self.mapfile = cast(libtorrent.peer_request, self.mapfile)  # Ensured by ``check_vod``
+        self.mapfile = cast(Callable[[int, int, int], libtorrent.peer_request], self.mapfile)
 
-        return self.mapfile(self.fileindex, byte_begin, 0).piece
+        return self.mapfile(cast(int, self.fileindex), byte_begin, 0).piece
 
     @check_vod(0)
     def calculateprogress(self, pieces: list[int], consec: bool) -> float:

src/tribler/core/libtorrent/restapi/libtorrent_endpoint.py

@@ -97,7 +97,7 @@ async def get_libtorrent_session_info(self, request: Request) -> RESTResponse:
         """
         session_stats: Future[dict[str, int]] = Future()
 
-        def on_session_stats_alert_received(alert: libtorrent.alert) -> None:
+        def on_session_stats_alert_received(alert: libtorrent.session_stats_alert) -> None:
             if not session_stats.done():
                 session_stats.set_result(alert.values)
 

src/tribler/core/libtorrent/restapi/torrentinfo_endpoint.py

@@ -161,7 +161,7 @@ async def get_torrent_info(self, request: Request) -> RESTResponse:  # noqa: C90
                 try:
                     try:
                         # libtorrent 1.2.19
-                        infohash = lt.parse_magnet_uri(uri)["info_hash"]
+                        infohash = lt.parse_magnet_uri(uri)["info_hash"]  # type: ignore[index] # (checker uses 2.X)
                     except TypeError:
                         # libtorrent 2.0.9
                         infohash = unhexlify(str(lt.parse_magnet_uri(uri).info_hash))
@@ -181,7 +181,7 @@ async def get_torrent_info(self, request: Request) -> RESTResponse:  # noqa: C90
             try:
                 try:
                     # libtorrent 1.2.19
-                    infohash = lt.parse_magnet_uri(uri)["info_hash"]
+                    infohash = lt.parse_magnet_uri(uri)["info_hash"]  # type: ignore[index] # (checker uses 2.X)
                 except TypeError:
                     # libtorrent 2.0.9
                     infohash = unhexlify(str(lt.parse_magnet_uri(uri).info_hash))

src/tribler/core/libtorrent/torrentdef.py

@@ -134,7 +134,7 @@ def __getitem__(self, key: bytes) -> Any: ...  # noqa: D105
 else:
     FileDict = dict
     InfoDict = dict
-    MetainfoDict = dict
+    MetainfoDict = dict[bytes, Any]
     TorrentParameters = dict
 
 
@@ -211,7 +211,8 @@ def __init__(self, metainfo: MetainfoDict | None = None,
             if not ignore_validation:
                 try:
                     self._torrent_info = lt.torrent_info(self.metainfo)
-                    self.infohash = self._torrent_info.info_hash()
+                    raw_infohash = self._torrent_info.info_hash()  # LT1.X: bytes, LT2.X: sha1_hash
+                    self.infohash = raw_infohash if isinstance(raw_infohash, bytes) else raw_infohash.to_bytes()
                 except RuntimeError as exc:
                     raise ValueError from exc
             else:
@@ -272,7 +273,7 @@ def load_torrent_info(self) -> None:
         Load the torrent info into memory from our metainfo if it does not exist.
         """
         if self._torrent_info is None:
-            self._torrent_info = lt.torrent_info(self.metainfo)
+            self._torrent_info = lt.torrent_info(cast(dict[bytes, Any], self.metainfo))
 
     def torrent_info_loaded(self) -> bool:
         """
@@ -320,7 +321,7 @@ def load_from_memory(bencoded_data: bytes) -> TorrentDef:
         if metainfo is None:
             msg = "Data is not a bencoded string"
             raise ValueError(msg)
-        return TorrentDef.load_from_dict(metainfo)
+        return TorrentDef.load_from_dict(cast(MetainfoDict, metainfo))
 
     @staticmethod
     def load_from_dict(metainfo: MetainfoDict) -> TorrentDef:

src/tribler/core/libtorrent/torrents.py

@@ -124,7 +124,7 @@ class TorrentFileResult(TypedDict):
     success: bool
     base_dir: Path
     torrent_file_path: str | None
-    metainfo: dict
+    metainfo: bytes
     infohash: bytes
 
 
@@ -185,17 +185,17 @@ def create_torrent_file(file_path_list: list[Path], params: InfoDict,  # noqa: C
     lt.set_piece_hashes(torrent, str(base_dir))
 
     t1 = torrent.generate()
-    torrent = lt.bencode(t1)
+    torrent_bytes = lt.bencode(t1)
 
     if torrent_filepath:
         with open(torrent_filepath, "wb") as f:
-            f.write(torrent)
+            f.write(torrent_bytes)
 
     return {
         "success": True,
         "base_dir": base_dir,
         "torrent_file_path": torrent_filepath,
-        "metainfo": torrent,
+        "metainfo": torrent_bytes,
         "infohash": sha1(lt.bencode(t1[b"info"])).digest()
     }