From 67a5a552fdc04f7ecbb8f582b26afad95bb87d2d Mon Sep 17 00:00:00 2001 From: Safihre Date: Mon, 17 Nov 2025 15:39:47 +0100 Subject: [PATCH] Add missing typing hints to several files --- .github/workflows/integration_testing.yml | 2 +- sabnzbd/config.py | 4 +- sabnzbd/decorators.py | 2 +- sabnzbd/directunpacker.py | 2 +- sabnzbd/filesystem.py | 18 +++---- sabnzbd/interface.py | 2 +- sabnzbd/misc.py | 58 ++++++++++++----------- sabnzbd/newsunpack.py | 12 ++--- sabnzbd/notifier.py | 2 +- sabnzbd/nzbparser.py | 2 +- sabnzbd/sorting.py | 2 +- tests/testhelper.py | 2 +- 12 files changed, 56 insertions(+), 52 deletions(-) diff --git a/.github/workflows/integration_testing.yml b/.github/workflows/integration_testing.yml index 55b887db6..82d4b879d 100644 --- a/.github/workflows/integration_testing.yml +++ b/.github/workflows/integration_testing.yml @@ -20,7 +20,7 @@ jobs: builder/SABnzbd.spec tests --line-length=120 - --target-version=py38 + --target-version=py39 --check --diff diff --git a/sabnzbd/config.py b/sabnzbd/config.py index 488a936e5..cd1ceedf3 100644 --- a/sabnzbd/config.py +++ b/sabnzbd/config.py @@ -307,7 +307,7 @@ class OptionList(Option): self, section: str, keyword: str, - default_val: Union[str, List, None] = None, + default_val: Union[str, list, None] = None, validation: Optional[Callable] = None, add: bool = True, public: bool = True, @@ -318,7 +318,7 @@ class OptionList(Option): default_val = [] super().__init__(section, keyword, default_val, add=add, public=public, protect=protect) - def set(self, value: Union[str, List]) -> Optional[str]: + def set(self, value: Union[str, list]) -> Optional[str]: """Set the list given a comma-separated string or a list""" error = None if value is not None: diff --git a/sabnzbd/decorators.py b/sabnzbd/decorators.py index d1e371447..8edb0d17f 100644 --- a/sabnzbd/decorators.py +++ b/sabnzbd/decorators.py @@ -70,7 +70,7 @@ def conditional_cache(cache_time: int): Empty results (None, empty collections, empty strings, False, 0) are not cached. If a keyword argument of `force=True` is used, the cache is skipped. - Unhashable types (such as List) can not be used as an input to the wrapped function in the current implementation! + Unhashable types (such as list) can not be used as an input to the wrapped function in the current implementation! :param cache_time: Time in seconds to cache non-empty results """ diff --git a/sabnzbd/directunpacker.py b/sabnzbd/directunpacker.py index e8a1e62a7..5772daed9 100644 --- a/sabnzbd/directunpacker.py +++ b/sabnzbd/directunpacker.py @@ -25,7 +25,7 @@ import subprocess import time import threading import logging -from typing import Optional, Dict, List, Tuple +from typing import Optional import sabnzbd import sabnzbd.cfg as cfg diff --git a/sabnzbd/filesystem.py b/sabnzbd/filesystem.py index 2734b8423..f73e1021f 100644 --- a/sabnzbd/filesystem.py +++ b/sabnzbd/filesystem.py @@ -33,7 +33,7 @@ import fnmatch import stat import ctypes import random -from typing import Union, List, Tuple, Any, Dict, Optional, BinaryIO +from typing import Union, Any, Optional, BinaryIO try: import win32api @@ -330,7 +330,7 @@ def sanitize_files(folder: Optional[str] = None, filelist: Optional[list[str]] = return output_filelist -def strip_extensions(name: str, ext_to_remove: tuple[str, ...] = (".nzb", ".par", ".par2")): +def strip_extensions(name: str, ext_to_remove: tuple[str, ...] = (".nzb", ".par", ".par2")) -> str: """Strip extensions from a filename, without sanitizing the filename""" name_base, ext = os.path.splitext(name) while ext.lower() in ext_to_remove: @@ -613,7 +613,7 @@ def make_script_path(script: str) -> Optional[str]: return script_path -def get_admin_path(name: str, future: bool): +def get_admin_path(name: str, future: bool) -> str: """Return news-style full path to job-admin folder of names job or else the old cache path """ @@ -660,7 +660,7 @@ def set_permissions(path: str, recursive: bool = True): UNWANTED_FILE_PERMISSIONS = stat.S_ISUID | stat.S_ISGID | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH -def removexbits(path: str, custom_permissions: int = None): +def removexbits(path: str, custom_permissions: Optional[int] = None): """Remove all the x-bits from files, respecting current or custom permissions""" if os.path.isfile(path): # Use custom permissions as base @@ -1033,7 +1033,7 @@ def diskspace(force: bool = False) -> dict[str, tuple[float, float]]: } -def get_new_id(prefix, folder, check_list=None): +def get_new_id(prefix: str, folder: str, check_list: Optional[list] = None) -> str: """Return unique prefixed admin identifier within folder optionally making sure that id is not in the check_list. """ @@ -1054,7 +1054,7 @@ def get_new_id(prefix, folder, check_list=None): raise IOError -def save_data(data, _id, path, do_pickle=True, silent=False): +def save_data(data: Any, _id: str, path: str, do_pickle: bool = True, silent: bool = False): """Save data to a diskfile""" if not silent: logging.debug("[%s] Saving data for %s in %s", sabnzbd.misc.caller_name(), _id, path) @@ -1081,7 +1081,7 @@ def save_data(data, _id, path, do_pickle=True, silent=False): time.sleep(0.1) -def load_data(data_id, path, remove=True, do_pickle=True, silent=False): +def load_data(data_id: str, path: str, remove: bool = True, do_pickle: bool = True, silent: bool = False) -> Any: """Read data from disk file""" path = os.path.join(path, data_id) @@ -1129,7 +1129,7 @@ def save_admin(data: Any, data_id: str): save_data(data, data_id, sabnzbd.cfg.admin_dir.get_path()) -def load_admin(data_id: str, remove=False, silent=False) -> Any: +def load_admin(data_id: str, remove: bool = False, silent: bool = False) -> Any: """Read data in admin folder in specified format""" logging.debug("[%s] Loading data for %s", sabnzbd.misc.caller_name(), data_id) return load_data(data_id, sabnzbd.cfg.admin_dir.get_path(), remove=remove, silent=silent) @@ -1196,7 +1196,7 @@ def purge_log_files(): logging.debug("Finished puring log files") -def directory_is_writable_with_file(mydir, myfilename): +def directory_is_writable_with_file(mydir: str, myfilename: str) -> bool: filename = os.path.join(mydir, myfilename) if os.path.exists(filename): try: diff --git a/sabnzbd/interface.py b/sabnzbd/interface.py index 9fbcb8dbe..050601fbe 100644 --- a/sabnzbd/interface.py +++ b/sabnzbd/interface.py @@ -34,7 +34,7 @@ import copy from random import randint from xml.sax.saxutils import escape from Cheetah.Template import Template -from typing import Optional, Callable, Union, Any, Dict, List +from typing import Optional, Callable, Union, Any from guessit.api import properties as guessit_properties import sabnzbd diff --git a/sabnzbd/misc.py b/sabnzbd/misc.py index 4631e5f0b..03781485c 100644 --- a/sabnzbd/misc.py +++ b/sabnzbd/misc.py @@ -41,7 +41,7 @@ import math import rarfile from threading import Thread from collections.abc import Iterable -from typing import Union, Tuple, Any, AnyStr, Optional, List, Dict, Collection +from typing import Union, Tuple, Any, AnyStr, Optional, Collection import sabnzbd import sabnzbd.getipaddress @@ -190,7 +190,7 @@ def clean_comma_separated_list(inp: Any) -> list[str]: return result_ids -def cmp(x, y): +def cmp(x: Any, y: Any) -> int: """ Replacement for built-in function cmp that was removed in Python 3 @@ -234,7 +234,7 @@ def cat_pp_script_sanitizer( return cat, pp, script -def name_to_cat(fname, cat=None): +def name_to_cat(fname: str, cat: Optional[str] = None) -> tuple[str, Optional[str]]: """Retrieve category from file name, but only if "cat" is None.""" if cat is None and fname.startswith("{{"): n = fname.find("}}") @@ -246,7 +246,9 @@ def name_to_cat(fname, cat=None): return fname, cat -def cat_to_opts(cat, pp=None, script=None, priority=None) -> tuple[str, int, str, int]: +def cat_to_opts( + cat: Optional[str], pp: Optional[int] = None, script: Optional[str] = None, priority: Optional[int] = None +) -> tuple[str, int, str, int]: """Derive options from category, if options not already defined. Specified options have priority over category-options. If no valid category is given, special category '*' will supply default values @@ -331,12 +333,12 @@ _wildcard_to_regex = { } -def wildcard_to_re(text): +def wildcard_to_re(text: str) -> str: """Convert plain wildcard string (with '*' and '?') to regex.""" return "".join([_wildcard_to_regex.get(ch, ch) for ch in text]) -def convert_filter(text): +def convert_filter(text: str) -> Optional[re.Pattern]: """Return compiled regex. If string starts with re: it's a real regex else quote all regex specials, replace '*' by '.*' @@ -353,7 +355,7 @@ def convert_filter(text): return None -def cat_convert(cat): +def cat_convert(cat: Optional[str]) -> Optional[str]: """Convert indexer's category/group-name to user categories. If no match found, but indexer-cat equals user-cat, then return user-cat If no match found, but the indexer-cat starts with the user-cat, return user-cat @@ -397,7 +399,7 @@ _SERVICE_KEY = "SYSTEM\\CurrentControlSet\\services\\" _SERVICE_PARM = "CommandLine" -def get_serv_parms(service): +def get_serv_parms(service: str) -> list[str]: """Get the service command line parameters from Registry""" service_parms = [] try: @@ -416,7 +418,7 @@ def get_serv_parms(service): return service_parms -def set_serv_parms(service, args): +def set_serv_parms(service: str, args: list) -> bool: """Set the service command line parameters in Registry""" serv = [] for arg in args: @@ -444,7 +446,7 @@ def get_from_url(url: str) -> Optional[str]: return None -def convert_version(text): +def convert_version(text: str) -> tuple[int, bool]: """Convert version string to numerical value and a testversion indicator""" version = 0 test = True @@ -551,7 +553,7 @@ def check_latest_version(): ) -def upload_file_to_sabnzbd(url, fp): +def upload_file_to_sabnzbd(url: str, fp: str): """Function for uploading nzbs to a running SABnzbd instance""" try: fp = urllib.parse.quote_plus(fp) @@ -644,7 +646,7 @@ def to_units(val: Union[int, float], postfix="") -> str: return f"{sign}{val:.{decimals}f}{units}" -def caller_name(skip=2): +def caller_name(skip: int = 2) -> str: """Get a name of a caller in the format module.method Originally used: https://gist.github.com/techtonik/2151727 Adapted for speed by using sys calls directly @@ -682,7 +684,7 @@ def exit_sab(value: int): os._exit(value) -def split_host(srv): +def split_host(srv: Optional[str]) -> tuple[Optional[str], Optional[int]]: """Split host:port notation, allowing for IPV6""" if not srv: return None, None @@ -704,7 +706,7 @@ def split_host(srv): return out[0], port -def get_cache_limit(): +def get_cache_limit() -> str: """Depending on OS, calculate cache limits. In ArticleCache it will make sure we stay within system limits for 32/64 bit @@ -742,7 +744,7 @@ def get_cache_limit(): return "" -def get_windows_memory(): +def get_windows_memory() -> int: """Use ctypes to extract available memory""" class MEMORYSTATUSEX(ctypes.Structure): @@ -768,14 +770,14 @@ def get_windows_memory(): return stat.ullTotalPhys -def get_macos_memory(): +def get_macos_memory() -> float: """Use system-call to extract total memory on macOS""" system_output = run_command(["sysctl", "hw.memsize"]) return float(system_output.split()[1]) @conditional_cache(cache_time=3600) -def get_cpu_name(): +def get_cpu_name() -> Optional[str]: """Find the CPU name (which needs a different method per OS), and return it If none found, return platform.platform()""" @@ -875,7 +877,7 @@ def on_cleanup_list(filename: str, skip_nzb: bool = False) -> bool: return False -def memory_usage(): +def memory_usage() -> Optional[str]: try: # Probably only works on Linux because it uses /proc//statm with open("/proc/%d/statm" % os.getpid()) as t: @@ -897,7 +899,7 @@ except Exception: _HAVE_STATM = _PAGE_SIZE and memory_usage() -def loadavg(): +def loadavg() -> str: """Return 1, 5 and 15 minute load average of host or "" if not supported""" p = "" if not sabnzbd.WINDOWS and not sabnzbd.MACOS: @@ -972,7 +974,7 @@ def bool_conv(value: Any) -> bool: return bool(int_conv(value)) -def create_https_certificates(ssl_cert, ssl_key): +def create_https_certificates(ssl_cert: str, ssl_key: str) -> bool: """Create self-signed HTTPS certificates and store in paths 'ssl_cert' and 'ssl_key'""" try: from sabnzbd.utils.certgen import generate_key, generate_local_cert @@ -1051,7 +1053,7 @@ def is_sample(filename: str) -> bool: return bool(re.search(RE_SAMPLE, filename)) -def find_on_path(targets): +def find_on_path(targets: Union[str, tuple[str, ...]]) -> Optional[str]: """Search the PATH for a program and return full path""" if sabnzbd.WINDOWS: paths = os.getenv("PATH").split(";") @@ -1224,7 +1226,7 @@ def match_str(text: AnyStr, matches: tuple[AnyStr, ...]) -> Optional[AnyStr]: return None -def recursive_html_escape(input_dict_or_list: Union[dict[str, Any], List], exclude_items: tuple[str, ...] = ()): +def recursive_html_escape(input_dict_or_list: Union[dict[str, Any], list], exclude_items: tuple[str, ...] = ()): """Recursively update the input_dict in-place with html-safe values""" if isinstance(input_dict_or_list, (dict, list)): if isinstance(input_dict_or_list, dict): @@ -1259,7 +1261,9 @@ def list2cmdline_unrar(lst: list[str]) -> str: return " ".join(nlst) -def build_and_run_command(command: list[str], windows_unrar_command: bool = False, text_mode: bool = True, **kwargs): +def build_and_run_command( + command: list[str], windows_unrar_command: bool = False, text_mode: bool = True, **kwargs +) -> subprocess.Popen: """Builds and then runs command with necessary flags and optional IONice and Nice commands. Optional Popen arguments can be supplied. On Windows we need to run our own list2cmdline for Unrar. @@ -1326,7 +1330,7 @@ def build_and_run_command(command: list[str], windows_unrar_command: bool = Fals return subprocess.Popen(command, **popen_kwargs) -def run_command(cmd: list[str], **kwargs): +def run_command(cmd: list[str], **kwargs) -> str: """Run simple external command and return output as a string.""" with build_and_run_command(cmd, **kwargs) as p: txt = p.stdout.read() @@ -1359,7 +1363,7 @@ def set_socks5_proxy(): socket.socket = socks.socksocket -def set_https_verification(value): +def set_https_verification(value: bool) -> bool: """Set HTTPS-verification state while returning current setting False = disable verification """ @@ -1381,7 +1385,7 @@ def request_repair(): pass -def check_repair_request(): +def check_repair_request() -> bool: """Return True if repair request found, remove afterwards""" path = os.path.join(cfg.admin_dir.get_path(), REPAIR_REQUEST) if os.path.exists(path): @@ -1615,7 +1619,7 @@ class SABRarFile(rarfile.RarFile): self._file_parser._info_list.append(rar_obj) self._file_parser._info_map[rar_obj.filename.rstrip("/")] = rar_obj - def filelist(self): + def filelist(self) -> list[str]: """Return list of filenames in archive.""" return [f.filename for f in self.infolist() if not f.isdir()] diff --git a/sabnzbd/newsunpack.py b/sabnzbd/newsunpack.py index 7dd905319..d7f545ce3 100644 --- a/sabnzbd/newsunpack.py +++ b/sabnzbd/newsunpack.py @@ -29,7 +29,7 @@ import io import shutil import functools import rarfile -from typing import BinaryIO, Optional, Any, Union +from typing import BinaryIO, Optional, Any, Union, Callable import sabnzbd from sabnzbd.encoding import correct_unknown_encoding, ubtou @@ -866,7 +866,7 @@ def rar_extract_core( ############################################################################## # 7Zip Functions ############################################################################## -def unseven(nzo: NzbObject, workdir_complete: str, one_folder: bool, sevens: list[str]): +def unseven(nzo: NzbObject, workdir_complete: str, one_folder: bool, sevens: list[str]) -> tuple[bool, list[str]]: """Unpack multiple sets '7z' of 7Zip files from 'download_path' to 'workdir_complete. When 'delete' is set, originals will be deleted. """ @@ -1762,7 +1762,7 @@ def sfv_check(sfvs: list[str], nzo: NzbObject) -> bool: return result -def parse_sfv(sfv_filename): +def parse_sfv(sfv_filename: str) -> dict[str, bytes]: """Parse SFV file and return dictionary of crc32's and filenames""" results = {} with open(sfv_filename, mode="rb") as sfv_list: @@ -1787,12 +1787,12 @@ def add_time_left(perc: float, start_time: Optional[float] = None, time_used: Op return "" -def pre_queue(nzo: NzbObject, pp, cat): +def pre_queue(nzo: NzbObject, pp: str, cat: str) -> list[Any]: """Run pre-queue script (if any) and process results. pp and cat are supplied separate since they can change. """ - def fix(p): + def fix(p: Any) -> str: # If added via API, some items can still be "None" (as a string) if is_none(p): return "" @@ -1909,6 +1909,6 @@ class SevenZip: p.wait() return data - def close(self): + def close(self) -> None: """Close file""" pass diff --git a/sabnzbd/notifier.py b/sabnzbd/notifier.py index 93cc7df20..75c29782e 100644 --- a/sabnzbd/notifier.py +++ b/sabnzbd/notifier.py @@ -31,7 +31,7 @@ import http.client import json import apprise from threading import Thread -from typing import Optional, Dict, Union +from typing import Optional, Union import sabnzbd import sabnzbd.cfg diff --git a/sabnzbd/nzbparser.py b/sabnzbd/nzbparser.py index b5644b85c..5bd750a4c 100644 --- a/sabnzbd/nzbparser.py +++ b/sabnzbd/nzbparser.py @@ -30,7 +30,7 @@ import zipfile import tempfile import cherrypy._cpreqbody -from typing import Optional, Dict, Any, Union, List, Tuple +from typing import Optional, Any, Union import sabnzbd from sabnzbd import nzbstuff diff --git a/sabnzbd/sorting.py b/sabnzbd/sorting.py index 6b6136053..52a5fe794 100644 --- a/sabnzbd/sorting.py +++ b/sabnzbd/sorting.py @@ -25,7 +25,7 @@ import re import guessit from rebulk.match import MatchesDict from string import whitespace, punctuation -from typing import Optional, Union, List, Tuple, Dict +from typing import Optional, Union import sabnzbd from sabnzbd.filesystem import ( diff --git a/tests/testhelper.py b/tests/testhelper.py index b36ed89bf..1f32d248a 100644 --- a/tests/testhelper.py +++ b/tests/testhelper.py @@ -22,7 +22,7 @@ import io import os import time from http.client import RemoteDisconnected -from typing import BinaryIO, Optional, Dict, List +from typing import BinaryIO, Optional import pytest from random import choice, randint