chore: update typing deps

hathor/api_util.py

@@ -152,7 +152,7 @@ def get_arg_default(args: dict[bytes, list[bytes]], key: str, default: T) -> T:
     bkey = key.encode()
     values = args.get(bkey)
     if not values:
-        return cast(T, default)
+        return default
     value: bytes = values[0]
     if isinstance(default, int):
         return cast(T, int(value))

hathor/dag_builder/vertex_exporter.py

@@ -289,6 +289,7 @@ def create_vertex_block(self, node: DAGNode) -> Block:
 
     def _get_ast_value_bytes(self, ast_node: ast.AST) -> bytes:
         if isinstance(ast_node, ast.Constant):
+            assert isinstance(ast_node.value, str)
             return bytes.fromhex(ast_node.value)
         elif isinstance(ast_node, ast.Name):
             return self.get_vertex_id(ast_node.id)

hathor/indexes/rocksdb_timestamp_index.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import TYPE_CHECKING, Iterator, Optional
+from typing import TYPE_CHECKING, Any, Iterator, Optional
 
 from structlog import get_logger
 
@@ -93,7 +93,7 @@ def _iter(self, from_timestamp: Optional[int] = None, from_tx: Optional[bytes] =
         """
         if from_timestamp is None and from_tx is not None:
             raise ValueError('from_tx needs from_timestamp, but it is None')
-        it = self._db.iterkeys(self._cf)
+        it: Any = self._db.iterkeys(self._cf)
         if reverse:
             it = reversed(it)
             if from_timestamp is None:

hathor/indexes/rocksdb_tokens_index.py

@@ -14,7 +14,7 @@
 
 from dataclasses import asdict, dataclass
 from enum import Enum
-from typing import TYPE_CHECKING, Iterator, NamedTuple, Optional, cast
+from typing import TYPE_CHECKING, Any, Iterator, NamedTuple, Optional, cast
 
 from structlog import get_logger
 from typing_extensions import assert_never, override
@@ -542,7 +542,7 @@ def _iter_transactions(self, token_uid: bytes, from_tx: Optional[_TxIndex] = Non
                            *, reverse: bool = False) -> Iterator[bytes]:
         """ Iterate over all transactions of a token, by default from oldest to newest.
         """
-        it = self._db.iterkeys(self._cf)
+        it: Any = self._db.iterkeys(self._cf)
         seek_key = self._to_key_txs(token_uid, from_tx)
         self.log.debug('seek to', token_uid=token_uid.hex(), key=seek_key.hex())
         if reverse:

hathor/indexes/rocksdb_tx_group_index.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from abc import abstractmethod
-from typing import Callable, Iterator, Optional, Sized, TypeVar
+from typing import Any, Callable, Generic, Iterator, Optional, Sized, TypeVar
 
 import rocksdb
 from structlog import get_logger
@@ -31,7 +31,7 @@
 GROUP_COUNT_VALUE_SIZE = 4  # in bytes
 
 
-class _RocksDBTxGroupStatsIndex(RocksDBIndexUtils):
+class _RocksDBTxGroupStatsIndex(RocksDBIndexUtils, Generic[KT]):
     def __init__(
         self,
         db: rocksdb.DB,
@@ -157,7 +157,7 @@ def _get_sorted_from_key(
         reverse: bool = False
     ) -> Iterator[bytes]:
         self.log.debug('seek to', key=key)
-        it = self._db.iterkeys(self._cf)
+        it: Any = self._db.iterkeys(self._cf)
         if reverse:
             it = reversed(it)
             # when reversed we increment the key by 1, which effectively goes to the end of a prefix
@@ -191,7 +191,7 @@ def _is_key_empty(self, key: KT) -> bool:
 
     @override
     def get_latest_tx_timestamp(self, key: KT) -> int | None:
-        it = self._db.iterkeys(self._cf)
+        it: Any = self._db.iterkeys(self._cf)
         it = reversed(it)
         # when reversed we increment the key by 1, which effectively goes to the end of a prefix
         it.seek_for_prev(incr_key(self._to_rocksdb_key(key)))

hathor/indexes/rocksdb_vertex_timestamp_index.py

@@ -14,7 +14,7 @@
 
 import struct
 from abc import ABC
-from typing import Iterator, final
+from typing import Any, Iterator, final
 
 import rocksdb
 from structlog import get_logger
@@ -99,7 +99,7 @@ def _iter_sorted(
         reverse: bool,
         inclusive: bool = False,
     ) -> Iterator[bytes]:
-        it = self._db.iterkeys(self._cf)
+        it: Any = self._db.iterkeys(self._cf)
         if reverse:
             it = reversed(it)
             if tx_start is None:

hathor/nanocontracts/custom_builtins.py

@@ -228,25 +228,26 @@ def __import__(
         name: str,
         globals: Mapping[str, object] | None = None,
         locals: Mapping[str, object] | None = None,
-        fromlist: Sequence[str] = (),
+        fromlist: Sequence[str] | None = None,
         level: int = 0,
     ) -> types.ModuleType:
+        fromlist_: Sequence[str] = fromlist or ()
         if level != 0:
             raise ImportError('Relative imports are not allowed')
-        if not fromlist and name != 'typing':
+        if not fromlist_ and name != 'typing':
             # XXX: typing is allowed here because Foo[T] triggers a __import__('typing', fromlist=None) for some reason
             raise ImportError('Only `from ... import ...` imports are allowed')
         if name not in allowed_imports:
             raise ImportError(f'Import from "{name}" is not allowed.')
 
         # Create a fake module class that will only be returned by this import call
         class FakeModule:
-            __slots__ = tuple(fromlist)
+            __slots__ = tuple(fromlist_)
 
         fake_module = FakeModule()
         allowed_fromlist = allowed_imports[name]
 
-        for import_what in fromlist:
+        for import_what in fromlist_:
             if import_what not in allowed_fromlist:
                 raise ImportError(f'Import from "{name}.{import_what}" is not allowed.')
 

hathor/nanocontracts/faux_immutable.py

@@ -114,7 +114,7 @@ def create_with_shell(cls: Callable[P, T], *args: P.args, **kwargs: P.kwargs) ->
     shell_type: type[T] = type(name, bases, attrs)
 
     # Use it to instantiate the object, init it, and return it. This mimics the default `__call__` behavior.
-    obj: T = cls.__new__(shell_type)
+    obj: T = cls.__new__(shell_type)  # type: ignore[call-overload]
     shell_type.__init__(obj, *args, **kwargs)
     return obj
 

hathor/nanocontracts/fields/dict_container.py

@@ -203,8 +203,7 @@ def get(self, key: K, /) -> V:
     def get(self, key: K, default: V | _T | None, /) -> V | _T | None:
         ...
 
-    # XXX: `misc` is ignored because mypy thinks this function does not accept all arguments of the second get overload
-    def get(self, key: K, default: V | _T | None = None, /) -> V | _T | None:  # type: ignore[misc]
+    def get(self, key: K, default: V | _T | None = None, /) -> V | _T | None:
         # return the value for key if key is in the storage, else default
         if key in self:
             return self[key]

hathor/nanocontracts/metered_exec.py

@@ -14,7 +14,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Callable, ParamSpec, TypeVar, cast
+from typing import Any, Callable, TypeVar, TypeVarTuple, Unpack, cast
 
 from structlog import get_logger
 
@@ -23,7 +23,7 @@
 logger = get_logger()
 
 _T = TypeVar('_T')
-_P = ParamSpec('_P')
+_Ts = TypeVarTuple('_Ts')
 
 
 # https://docs.python.org/3/library/sys.html#sys.settrace
@@ -77,7 +77,7 @@ def exec(self, source: str, /) -> dict[str, Any]:
         del env['__builtins__']
         return env
 
-    def call(self, func: Callable[_P, _T], /, *, args: _P.args) -> _T:
+    def call(self, func: Callable[[Unpack[_Ts]], _T], /, *, args: tuple[Unpack[_Ts]]) -> _T:
         """ This is equivalent to `func(*args, **kwargs)` but with execution metering and memory limiting.
         """
         from hathor import NCFail

hathor/nanocontracts/nc_types/dataclass_nc_type.py

@@ -78,9 +78,8 @@ def _from_type(cls, type_: type[D], /, *, type_map: NCType.TypeMap) -> Self:
         # XXX: the order is important, but `dict` and `fields` should have a stable order
         values: dict[str, NCType] = {}
         for field in fields(type_):
-            values[field.name] = NCType.from_type(field.type, type_map=type_map)
-        # XXX: ignore arg-type because after using is_dataclass(type_) mypy gets confused about type_'s type
-        return cls(values, type_)  # type: ignore[arg-type]
+            values[field.name] = NCType.from_type(field.type, type_map=type_map)  # type: ignore[arg-type]
+        return cls(values, type_)
 
     @override
     def _check_value(self, value: D, /, *, deep: bool) -> None:

hathor/nanocontracts/nc_types/utils.py

@@ -191,7 +191,7 @@ def _get_aliased_type(type_: type | UnionType, alias_map: TypeAliasMap) -> tuple
             final_type = reduce(or_, aliased_args)  # = type_args[0] | type_args[1] | ... | type_args[N]
             # XXX: for some reason, only sometimes doing T | None, results in typing.Union instead of types.UnionType
             assert isinstance(final_type, (UnionType, _UnionGenericAlias)), '| of types results in union'
-            return final_type, replaced
+            return final_type, replaced  # type: ignore[return-value]
 
         # XXX: special case, when going from list -> tuple, we need to add an ellipsis, that is to say, the equivalent
         #      type for `list[T]` is `tuple[T, ...]`

hathor/nanocontracts/types.py

@@ -45,7 +45,7 @@
 
 
 # Types to be used by blueprints.
-class Address(bytes, metaclass=FauxImmutableMeta):  # type: ignore[misc]
+class Address(bytes, metaclass=FauxImmutableMeta):
     __allow_faux_inheritance__ = True
     __allow_faux_dunder__ = ('__str__', '__repr__')
     __slots__ = ()
@@ -66,27 +66,27 @@ def __repr__(self) -> str:
         return f"Address.from_str({encoded_address!r})"
 
 
-class VertexId(bytes, metaclass=FauxImmutableMeta):  # type: ignore[misc]
+class VertexId(bytes, metaclass=FauxImmutableMeta):
     __slots__ = ()
     __allow_faux_inheritance__ = True
 
 
-class BlueprintId(VertexId):  # type: ignore[misc]
+class BlueprintId(VertexId):
     __slots__ = ()
     __allow_faux_inheritance__ = True
 
 
-class ContractId(VertexId):  # type: ignore[misc]
+class ContractId(VertexId):
     __slots__ = ()
     __allow_faux_inheritance__ = True
 
 
-class TokenUid(bytes, metaclass=FauxImmutableMeta):  # type: ignore[misc]
+class TokenUid(bytes, metaclass=FauxImmutableMeta):
     __slots__ = ()
     __allow_faux_inheritance__ = True
 
 
-class TxOutputScript(bytes, metaclass=FauxImmutableMeta):  # type: ignore[misc]
+class TxOutputScript(bytes, metaclass=FauxImmutableMeta):
     __slots__ = ()
     __allow_faux_inheritance__ = True
 

hathor/wallet/resources/thin_wallet/send_tokens.py

@@ -155,10 +155,9 @@ def _render_POST_stratum(self, context: _Context) -> None:
 
         # When using stratum to solve pow, we already set timestamp and parents
         stratum_deferred: Deferred[None] = Deferred()
-        # FIXME: Skipping mypy on the line below for now, as it looks like it's wrong but we don't have tests for it.
-        stratum_deferred.addCallback(self._stratum_deferred_resolve, request)  # type: ignore
-
-        fn_timeout = partial(self._stratum_timeout, request=request, tx=tx)
+        # FIXME: Skipping mypy on the lines below for now, as it looks like it's wrong but we don't have tests for it.
+        stratum_deferred.addCallback(self._stratum_deferred_resolve, request)  # type: ignore[call-overload]
+        fn_timeout = partial(self._stratum_timeout, request=request, tx=tx)  # type: ignore[call-arg]
         stratum_deferred.addTimeout(TIMEOUT_STRATUM_RESOLVE_POW, self.manager.reactor, onTimeoutCancel=fn_timeout)
 
         # this prepares transaction for mining

hathor_tests/event/websocket/test_factory.py

@@ -74,7 +74,8 @@ def test_broadcast_event(can_receive_event: bool) -> None:
     factory.broadcast_event(event)
 
     if not can_receive_event:
-        return connection.send_event_response.assert_not_called()
+        connection.send_event_response.assert_not_called()
+        return
 
     response = EventResponse(
         peer_id='my_peer_id',
@@ -136,7 +137,8 @@ def test_send_next_event_to_connection(next_expected_event_id: int, can_receive_
     clock.advance(0)
 
     if not can_receive_event or next_expected_event_id > n_starting_events - 1:
-        return connection.send_event_response.assert_not_called()
+        connection.send_event_response.assert_not_called()
+        return
 
     calls = []
     for _id in range(next_expected_event_id, n_starting_events):

hathor_tests/nanocontracts/blueprints/unittest.py

@@ -1,4 +1,4 @@
-from io import TextIOWrapper
+from io import StringIO, TextIOWrapper
 from typing import Sequence
 
 from hathor.crypto.util import decode_address
@@ -90,7 +90,7 @@ def register_blueprint_file(self, path: str, blueprint_id: BlueprintId | None =
 
     def _register_blueprint_contents(
         self,
-        contents: TextIOWrapper,
+        contents: TextIOWrapper | StringIO,
         blueprint_id: BlueprintId | None = None,
         *,
         skip_verification: bool = False,

hathor_tests/nanocontracts/test_blueprint_syntax.py

@@ -113,7 +113,7 @@ def test_public_missing_self(self) -> None:
         with pytest.raises(BlueprintSyntaxError, match=re.escape(msg)):
             class MyBlueprint(Blueprint):
                 @public
-                def initialize() -> None:  # type: ignore
+                def initialize() -> None:
                     pass
 
     def test_public_wrong_self(self) -> None:
@@ -129,7 +129,7 @@ def test_public_typed_self(self) -> None:
         with pytest.raises(BlueprintSyntaxError, match=re.escape(msg)):
             class MyBlueprint(Blueprint):
                 @public
-                def initialize(self: int) -> None:  # type: ignore
+                def initialize(self: int) -> None:
                     pass
 
     def test_view_missing_self(self) -> None:
@@ -141,7 +141,7 @@ def initialize(self, ctx: Context) -> None:
                     pass
 
                 @view
-                def nop() -> None:  # type: ignore
+                def nop() -> None:
                     pass
 
     def test_view_wrong_self(self) -> None:
@@ -165,7 +165,7 @@ def initialize(self, ctx: Context) -> None:
                     pass
 
                 @view
-                def nop(self: int) -> None:  # type: ignore
+                def nop(self: int) -> None:
                     pass
 
     def test_fallback_missing_self(self) -> None:
@@ -177,7 +177,7 @@ def initialize(self, ctx: Context) -> None:
                     pass
 
                 @fallback
-                def fallback() -> None:  # type: ignore
+                def fallback() -> None:
                     pass
 
     def test_fallback_wrong_self(self) -> None:
@@ -201,7 +201,7 @@ def initialize(self, ctx: Context) -> None:
                     pass
 
                 @fallback
-                def fallback(self: int) -> None:  # type: ignore
+                def fallback(self: int) -> None:
                     pass
 
     def test_public_missing_context(self) -> None:

hathor_tests/nanocontracts/test_storage.py

@@ -89,7 +89,7 @@ def test_bool_false(self) -> None:
 
     def test_tuple(self) -> None:
         value: NCType[tuple[str, int, set[int], bool]]
-        value = make_nc_type(tuple[str, int, set[int], bool])  # type: ignore[arg-type]
+        value = make_nc_type(tuple[str, int, set[int], bool])
         self._run_test(('str', 1, {3}, True), value)
 
     def test_changes_tracker_delete(self) -> None: