add typing hints/annotations

.github/workflows/python-app.yml

@@ -35,7 +35,7 @@ jobs:
       matrix:
         QBT_VER: [4.4.0beta3, 4.3.8, 4.3.7, 4.3.6, 4.3.5, 4.3.4.1, 4.3.3, 4.3.2, 4.3.1, 4.3.0.1, 4.2.5, 4.2.0]
         # QBT_VER: [4.4.0beta3]
-        python-version: [3.9, 3.8, 3.7, 3.6, 2.7, 3.10, pypy2, pypy3]
+        python-version: ["3.10", "3.9", "3.8", "3.7", "3.6", "2.7", "pypy2", "pypy3"]
         # python-version: [3.9]
 
     # TODO: each step currently has an over-complicated conditional to prevent always running all tests.

qbittorrentapi/_attrdict.pyi

@@ -0,0 +1,17 @@
+import abc
+from collections import Mapping, MutableMapping
+
+def merge(left, right): ...
+
+class Attr(Mapping, metaclass=abc.ABCMeta):
+    def __call__(self, key): ...
+    def __getattr__(self, key): ...
+    def __add__(self, other): ...
+    def __radd__(self, other): ...
+
+class MutableAttr(Attr, MutableMapping, metaclass=abc.ABCMeta):
+    def __setattr__(self, key, value) -> None: ...
+    def __delattr__(self, key, force: bool = ...) -> None: ...
+
+class AttrDict(dict, MutableAttr):
+    def __init__(self, *args, **kwargs) -> None: ...

qbittorrentapi/app.py

@@ -138,9 +138,9 @@ def app_web_api_version(self, **kwargs):
 
         :return: string
         """
-        if self._MOCK_WEB_API_VERSION:
-            return self._MOCK_WEB_API_VERSION
-        return self._get(_name=APINames.Application, _method="webapiVersion", **kwargs)
+        return self._MOCK_WEB_API_VERSION or self._get(
+            _name=APINames.Application, _method="webapiVersion", **kwargs
+        )
 
     @endpoint_introduced("2.3", "app/buildInfo")
     @response_json(BuildInfoDictionary)

qbittorrentapi/app.pyi

@@ -0,0 +1,40 @@
+from typing import Dict, Text
+
+from qbittorrentapi.definitions import ClientCache
+from qbittorrentapi.definitions import Dictionary
+from qbittorrentapi.request import Request
+
+class ApplicationPreferencesDictionary(Dictionary): ...
+class BuildInfoDictionary(Dictionary): ...
+
+class Application(ClientCache):
+    @property
+    def version(self) -> Text: ...
+    @property
+    def web_api_version(self) -> Text: ...
+    webapiVersion = web_api_version
+    @property
+    def build_info(self) -> BuildInfoDictionary: ...
+    buildInfo = build_info
+    def shutdown(self) -> None: ...
+    @property
+    def preferences(self) -> ApplicationPreferencesDictionary: ...
+    @preferences.setter
+    def preferences(self, v: Dict) -> None: ...
+    def set_preferences(self, prefs: Dict = None, **kwargs) -> None: ...
+    setPreferences = set_preferences
+    @property
+    def default_save_path(self, **kwargs) -> Text: ...
+    defaultSavePath = default_save_path
+
+class AppAPIMixIn(Request):
+    @property
+    def app(self) -> Application: ...
+    application = app
+    def app_version(self, **kwargs) -> Text: ...
+    def app_web_api_version(self, **kwargs) -> Text: ...
+    def app_build_info(self, **kwargs) -> BuildInfoDictionary: ...
+    def app_shutdown(self, **kwargs) -> None: ...
+    def app_preferences(self, **kwargs) -> ApplicationPreferencesDictionary: ...
+    def app_set_preferences(self, prefs: Dict = None, **kwargs) -> None: ...
+    def app_default_save_path(self, **kwargs) -> Text: ...

qbittorrentapi/auth.py

@@ -42,7 +42,7 @@ class AuthAPIMixIn(Request):
     :Usage:
         >>> from qbittorrentapi import Client
         >>> client = Client(host='localhost:8080', username='admin', password='adminadmin')
-        >>> client.auth_is_logged_in()
+        >>> _ = client.is_logged_in
         >>> client.auth_log_in(username='admin', password='adminadmin')
         >>> client.auth_log_out()
     """

qbittorrentapi/auth.pyi

@@ -0,0 +1,24 @@
+from typing import Text
+
+from qbittorrentapi.definitions import ClientCache
+from qbittorrentapi.request import Request
+
+class Authorization(ClientCache):
+    @property
+    def is_logged_in(self) -> bool: ...
+    def log_in(self, username: Text = None, password: Text = None, **kwargs): ...
+    def log_out(self, **kwargs) -> None: ...
+
+class AuthAPIMixIn(Request):
+    @property
+    def auth(self) -> Authorization: ...
+    authorization = auth
+    @property
+    def is_logged_in(self) -> bool: ...
+    username: Text
+    def auth_log_in(
+        self, username: Text = None, password: Text = None, **kwargs
+    ) -> None: ...
+    @property
+    def _SID(self) -> Text | None: ...
+    def auth_log_out(self, **kwargs) -> None: ...

qbittorrentapi/client.pyi

@@ -0,0 +1,29 @@
+from typing import Text
+
+from qbittorrentapi.app import AppAPIMixIn
+from qbittorrentapi.auth import AuthAPIMixIn
+from qbittorrentapi.log import LogAPIMixIn
+from qbittorrentapi.rss import RSSAPIMixIn
+from qbittorrentapi.search import SearchAPIMixIn
+from qbittorrentapi.sync import SyncAPIMixIn
+from qbittorrentapi.torrents import TorrentsAPIMixIn
+from qbittorrentapi.transfer import TransferAPIMixIn
+
+class Client(
+    AppAPIMixIn,
+    AuthAPIMixIn,
+    LogAPIMixIn,
+    SyncAPIMixIn,
+    TransferAPIMixIn,
+    TorrentsAPIMixIn,
+    RSSAPIMixIn,
+    SearchAPIMixIn,
+):
+    def __init__(
+        self,
+        host: Text = "",
+        port: Text | int = None,
+        username: Text = None,
+        password: Text = None,
+        **kwargs
+    ) -> None: ...

qbittorrentapi/decorators.py

@@ -2,30 +2,13 @@
 from json import loads
 from logging import getLogger
 
-from pkg_resources import parse_version
-
 from qbittorrentapi.exceptions import APIError
 from qbittorrentapi.exceptions import HTTP403Error
+from qbittorrentapi.request import HelpersMixIn
 
 logger = getLogger(__name__)
 
 
-def _is_version_less_than(ver1, ver2, lteq=True):
-    """
-    Determine if ver1 is equal to or later than ver2.
-
-    Note: changes need to be reflected in request.Request._is_version_less_than as well
-
-    :param ver1: version to check
-    :param ver2: current version of application
-    :param lteq: True for Less Than or Equals; False for just Less Than
-    :return: True or False
-    """
-    if lteq:
-        return parse_version(ver1) <= parse_version(ver2)
-    return parse_version(ver1) < parse_version(ver2)
-
-
 class Alias(object):
 
     """
@@ -196,7 +179,9 @@ def _version_too_old(client, version_to_compare):
     Return True if provided API version is older than the connected API, False otherwise
     """
     current_version = client.app_web_api_version()
-    if _is_version_less_than(current_version, version_to_compare, lteq=False):
+    if HelpersMixIn._is_version_less_than(
+        current_version, version_to_compare, lteq=False
+    ):
         return True
     return False
 
@@ -206,7 +191,9 @@ def _version_too_new(client, version_to_compare):
     Return True if provided API version is newer or equal to the connected API, False otherwise
     """
     current_version = client.app_web_api_version()
-    if _is_version_less_than(version_to_compare, current_version, lteq=True):
+    if HelpersMixIn._is_version_less_than(
+        version_to_compare, current_version, lteq=True
+    ):
         return True
     return False
 

qbittorrentapi/decorators.pyi

@@ -0,0 +1,20 @@
+from typing import Any, Callable, Text
+
+from qbittorrentapi.client import Client
+from qbittorrentapi.exceptions import APIError as APIError, HTTP403Error as HTTP403Error
+
+class Alias:
+    aliases: Any
+    def __init__(self, *aliases: Text) -> None: ...
+    def __call__(self, f: Callable) -> Callable: ...
+
+def aliased(aliased_class: Text): ...
+def login_required(func: Callable): ...
+def handle_hashes(func: Callable): ...
+def response_text(response_class): ...
+def response_json(response_class): ...
+def _version_too_old(client: Client, version_to_compare: Text) -> bool: ...
+def _version_too_new(client: Client, version_to_compare: Text) -> bool: ...
+def _check_for_raise(client: Client, error_message: Text) -> None: ...
+def endpoint_introduced(version_introduced: Text, endpoint: Text): ...
+def version_removed(version_obsoleted: Text, endpoint: Text): ...

qbittorrentapi/definitions.pyi

@@ -0,0 +1,74 @@
+from enum import Enum
+from typing import Dict, List, Text, Tuple, TypeVar
+
+try:
+    from collections import UserList
+except ImportError:
+    from UserList import UserList
+
+from qbittorrentapi._attrdict import AttrDict as AttrDict
+from qbittorrentapi.client import Client
+
+class APINames(Enum):
+    Authorization: Text
+    Application: Text
+    Log: Text
+    Sync: Text
+    Transfer: Text
+    Torrents: Text
+    RSS: Text
+    Search: Text
+    EMPTY: Text
+
+class TorrentStates(Enum):
+    ERROR: Text
+    MISSING_FILES: Text
+    UPLOADING: Text
+    PAUSED_UPLOAD: Text
+    QUEUED_UPLOAD: Text
+    STALLED_UPLOAD: Text
+    CHECKING_UPLOAD: Text
+    FORCED_UPLOAD: Text
+    ALLOCATING: Text
+    DOWNLOADING: Text
+    METADATA_DOWNLOAD: Text
+    FORCED_METADATA_DOWNLOAD: Text
+    PAUSED_DOWNLOAD: Text
+    QUEUED_DOWNLOAD: Text
+    FORCED_DOWNLOAD: Text
+    STALLED_DOWNLOAD: Text
+    CHECKING_DOWNLOAD: Text
+    CHECKING_RESUME_DATA: Text
+    MOVING: Text
+    UNKNOWN: Text
+    @property
+    def is_downloading(self) -> bool: ...
+    @property
+    def is_uploading(self) -> bool: ...
+    @property
+    def is_complete(self) -> bool: ...
+    @property
+    def is_checking(self) -> bool: ...
+    @property
+    def is_errored(self) -> bool: ...
+    @property
+    def is_paused(self) -> bool: ...
+
+class ClientCache:
+    _client: Client
+    def __init__(self, *args, **kwargs) -> None: ...
+
+class Dictionary(ClientCache, AttrDict):
+    def __init__(self, data: Dict = None, client: Client = None): ...
+
+EntryClassT = TypeVar("EntryClassT", bound=Dictionary)
+
+class List(ClientCache, UserList):
+    def __init__(
+        self,
+        list_entries: List | Tuple = None,
+        entry_class: EntryClassT = None,
+        client: Client = None,
+    ) -> None: ...
+
+class ListEntry(Dictionary): ...

qbittorrentapi/exceptions.pyi

@@ -0,0 +1,27 @@
+from requests.exceptions import HTTPError as RequestsHTTPError, RequestException
+
+class APIError(Exception): ...
+class FileError(IOError, APIError): ...
+class TorrentFileError(FileError): ...
+class TorrentFileNotFoundError(TorrentFileError): ...
+class TorrentFilePermissionError(TorrentFileError): ...
+class APIConnectionError(RequestException, APIError): ...
+class LoginFailed(APIConnectionError): ...
+class HTTPError(RequestsHTTPError, APIConnectionError): ...
+class HTTP4XXError(HTTPError): ...
+class HTTP5XXError(HTTPError): ...
+class HTTP400Error(HTTP4XXError): ...
+class HTTP401Error(HTTP4XXError): ...
+class HTTP403Error(HTTP4XXError): ...
+class HTTP404Error(HTTP4XXError): ...
+class HTTP409Error(HTTP4XXError): ...
+class HTTP415Error(HTTP4XXError): ...
+class HTTP500Error(HTTP5XXError): ...
+class MissingRequiredParameters400Error(HTTP400Error): ...
+class InvalidRequest400Error(HTTP400Error): ...
+class Unauthorized401Error(HTTP401Error): ...
+class Forbidden403Error(HTTP403Error): ...
+class NotFound404Error(HTTP404Error): ...
+class Conflict409Error(HTTP409Error): ...
+class UnsupportedMediaType415Error(HTTP415Error): ...
+class InternalServerError500Error(HTTP500Error): ...

qbittorrentapi/log.pyi

@@ -0,0 +1,66 @@
+from typing import Iterable, Text
+
+from qbittorrentapi.definitions import ClientCache
+from qbittorrentapi.definitions import List
+from qbittorrentapi.definitions import ListEntry
+from qbittorrentapi.request import Request
+
+class LogPeersList(List):
+    def __init__(
+        self, list_entries: Iterable = None, client: Request = None
+    ) -> None: ...
+
+class LogPeer(ListEntry): ...
+
+class LogMainList(List):
+    def __init__(
+        self, list_entries: Iterable = None, client: Request = None
+    ) -> None: ...
+
+class LogEntry(ListEntry): ...
+
+class Log(ClientCache):
+    main: _Main
+    def __init__(self, client: Request) -> None: ...
+    def peers(self, last_known_id: Text | int = None, **kwargs) -> LogPeersList[LogPeer]: ...
+    class _Main(ClientCache):
+        def _api_call(
+            self,
+            normal: bool = None,
+            info: bool = None,
+            warning: bool = None,
+            critical: bool = None,
+            last_known_id: bool = None,
+            **kwargs
+        ) -> LogMainList[LogEntry]: ...
+        def __call__(
+            self,
+            normal: bool = None,
+            info: bool = None,
+            warning: bool = None,
+            critical: bool = None,
+            last_known_id: bool = None,
+            **kwargs
+        ) -> LogMainList[LogEntry]: ...
+        def info(self, last_known_id: Text | int = None, **kwargs) -> LogMainList[LogEntry]: ...
+        def normal(self, last_known_id: Text | int = None, **kwargs) -> LogMainList[LogEntry]: ...
+        def warning(
+            self, last_known_id: Text | int = None, **kwargs
+        ) -> LogMainList[LogEntry]: ...
+        def critical(
+            self, last_known_id: Text | int = None, **kwargs
+        ) -> LogMainList[LogEntry]: ...
+
+class LogAPIMixIn(Request):
+    @property
+    def log(self) -> Log: ...
+    def log_main(
+        self,
+        normal: bool = None,
+        info: bool = None,
+        warning: bool = None,
+        critical: bool = None,
+        last_known_id: bool = None,
+        **kwargs
+    ) -> LogMainList[LogEntry]: ...
+    def log_peers(self, last_known_id: Text | int = None, **kwargs) -> LogPeersList[LogPeer]: ...