refactor: migrate all type annotations to PEP 585

hathor/api_util.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import re
-from typing import Any, Dict, List, Optional, Tuple, TypeVar, Union, cast
+from typing import Any, Optional, TypeVar, Union, cast
 
 from twisted.web.http import Request
 from twisted.web.resource import Resource as TwistedResource
@@ -57,7 +57,7 @@ def get_missing_params_msg(param_name):
     return json_dumpb({'success': False, 'message': f'Missing parameter: {param_name}'})
 
 
-def parse_args(args: Dict[bytes, List[bytes]], expected_args: List[str]) -> Dict[str, Any]:
+def parse_args(args: dict[bytes, list[bytes]], expected_args: list[str]) -> dict[str, Any]:
     """Parse all expected arguments. If there are missing arguments, returns the missing arguments
     """
     expected_set = set(expected_args)
@@ -70,7 +70,7 @@ def parse_args(args: Dict[bytes, List[bytes]], expected_args: List[str]) -> Dict
     if diff:
         return {'success': False, 'missing': ', '.join(sorted(diff))}
 
-    ret: Dict[str, str] = dict()
+    ret: dict[str, str] = dict()
     for arg2 in expected_args:
         key_str = arg2.encode('utf-8')
         first_param = args[key_str][0]
@@ -93,7 +93,7 @@ def parse_int(raw: Union[str, bytes], *,
     return value
 
 
-def validate_tx_hash(hash_hex: str, tx_storage: TransactionStorage) -> Tuple[bool, str]:
+def validate_tx_hash(hash_hex: str, tx_storage: TransactionStorage) -> tuple[bool, str]:
     """ Validate if the tx hash is valid and if it exists
         Return success and a message in case of failure
     """
@@ -118,18 +118,18 @@ def validate_tx_hash(hash_hex: str, tx_storage: TransactionStorage) -> Tuple[boo
 
 
 class Resource(TwistedResource):
-    openapi: Dict[str, Any] = {}
+    openapi: dict[str, Any] = {}
 
 
-def get_args(request: Request) -> Dict[bytes, List[bytes]]:
+def get_args(request: Request) -> dict[bytes, list[bytes]]:
     """Type-friendly way to access request.args, also always returns a dict instead of None."""
-    args = cast(Optional[Dict[bytes, List[bytes]]], request.args)
+    args = cast(Optional[dict[bytes, list[bytes]]], request.args)
     if args is None:
         return {}
     return args
 
 
-def get_arg_default(args: Dict[bytes, List[bytes]], key: str, default: T) -> T:
+def get_arg_default(args: dict[bytes, list[bytes]], key: str, default: T) -> T:
     """Get a value with given key from an request.args formatted dict, return default if key was not found.
 
     Examples:

hathor/builder/builder.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from enum import Enum
-from typing import Any, Dict, List, NamedTuple, Optional, Set
+from typing import Any, NamedTuple, Optional
 
 from structlog import get_logger
 
@@ -75,8 +75,8 @@ def __init__(self) -> None:
 
         self._settings: HathorSettingsType = HathorSettings()
         self._rng: Random = Random()
-        self._checkpoints: Optional[List[Checkpoint]] = None
-        self._capabilities: Optional[List[str]] = None
+        self._checkpoints: Optional[list[Checkpoint]] = None
+        self._capabilities: Optional[list[str]] = None
 
         self._peer_id: Optional[PeerId] = None
         self._network: Optional[str] = None
@@ -116,7 +116,7 @@ def __init__(self) -> None:
 
         self._full_verification: Optional[bool] = None
 
-        self._soft_voided_tx_ids: Optional[Set[bytes]] = None
+        self._soft_voided_tx_ids: Optional[set[bytes]] = None
 
     def build(self) -> BuildArtifacts:
         if self.artifacts is not None:
@@ -151,7 +151,7 @@ def build(self) -> BuildArtifacts:
         if self._enable_utxo_index:
             indexes.enable_utxo_index()
 
-        kwargs: Dict[str, Any] = {}
+        kwargs: dict[str, Any] = {}
 
         if self._full_verification is not None:
             kwargs['full_verification'] = self._full_verification
@@ -214,12 +214,12 @@ def set_rng(self, rng: Random) -> 'Builder':
         self._rng = rng
         return self
 
-    def set_checkpoints(self, checkpoints: List[Checkpoint]) -> 'Builder':
+    def set_checkpoints(self, checkpoints: list[Checkpoint]) -> 'Builder':
         self.check_if_can_modify()
         self._checkpoints = checkpoints
         return self
 
-    def set_capabilities(self, capabilities: List[str]) -> 'Builder':
+    def set_capabilities(self, capabilities: list[str]) -> 'Builder':
         self.check_if_can_modify()
         self._capabilities = capabilities
         return self
@@ -237,7 +237,7 @@ def _get_reactor(self) -> Reactor:
             return self._reactor
         raise ValueError('reactor not set')
 
-    def _get_soft_voided_tx_ids(self) -> Set[bytes]:
+    def _get_soft_voided_tx_ids(self) -> set[bytes]:
         if self._soft_voided_tx_ids is not None:
             return self._soft_voided_tx_ids
 
@@ -510,7 +510,7 @@ def disable_full_verification(self) -> 'Builder':
         self._full_verification = False
         return self
 
-    def set_soft_voided_tx_ids(self, soft_voided_tx_ids: Set[bytes]) -> 'Builder':
+    def set_soft_voided_tx_ids(self, soft_voided_tx_ids: set[bytes]) -> 'Builder':
         self.check_if_can_modify()
         self._soft_voided_tx_ids = soft_voided_tx_ids
         return self

hathor/builder/resources_builder.py

@@ -14,7 +14,7 @@
 
 import os
 from argparse import Namespace
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 from autobahn.twisted.resource import WebSocketResource
 from structlog import get_logger
@@ -50,7 +50,7 @@ def build(self, args: Namespace) -> Optional[server.Site]:
         return None
 
     def create_prometheus(self, args: Namespace) -> PrometheusMetricsExporter:
-        kwargs: Dict[str, Any] = {
+        kwargs: dict[str, Any] = {
             'metrics': self.manager.metrics,
             'metrics_prefix': args.prometheus_prefix
         }

hathor/cli/main.py

@@ -16,7 +16,7 @@
 import sys
 from collections import defaultdict
 from types import ModuleType
-from typing import Dict, List, Optional
+from typing import Optional
 
 from structlog import get_logger
 
@@ -26,9 +26,9 @@
 class CliManager:
     def __init__(self) -> None:
         self.basename: str = os.path.basename(sys.argv[0])
-        self.command_list: Dict[str, ModuleType] = {}
-        self.cmd_description: Dict[str, str] = {}
-        self.groups: Dict[str, List[str]] = defaultdict(list)
+        self.command_list: dict[str, ModuleType] = {}
+        self.cmd_description: dict[str, str] = {}
+        self.groups: dict[str, list[str]] = defaultdict(list)
         self.longest_cmd: int = 0
 
         from . import (

hathor/cli/mining.py

@@ -21,7 +21,6 @@
 from argparse import ArgumentParser, Namespace
 from json.decoder import JSONDecodeError
 from multiprocessing import Process, Queue
-from typing import Tuple
 
 import requests
 
@@ -72,7 +71,7 @@ def execute(args: Namespace) -> None:
 
     total = 0
     conn_retries = 0
-    q_in: Queue[Tuple[Block, int, int, int]]
+    q_in: Queue[tuple[Block, int, int, int]]
     q_out: Queue[Block]
     q_in, q_out = Queue(), Queue()
     while True:

hathor/cli/nginx_config.py

@@ -15,14 +15,14 @@
 import json
 import os
 from enum import Enum
-from typing import Any, Dict, List, NamedTuple, Optional, TextIO, Tuple
+from typing import Any, NamedTuple, Optional, TextIO
 
 from hathor.cli.openapi_json import get_openapi_dict
 
 BASE_PATH = os.path.join(os.path.dirname(__file__), 'nginx_files')
 
 
-def get_openapi(src_file: Optional[TextIO] = None) -> Dict[str, Any]:
+def get_openapi(src_file: Optional[TextIO] = None) -> dict[str, Any]:
     """ Open and parse the json file or generate OpenAPI dict on-the-fly
     """
     if src_file is None:
@@ -101,14 +101,14 @@ def _scale_rate_limit(raw_rate: str, rate_k: float) -> str:
     return f'{int(scaled_rate_amount)}{rate_units}'
 
 
-def _get_visibility(source: Dict[str, Any], fallback: Visibility) -> Tuple[Visibility, bool]:
+def _get_visibility(source: dict[str, Any], fallback: Visibility) -> tuple[Visibility, bool]:
     if 'x-visibility' in source:
         return Visibility(source['x-visibility']), False
     else:
         return fallback, True
 
 
-def generate_nginx_config(openapi: Dict[str, Any], *, out_file: TextIO, rate_k: float = 1.0,
+def generate_nginx_config(openapi: dict[str, Any], *, out_file: TextIO, rate_k: float = 1.0,
                           fallback_visibility: Visibility = Visibility.PRIVATE,
                           disable_rate_limits: bool = False) -> None:
     """ Entry point of the functionality provided by the cli
@@ -120,16 +120,16 @@ def generate_nginx_config(openapi: Dict[str, Any], *, out_file: TextIO, rate_k:
     settings = HathorSettings()
     api_prefix = settings.API_VERSION_PREFIX
 
-    locations: Dict[str, Dict[str, Any]] = {}
-    limit_rate_zones: List[RateLimitZone] = []
+    locations: dict[str, dict[str, Any]] = {}
+    limit_rate_zones: list[RateLimitZone] = []
     for path, params in openapi['paths'].items():
         visibility, did_fallback = _get_visibility(params, fallback_visibility)
         if did_fallback:
             warn(f'Visibility not set for path `{path}`, falling back to {fallback_visibility}')
         if visibility is Visibility.PRIVATE:
             continue
 
-        location_params: Dict[str, Any] = {
+        location_params: dict[str, Any] = {
             'rate_limits': [],
             'path_vars_re': params.get('x-path-params-regex', {}),
         }

hathor/cli/openapi_files/register.py

@@ -12,15 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import List, Type, TypeVar
+from typing import TypeVar
 
 from hathor.api_util import Resource
 
-_registered_resources: List[Type[Resource]] = []
+_registered_resources: list[type[Resource]] = []
 
 
 # XXX: this type var is used to indicate that the returned class is the same as the input class
-ResourceClass = TypeVar('ResourceClass', bound=Type[Resource])
+ResourceClass = TypeVar('ResourceClass', bound=type[Resource])
 
 
 def register_resource(resource_class: ResourceClass) -> ResourceClass:
@@ -31,7 +31,7 @@ def register_resource(resource_class: ResourceClass) -> ResourceClass:
     return resource_class
 
 
-def get_registered_resources() -> List[Type[Resource]]:
+def get_registered_resources() -> list[type[Resource]]:
     """ Returns a list with all the resources registered for the docs
     """
     import hathor.p2p.resources  # noqa: 401

hathor/cli/openapi_json.py

@@ -14,27 +14,27 @@
 
 import json
 import os
-from typing import Any, Dict
+from typing import Any
 
 BASE_PATH = os.path.join(os.path.dirname(__file__), 'openapi_files')
 DEFAULT_OUTPUT_PATH = os.path.join(BASE_PATH, 'openapi.json')
 
 
-def get_base() -> Dict[str, Any]:
+def get_base() -> dict[str, Any]:
     """ Returns the base configuration from OpenAPI json
     """
     with open(os.path.join(BASE_PATH, 'openapi_base.json'), 'r') as f:
         return json.load(f)
 
 
-def get_components() -> Dict[str, Any]:
+def get_components() -> dict[str, Any]:
     """ Returns the components from OpenAPI json
     """
     with open(os.path.join(BASE_PATH, 'openapi_components.json'), 'r') as f:
         return json.load(f)
 
 
-def get_openapi_dict() -> Dict[str, Any]:
+def get_openapi_dict() -> dict[str, Any]:
     """ Returns the generated OpenAPI dict
     """
     from hathor.cli.openapi_files.register import get_registered_resources

hathor/cli/run_node.py

@@ -15,7 +15,7 @@
 import os
 import sys
 from argparse import SUPPRESS, ArgumentParser, Namespace
-from typing import Any, Callable, List, Tuple
+from typing import Any, Callable
 
 from pydantic import ValidationError
 from structlog import get_logger
@@ -28,7 +28,7 @@
 
 
 class RunNode:
-    UNSAFE_ARGUMENTS: List[Tuple[str, Callable[[Namespace], bool]]] = [
+    UNSAFE_ARGUMENTS: list[tuple[str, Callable[[Namespace], bool]]] = [
         ('--test-mode-tx-weight', lambda args: bool(args.test_mode_tx_weight)),
         ('--enable-crash-api', lambda args: bool(args.enable_crash_api)),
         ('--x-sync-bridge', lambda args: bool(args.x_sync_bridge)),
@@ -361,7 +361,7 @@ def init_sysctl(self, description: str) -> None:
         endpoint = serverFromString(self.reactor, description)
         endpoint.listen(factory)
 
-    def parse_args(self, argv: List[str]) -> Namespace:
+    def parse_args(self, argv: list[str]) -> Namespace:
         return self.parser.parse_args(argv)
 
     def run(self) -> None:

hathor/cli/shell.py

@@ -13,12 +13,12 @@
 # limitations under the License.
 
 from argparse import Namespace
-from typing import Any, Callable, Dict, List
+from typing import Any, Callable
 
 from hathor.cli.run_node import RunNode
 
 
-def get_ipython(extra_args: List[Any], imported_objects: Dict[str, Any]) -> Callable[[], None]:
+def get_ipython(extra_args: list[Any], imported_objects: dict[str, Any]) -> Callable[[], None]:
     from IPython import start_ipython
 
     def run_ipython():
@@ -37,7 +37,7 @@ def register_signal_handlers(self, args: Namespace) -> None:
     def prepare(self, args: Namespace, *, register_resources: bool = True) -> None:
         super().prepare(args, register_resources=False)
 
-        imported_objects: Dict[str, Any] = {}
+        imported_objects: dict[str, Any] = {}
         imported_objects['tx_storage'] = self.tx_storage
         if args.wallet:
             imported_objects['wallet'] = self.wallet
@@ -51,9 +51,9 @@ def prepare(self, args: Namespace, *, register_resources: bool = True) -> None:
         print('------------------------')
         print()
 
-    def parse_args(self, argv: List[str]) -> Namespace:
+    def parse_args(self, argv: list[str]) -> Namespace:
         # TODO: add help for the `--` extra argument separator
-        extra_args: List[str] = []
+        extra_args: list[str] = []
         if '--' in argv:
             idx = argv.index('--')
             extra_args = argv[idx + 1:]