Merge pull request #1542 from njgheorghita/type-hints-utils-complete

Finish type hints in web3._utils

Author: njgheorghita

setup.py

@@ -77,13 +77,12 @@
         "ipfshttpclient>=0.4.12,<1",
         "jsonschema>=3.0.0,<4.0.0",
         "lru-dict>=1.1.6,<2.0.0",
-        # remove mypy_extensions after python_requires>=3.8
-        # see web3._utils.compat
-        "mypy_extensions>=0.4.1,<1.0.0",
         "protobuf>=3.10.0,<4",
         "pypiwin32>=223;platform_system=='Windows'",
         "requests>=2.16.0,<3.0.0",
         "websockets>=8.1.0,<9.0.0",
+        # remove mypy_extensions & typing-extensions after python_requires>=3.8
+        # see web3._utils.compat
         "typing-extensions>=3.7.4.1,<4",
     ],
     setup_requires=['setuptools-markdown'],

tox.ini

@@ -63,4 +63,4 @@ extras=linter
 commands=
   flake8 {toxinidir}/web3 {toxinidir}/ens {toxinidir}/ethpm {toxinidir}/tests
   isort --recursive --check-only --diff {toxinidir}/web3/ {toxinidir}/ens/ {toxinidir}/ethpm/ {toxinidir}/tests/
-  mypy -p web3._utils.abi -p web3._utils.admin -p web3._utils.blocks -p web3._utils.caching -p web3._utils.contracts -p web3._utils.datatypes -p web3._utils.decorators -p web3._utils.empty -p web3._utils.ens -p web3._utils.events -p web3._utils.filters -p web3._utils.method_formatters -p web3._utils.module_testing -p web3._utils.request -p web3._utils.shh -p web3._utils.threads -p web3._utils.transactions -p web3._utils.txpool -p web3._utils.validation -p web3._utils.windows -p web3.providers -p web3.main -p web3.contract -p web3.datastructures -p web3.eth -p web3.exceptions -p web3.geth -p web3.iban -p web3.logs -p web3.manager -p web3.module -p web3.net -p web3.parity -p web3.middleware -p web3.method -p web3.pm -p web3.auto -p web3.gas_strategies -p web3.testing -p web3.tools -p web3.version -p ethpm -p ens --config-file {toxinidir}/mypy.ini
+  mypy -p web3._utils -p web3.providers -p web3.main -p web3.contract -p web3.datastructures -p web3.eth -p web3.exceptions -p web3.geth -p web3.iban -p web3.logs -p web3.manager -p web3.module -p web3.net -p web3.parity -p web3.middleware -p web3.method -p web3.pm -p web3.auto -p web3.gas_strategies -p web3.testing -p web3.tools -p web3.version -p ethpm -p ens --config-file {toxinidir}/mypy.ini

web3/_utils/admin.py

@@ -4,10 +4,9 @@
     Tuple,
 )
 
-from typing_extensions import (
+from web3._utils.compat import (
     Protocol,
 )
-
 from web3._utils.rpc_abi import (
     RPC,
 )

web3/_utils/compat/__init__.py

@@ -1,7 +1,6 @@
 # remove once web3 supports python>=3.8
 # TypedDict was added to typing in 3.8
 try:
-    from typing import Literal, TypedDict  # type: ignore
+    from typing import Literal, Protocol, TypedDict  # type: ignore
 except ImportError:
-    from mypy_extensions import TypedDict  # noqa: F401
-    from typing_extensions import Literal  # noqa: F401
+    from typing_extensions import Literal, Protocol, TypedDict  # type: ignore # noqa: F401

web3/_utils/encoding.py

@@ -1,10 +1,24 @@
 # String encodings and numeric representations
 import json
 import re
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    Sequence,
+    Type,
+    Union,
+)
 
 from eth_abi.encoding import (
     BaseArrayEncoder,
 )
+from eth_typing import (
+    HexStr,
+    Primitives,
+    TypeStr,
+)
 from eth_utils import (
     add_0x_prefix,
     big_endian_to_int,
@@ -47,7 +61,7 @@
 )
 
 
-def hex_encode_abi_type(abi_type, value, force_size=None):
+def hex_encode_abi_type(abi_type: TypeStr, value: Any, force_size: int=None) -> HexStr:
     """
     Encodes value into a hex string in format of abi_type
     """
@@ -57,7 +71,9 @@ def hex_encode_abi_type(abi_type, value, force_size=None):
     data_size = force_size or size_of_type(abi_type)
     if is_array_type(abi_type):
         sub_type = sub_type_of_array_type(abi_type)
-        return "".join([remove_0x_prefix(hex_encode_abi_type(sub_type, v, 256)) for v in value])
+        return HexStr(
+            "".join([remove_0x_prefix(hex_encode_abi_type(sub_type, v, 256)) for v in value])
+        )
     elif is_bool_type(abi_type):
         return to_hex_with_size(value, data_size)
     elif is_uint_type(abi_type):
@@ -79,7 +95,7 @@ def hex_encode_abi_type(abi_type, value, force_size=None):
         )
 
 
-def to_hex_twos_compliment(value, bit_size):
+def to_hex_twos_compliment(value: Any, bit_size: int) -> HexStr:
     """
     Converts integer value to twos compliment hex representation with given bit_size
     """
@@ -88,34 +104,34 @@ def to_hex_twos_compliment(value, bit_size):
 
     value = (1 << bit_size) + value
     hex_value = hex(value)
-    hex_value = hex_value.rstrip("L")
+    hex_value = HexStr(hex_value.rstrip("L"))
     return hex_value
 
 
-def to_hex_with_size(value, bit_size):
+def to_hex_with_size(value: Any, bit_size: int) -> HexStr:
     """
     Converts a value to hex with given bit_size:
     """
     return pad_hex(to_hex(value), bit_size)
 
 
-def pad_hex(value, bit_size):
+def pad_hex(value: Any, bit_size: int) -> HexStr:
     """
     Pads a hex string up to the given bit_size
     """
     value = remove_0x_prefix(value)
     return add_0x_prefix(value.zfill(int(bit_size / 4)))
 
 
-def trim_hex(hexstr):
+def trim_hex(hexstr: HexStr) -> HexStr:
     if hexstr.startswith('0x0'):
-        hexstr = re.sub('^0x0+', '0x', hexstr)
+        hexstr = HexStr(re.sub('^0x0+', '0x', hexstr))
         if hexstr == '0x':
-            hexstr = '0x0'
+            hexstr = HexStr('0x0')
     return hexstr
 
 
-def to_int(value=None, hexstr=None, text=None):
+def to_int(value: Primitives=None, hexstr: HexStr=None, text: str=None) -> int:
     """
     Converts value to it's integer representation.
 
@@ -142,14 +158,14 @@ def to_int(value=None, hexstr=None, text=None):
 
 
 @curry
-def pad_bytes(fill_with, num_bytes, unpadded):
+def pad_bytes(fill_with: bytes, num_bytes: int, unpadded: bytes) -> bytes:
     return unpadded.rjust(num_bytes, fill_with)
 
 
 zpad_bytes = pad_bytes(b'\0')
 
 
-def to_bytes(primitive=None, hexstr=None, text=None):
+def to_bytes(primitive: Primitives=None, hexstr: HexStr=None, text: str=None) -> bytes:
     assert_one_val(primitive, hexstr=hexstr, text=text)
 
     if is_boolean(primitive):
@@ -160,14 +176,14 @@ def to_bytes(primitive=None, hexstr=None, text=None):
         return to_bytes(hexstr=to_hex(primitive))
     elif hexstr is not None:
         if len(hexstr) % 2:
-            hexstr = '0x0' + remove_0x_prefix(hexstr)
+            hexstr = HexStr('0x0' + remove_0x_prefix(hexstr))
         return decode_hex(hexstr)
     elif text is not None:
         return text.encode('utf-8')
     raise TypeError("expected an int in first arg, or keyword of hexstr or text")
 
 
-def to_text(primitive=None, hexstr=None, text=None):
+def to_text(primitive: Primitives=None, hexstr: HexStr=None, text: str=None) -> str:
     assert_one_val(primitive, hexstr=hexstr, text=text)
 
     if hexstr is not None:
@@ -185,13 +201,15 @@ def to_text(primitive=None, hexstr=None, text=None):
 
 
 @curry
-def text_if_str(to_type, text_or_primitive):
+def text_if_str(
+    to_type: Callable[..., str], text_or_primitive: Union[Primitives, HexStr, str]
+) -> str:
     """
     Convert to a type, assuming that strings can be only unicode text (not a hexstr)
 
     @param to_type is a function that takes the arguments (primitive, hexstr=hexstr, text=text),
         eg~ to_bytes, to_text, to_hex, to_int, etc
-    @param hexstr_or_primitive in bytes, str, or int.
+    @param text_or_primitive in bytes, str, or int.
     """
     if isinstance(text_or_primitive, str):
         (primitive, text) = (None, text_or_primitive)
@@ -201,17 +219,19 @@ def text_if_str(to_type, text_or_primitive):
 
 
 @curry
-def hexstr_if_str(to_type, hexstr_or_primitive):
+def hexstr_if_str(
+    to_type: Callable[..., HexStr], hexstr_or_primitive: Union[Primitives, HexStr, str]
+) -> HexStr:
     """
     Convert to a type, assuming that strings can be only hexstr (not unicode text)
 
     @param to_type is a function that takes the arguments (primitive, hexstr=hexstr, text=text),
         eg~ to_bytes, to_text, to_hex, to_int, etc
-    @param text_or_primitive in bytes, str, or int.
+    @param hexstr_or_primitive in bytes, str, or int.
     """
     if isinstance(hexstr_or_primitive, str):
         (primitive, hexstr) = (None, hexstr_or_primitive)
-        if remove_0x_prefix(hexstr) and not is_hex(hexstr):
+        if remove_0x_prefix(HexStr(hexstr)) and not is_hex(hexstr):
             raise ValueError(
                 "when sending a str, it must be a hex string. Got: {0!r}".format(
                     hexstr_or_primitive,
@@ -230,21 +250,21 @@ class FriendlyJsonSerde:
     information on which fields failed, to show more
     helpful information in the raised error messages.
     """
-    def _json_mapping_errors(self, mapping):
+    def _json_mapping_errors(self, mapping: Dict[Any, Any]) -> Iterable[str]:
         for key, val in mapping.items():
             try:
                 self._friendly_json_encode(val)
             except TypeError as exc:
                 yield "%r: because (%s)" % (key, exc)
 
-    def _json_list_errors(self, iterable):
+    def _json_list_errors(self, iterable: Iterable[Any]) -> Iterable[str]:
         for index, element in enumerate(iterable):
             try:
                 self._friendly_json_encode(element)
             except TypeError as exc:
                 yield "%d: because (%s)" % (index, exc)
 
-    def _friendly_json_encode(self, obj, cls=None):
+    def _friendly_json_encode(self, obj: Dict[Any, Any], cls: Type[json.JSONEncoder]=None) -> str:
         try:
             encoded = json.dumps(obj, cls=cls)
             return encoded
@@ -258,7 +278,7 @@ def _friendly_json_encode(self, obj, cls=None):
             else:
                 raise full_exception
 
-    def json_decode(self, json_str):
+    def json_decode(self, json_str: str) -> Dict[Any, Any]:
         try:
             decoded = json.loads(json_str)
             return decoded
@@ -268,14 +288,14 @@ def json_decode(self, json_str):
             # so we have to re-raise the same type.
             raise json.decoder.JSONDecodeError(err_msg, exc.doc, exc.pos)
 
-    def json_encode(self, obj, cls=None):
+    def json_encode(self, obj: Dict[Any, Any], cls: Type[json.JSONEncoder]=None) -> str:
         try:
             return self._friendly_json_encode(obj, cls=cls)
         except TypeError as exc:
             raise TypeError("Could not encode to JSON: {}".format(exc))
 
 
-def to_4byte_hex(hex_or_str_or_bytes):
+def to_4byte_hex(hex_or_str_or_bytes: Union[HexStr, str, bytes, int]) -> HexStr:
     size_of_4bytes = 4 * 8
     byte_str = hexstr_if_str(to_bytes, hex_or_str_or_bytes)
     if len(byte_str) > 4:
@@ -289,15 +309,15 @@ def to_4byte_hex(hex_or_str_or_bytes):
 class DynamicArrayPackedEncoder(BaseArrayEncoder):
     is_dynamic = True
 
-    def encode(self, value):
+    def encode(self, value: Sequence[Any]) -> bytes:
         encoded_elements = self.encode_elements(value)
         encoded_value = encoded_elements
 
         return encoded_value
 
 
 #  TODO: Replace with eth-abi packed encoder once web3 requires eth-abi>=2
-def encode_single_packed(_type, value):
+def encode_single_packed(_type: TypeStr, value: Any) -> bytes:
     import codecs
     from eth_abi import (
         grammar as abi_type_parser,
@@ -315,18 +335,19 @@ def encode_single_packed(_type, value):
         return codecs.encode(value, 'utf8')
     elif abi_type.base == "bytes":
         return value
+    return None
 
 
 class Web3JsonEncoder(json.JSONEncoder):
-    def default(self, obj):
+    def default(self, obj: Any) -> Union[Dict[Any, Any], HexStr]:
         if isinstance(obj, AttributeDict):
             return {k: v for k, v in obj.items()}
         if isinstance(obj, HexBytes):
-            return obj.hex()
+            return HexStr(obj.hex())
         return json.JSONEncoder.default(self, obj)
 
 
-def to_json(obj):
+def to_json(obj: Dict[Any, Any]) -> str:
     '''
     Convert a complex object (like a transaction object) to a JSON string
     '''