Improved type annotations

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Replaced `pyflakes` with `ruff` for linting
 * Refactored logic for resuming large file uploads to unify code paths, correct inconsistencies, and enhance configurability (#381)
 * Automatically set copyright date when generating the docs
+* Use modern type hints in documentation (achieved through combination of PEP 563 & 585 and `sphinx-autodoc-typehints`)
 
 ## [1.21.0] - 2023-04-17
 

b2sdk/__init__.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 # Set default logging handler to avoid "No handler found" warnings.
 import logging

b2sdk/__main__.py

@@ -7,3 +7,4 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations

b2sdk/_pyinstaller/__init__.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 import os
 

b2sdk/_pyinstaller/hook-b2sdk.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 from PyInstaller.utils.hooks import copy_metadata
 

b2sdk/_v3/__init__.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 # this file maps the external interface into internal interface
 # it will come handy if we ever need to move something

b2sdk/_v3/exception.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 from b2sdk.account_info.exception import AccountInfoError
 from b2sdk.account_info.exception import CorruptAccountInfo

b2sdk/account_info/__init__.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 from .in_memory import InMemoryAccountInfo
 

b2sdk/account_info/abstract.py

@@ -7,8 +7,9 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
+
 from abc import abstractmethod
-from typing import Optional, List, Tuple
 
 from b2sdk.account_info import exception
 from b2sdk.raw_api import ALL_CAPABILITIES
@@ -52,7 +53,7 @@ def clear(self):
         """
 
     @abstractmethod
-    def list_bucket_names_ids(self) -> List[Tuple[str, str]]:
+    def list_bucket_names_ids(self) -> list[tuple[str, str]]:
         """
         List buckets in the cache.
 
@@ -96,7 +97,7 @@ def get_bucket_id_or_none_from_bucket_name(self, bucket_name):
         """
 
     @abstractmethod
-    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> Optional[str]:
+    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> str | None:
         """
         Look up the bucket name for the given bucket id.
         """

b2sdk/account_info/exception.py

@@ -7,8 +7,9 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
-from abc import (ABCMeta)
+from abc import ABCMeta
 
 from ..exception import B2Error
 
@@ -30,13 +31,12 @@ def __init__(self, file_name):
         :param file_name: an account info file name
         :type file_name: str
         """
-        super(CorruptAccountInfo, self).__init__()
+        super().__init__()
         self.file_name = file_name
 
     def __str__(self):
-        return 'Account info file (%s) appears corrupted.  Try removing and then re-authorizing the account.' % (
-            self.file_name,
-        )
+        return f'Account info file ({self.file_name}) appears corrupted. ' \
+               f'Try removing and then re-authorizing the account.'
 
 
 class MissingAccountData(AccountInfoError):
@@ -49,8 +49,8 @@ def __init__(self, key):
         :param key: a key for getting account data
         :type key: str
         """
-        super(MissingAccountData, self).__init__()
+        super().__init__()
         self.key = key
 
     def __str__(self):
-        return 'Missing account data: %s' % (self.key,)
+        return f'Missing account data: {self.key}'

b2sdk/account_info/in_memory.py

@@ -7,14 +7,13 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
-from typing import Optional, List, Tuple
+from functools import wraps
 
 from .exception import MissingAccountData
 from .upload_url_pool import UrlPoolAccountInfo
 
-from functools import wraps
-
 
 def _raise_missing_if_result_is_none(function):
     """
@@ -39,12 +38,12 @@ class InMemoryAccountInfo(UrlPoolAccountInfo):
     """
 
     def __init__(self, *args, **kwargs):
-        super(InMemoryAccountInfo, self).__init__(*args, **kwargs)
+        super().__init__(*args, **kwargs)
         self._clear_in_memory_account_fields()
 
     def clear(self):
         self._clear_in_memory_account_fields()
-        return super(InMemoryAccountInfo, self).clear()
+        return super().clear()
 
     def _clear_in_memory_account_fields(self):
         self._account_id = None
@@ -82,13 +81,13 @@ def refresh_entire_bucket_name_cache(self, name_id_iterable):
     def get_bucket_id_or_none_from_bucket_name(self, bucket_name):
         return self._buckets.get(bucket_name)
 
-    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> Optional[str]:
+    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> str | None:
         for name, cached_id_ in self._buckets.items():
             if cached_id_ == bucket_id:
                 return name
         return None
 
-    def list_bucket_names_ids(self) -> List[Tuple[str, str]]:
+    def list_bucket_names_ids(self) -> list[tuple[str, str]]:
         return [(name, id_) for name, id_ in self._buckets.items()]
 
     def save_bucket(self, bucket):

b2sdk/account_info/sqlite_account_info.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 import json
 import logging
@@ -16,8 +17,6 @@
 import stat
 import threading
 
-from typing import List, Optional, Tuple
-
 from .exception import CorruptAccountInfo, MissingAccountData
 from .upload_url_pool import UrlPoolAccountInfo
 
@@ -42,7 +41,7 @@ class SqliteAccountInfo(UrlPoolAccountInfo):
     completed.
     """
 
-    def __init__(self, file_name=None, last_upgrade_to_run=None, profile: Optional[str] = None):
+    def __init__(self, file_name=None, last_upgrade_to_run=None, profile: str | None = None):
         """
         Initialize SqliteAccountInfo.
 
@@ -78,7 +77,7 @@ def __init__(self, file_name=None, last_upgrade_to_run=None, profile: Optional[s
         self._validate_database(last_upgrade_to_run)
         with self._get_connection() as conn:
             self._create_tables(conn, last_upgrade_to_run)
-        super(SqliteAccountInfo, self).__init__()
+        super().__init__()
 
     # dirty trick to use parameters in the docstring
     if getattr(__init__, '__doc__', None):  # don't break when using `python -oo`
@@ -94,7 +93,7 @@ def __init__(self, file_name=None, last_upgrade_to_run=None, profile: Optional[s
     @classmethod
     def _get_user_account_info_path(cls, file_name=None, profile=None):
         if profile and not B2_ACCOUNT_INFO_PROFILE_NAME_REGEXP.match(profile):
-            raise ValueError('Invalid profile name: {}'.format(profile))
+            raise ValueError(f'Invalid profile name: {profile}')
 
         if file_name:
             if profile:
@@ -111,7 +110,7 @@ def _get_user_account_info_path(cls, file_name=None, profile=None):
             user_account_info_path = B2_ACCOUNT_INFO_DEFAULT_FILE
         elif XDG_CONFIG_HOME_ENV_VAR in os.environ:
             config_home = os.environ[XDG_CONFIG_HOME_ENV_VAR]
-            file_name = 'db-{}.sqlite'.format(profile) if profile else 'account_info'
+            file_name = f'db-{profile}.sqlite' if profile else 'account_info'
             user_account_info_path = os.path.join(config_home, 'b2', file_name)
             if not os.path.exists(os.path.join(config_home, 'b2')):
                 os.makedirs(os.path.join(config_home, 'b2'), mode=0o755)
@@ -360,7 +359,7 @@ def _create_tables(self, conn, last_upgrade_to_run):
                 ]
             )
 
-    def _ensure_update(self, update_number, update_commands: List[str]):
+    def _ensure_update(self, update_number, update_commands: list[str]):
         """
         Run the update with the given number if it hasn't been done yet.
 
@@ -554,7 +553,7 @@ def get_s3_api_url(self):
     def _get_account_info_or_raise(self, column_name):
         try:
             with self._get_connection() as conn:
-                cursor = conn.execute('SELECT %s FROM account;' % (column_name,))
+                cursor = conn.execute(f'SELECT {column_name} FROM account;')
                 value = cursor.fetchone()[0]
                 return value
         except Exception as e:
@@ -590,10 +589,10 @@ def get_bucket_id_or_none_from_bucket_name(self, bucket_name):
             'SELECT bucket_id FROM bucket WHERE bucket_name = ?;', (bucket_name,)
         )
 
-    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> Optional[str]:
+    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> str | None:
         return self._safe_query('SELECT bucket_name FROM bucket WHERE bucket_id = ?;', (bucket_id,))
 
-    def list_bucket_names_ids(self) -> List[Tuple[str, str]]:
+    def list_bucket_names_ids(self) -> list[tuple[str, str]]:
         with self._get_connection() as conn:
             cursor = conn.execute('SELECT bucket_name, bucket_id FROM bucket;')
             return cursor.fetchall()

b2sdk/account_info/stub.py

@@ -7,8 +7,8 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
-from typing import Optional, List, Tuple
 import collections
 import threading
 
@@ -73,10 +73,10 @@ def refresh_entire_bucket_name_cache(self, name_id_iterable):
     def get_bucket_id_or_none_from_bucket_name(self, bucket_name):
         return None
 
-    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> Optional[str]:
+    def get_bucket_name_or_none_from_bucket_id(self, bucket_id: str) -> str | None:
         return None
 
-    def list_bucket_names_ids(self) -> List[Tuple[str, str]]:
+    def list_bucket_names_ids(self) -> list[tuple[str, str]]:
         return list((bucket.bucket_name, bucket.bucket_id) for bucket in self.buckets.values())
 
     def save_bucket(self, bucket):

b2sdk/account_info/test_upload_url_concurrency.py

@@ -7,6 +7,7 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
 import os
 import threading

b2sdk/account_info/upload_url_pool.py

@@ -7,10 +7,11 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+from __future__ import annotations
 
-from abc import abstractmethod
 import collections
 import threading
+from abc import abstractmethod
 
 from .abstract import AbstractAccountInfo
 
@@ -79,13 +80,13 @@ class UrlPoolAccountInfo(AbstractAccountInfo):
     )  #: A url pool class to use for large files.
 
     def __init__(self):
-        super(UrlPoolAccountInfo, self).__init__()
+        super().__init__()
         self._reset_upload_pools()
 
     @abstractmethod
     def clear(self):
         self._reset_upload_pools()
-        return super(UrlPoolAccountInfo, self).clear()
+        return super().clear()
 
     def _reset_upload_pools(self):
         self._bucket_uploads = self.BUCKET_UPLOAD_POOL_CLASS()