Add type hints to download manager

minigalaxy/download_manager.py

@@ -21,6 +21,7 @@
 import time
 import threading
 import queue
+from typing import Tuple, Dict, List
 
 from requests import Session
 from requests.exceptions import RequestException
@@ -35,15 +36,15 @@ class QueuedDownloadItem:
     """
     Wrap downloads in a simple class so we can manage when to download them
     """
-    def __init__(self, download, priority=1):
+    def __init__(self, download: Download, priority: int=1):
         """
         Create a QueuedDownloadItem with a given download and priority level
         """
         self.priority = priority
         self.item = download
         self.queue_time = time.time()
 
-    def __lt__(self, other):
+    def __lt__(self, other) -> bool:
         """
         Only compare QueuedDownloadItems on their priority level and
         time added to the queue.
@@ -98,7 +99,7 @@ def __init__(self, session: Session):
                 download_thread.start()
                 self.workers.append(download_thread)
 
-    def download(self, download):
+    def download(self, download: Download) -> None:
         """
         Add a download or list of downloads to the queue for downloading
         You can download a single Download or a list of Downloads
@@ -113,8 +114,8 @@ def download(self, download):
             for d in download:
                 self.put_in_proper_queue(d)
 
-    def put_in_proper_queue(self, download):
-        "Put the download in the proper queue"
+    def put_in_proper_queue(self, download: Download) -> None:
+        """Put the download in the proper queue"""
         # Add game type downloads to the game queue
         if download.download_type == DownloadType.GAME:
             self.__game_queue.put(QueuedDownloadItem(download, 1))
@@ -130,7 +131,7 @@ def put_in_proper_queue(self, download):
             # Add other items to the UI queue
             self.__ui_queue.put(QueuedDownloadItem(download, 0))
 
-    def download_now(self, download):
+    def download_now(self, download: Download) -> None:
         """
         Download an item with a higher priority
         Any item with the download_now priority set will get downloaded
@@ -144,7 +145,7 @@ def download_now(self, download):
         """
         self.__ui_queue.put(QueuedDownloadItem(download, 0))
 
-    def cancel_download(self, downloads):
+    def cancel_download(self, downloads: List[Download]) -> None:
         """
         Cancel a download or a list of downloads
 
@@ -167,7 +168,7 @@ def cancel_download(self, downloads):
         # cancel list
         self.cancel_queued_downloads(download_dict)
 
-    def cancel_active_downloads(self, download_dict):
+    def cancel_active_downloads(self, download_dict: Dict) -> None:
         """
         Cancel active downloads
         This is called by cancel_download
@@ -184,7 +185,7 @@ def cancel_active_downloads(self, download_dict):
                     # Remove it from the downloads to cancel
                     del download_dict[download]
 
-    def cancel_queued_downloads(self, download_dict):
+    def cancel_queued_downloads(self, download_dict: Dict) -> None:
         """
         Cancel selected downloads in the queue
         This is called by cancel_download
@@ -219,7 +220,7 @@ def cancel_queued_downloads(self, download_dict):
                     item = new_queue.get()
                     download_queue.put(item)
 
-    def cancel_current_downloads(self):
+    def cancel_current_downloads(self) -> None:
         """
         Cancel the currentl downloads
         """
@@ -228,7 +229,7 @@ def cancel_current_downloads(self):
             for download in self.active_downloads:
                 self.__cancel[download] = True
 
-    def cancel_all_downloads(self):
+    def cancel_all_downloads(self) -> None:
         """
         Cancel all current downloads queued
         """
@@ -244,16 +245,16 @@ def cancel_all_downloads(self):
 
         self.cancel_current_downloads()
 
-    def __remove_download_from_active_downloads(self, download):
-        "Remove a download from the list of active downloads"
+    def __remove_download_from_active_downloads(self, download: Download) -> None:
+        """Remove a download from the list of active downloads"""
         with self.active_downloads_lock:
             if download in self.active_downloads:
                 self.logger.debug("Removing download from active downloads list")
                 del self.active_downloads[download]
             else:
                 self.logger.debug("Didn't find download in active downloads list")
 
-    def __download_thread(self, download_queue):
+    def __download_thread(self, download_queue: queue.PriorityQueue) -> None:
         """
         The main DownloadManager thread calls this when it is created
         It checks the queue, starting new downloads when they are available
@@ -265,13 +266,13 @@ def __download_thread(self, download_queue):
                 with self.active_downloads_lock:
                     download = download_queue.get().item
                     self.active_downloads[download] = download
-                self.__download_file(download, download_queue)
+                self.__download_file(download)
                 # Mark the task as done to keep counts correct so
                 # we can use join() or other functions later
                 download_queue.task_done()
             time.sleep(0.01)
 
-    def __download_file(self, download, download_queue):
+    def __download_file(self, download: Download) -> None:
         """
         This is called by __download_thread to download a file
         It is also called directly by the thread created in download_now
@@ -309,7 +310,7 @@ def __download_file(self, download, download_queue):
             # We may want to unset current_downloads here
             # For example, if a download was added that is impossible to complete
 
-    def __prepare_location(self, save_location):
+    def __prepare_location(self, save_location: str) -> None:
         """
         Make sure the download directory exists and the file doesn't already exist
 
@@ -326,7 +327,7 @@ def __prepare_location(self, save_location):
             shutil.rmtree(save_location)
             self.logger.debug("{} is a directory. Will remove it, to make place for installer.".format(save_location))
 
-    def __get_start_point_and_download_mode(self, download):
+    def __get_start_point_and_download_mode(self, download) -> Tuple[int, str]:
         """
         Resume the previous download if possible
 
@@ -344,7 +345,7 @@ def __get_start_point_and_download_mode(self, download):
                 os.remove(download.save_location)
         return start_point, download_mode
 
-    def __download_operation(self, download, start_point, download_mode):
+    def __download_operation(self, download: Download, start_point: int, download_mode: str) -> bool:
         """
         Download the file
         This is called by __download_file to actually perform the download
@@ -384,7 +385,7 @@ def __download_operation(self, download, start_point, download_mode):
         self.logger.debug("Returning result from _download_operation: {}".format(result))
         return result
 
-    def __is_same_download_as_before(self, download):
+    def __is_same_download_as_before(self, download: Download) -> bool:
         """
         Return true if the download is the same as an item with the same save_location
         already downloaded.