Add type hints to web3.providers

Author: njgheorghita

web3/manager.py

@@ -74,7 +74,8 @@ def __init__(
         if middlewares is None:
             middlewares = self.default_middlewares(web3)
 
-        self.middleware_onion: NamedElementOnion[str, Middleware] = NamedElementOnion(middlewares)
+        # web3/manager.py:77: error: "NamedElementOnion" expects 2 type arguments, but 1 given
+        self.middleware_onion: NamedElementOnion[Sequence[Tuple[Middleware, str]]] = NamedElementOnion(middlewares)  # type: ignore # noqa: E501
 
         if provider is None:
             self.provider = AutoProvider()

web3/providers/auto.py

@@ -1,8 +1,22 @@
 import os
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Optional,
+    Sequence,
+    Tuple,
+    Type,
+    Union,
+)
 from urllib.parse import (
     urlparse,
 )
 
+from eth_typing import (
+    URI,
+)
+
 from web3.exceptions import (
     CannotHandleRequest,
 )
@@ -12,20 +26,26 @@
     IPCProvider,
     WebsocketProvider,
 )
+from web3.types import (
+    RPCEndpoint,
+    RPCResponse,
+)
 
 HTTP_SCHEMES = {'http', 'https'}
 WS_SCHEMES = {'ws', 'wss'}
 
 
-def load_provider_from_environment():
-    uri_string = os.environ.get('WEB3_PROVIDER_URI', '')
+def load_provider_from_environment() -> BaseProvider:
+    uri_string = URI(os.environ.get('WEB3_PROVIDER_URI', ''))
     if not uri_string:
         return None
 
     return load_provider_from_uri(uri_string)
 
 
-def load_provider_from_uri(uri_string, headers=None):
+def load_provider_from_uri(
+    uri_string: URI, headers: Dict[str, Tuple[str, str]]=None
+) -> BaseProvider:
     uri = urlparse(uri_string)
     if uri.scheme == 'file':
         return IPCProvider(uri.path)
@@ -52,7 +72,10 @@ class AutoProvider(BaseProvider):
     )
     _active_provider = None
 
-    def __init__(self, potential_providers=None):
+    def __init__(
+        self,
+        potential_providers: Sequence[Union[Callable[..., BaseProvider], Type[BaseProvider]]]=None
+    ) -> None:
         """
         :param iterable potential_providers: ordered series of provider classes to attempt with
 
@@ -65,17 +88,17 @@ def __init__(self, potential_providers=None):
         else:
             self._potential_providers = self.default_providers
 
-    def make_request(self, method, params):
+    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
         try:
             return self._proxy_request(method, params)
         except IOError as exc:
             return self._proxy_request(method, params, use_cache=False)
 
-    def isConnected(self):
+    def isConnected(self) -> bool:
         provider = self._get_active_provider(use_cache=True)
         return provider is not None and provider.isConnected()
 
-    def _proxy_request(self, method, params, use_cache=True):
+    def _proxy_request(self, method: RPCEndpoint, params: Any, use_cache: bool=True) -> RPCResponse:
         provider = self._get_active_provider(use_cache)
         if provider is None:
             raise CannotHandleRequest(
@@ -87,7 +110,7 @@ def _proxy_request(self, method, params, use_cache=True):
 
         return provider.make_request(method, params)
 
-    def _get_active_provider(self, use_cache):
+    def _get_active_provider(self, use_cache: bool) -> Optional[BaseProvider]:
         if use_cache and self._active_provider is not None:
             return self._active_provider
 

web3/providers/base.py

@@ -1,4 +1,11 @@
 import itertools
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Sequence,
+    Tuple,
+)
 
 from eth_utils import (
     to_bytes,
@@ -8,24 +15,41 @@
 from web3._utils.encoding import (
     FriendlyJsonSerde,
 )
+from web3.datastructures import (
+    NamedElementOnion,
+)
 from web3.middleware import (
     combine_middlewares,
 )
+from web3.types import (
+    Middleware,
+    RPCEndpoint,
+    RPCResponse,
+)
+
+if TYPE_CHECKING:
+    from web3 import Web3  # noqa: F401
 
 
 class BaseProvider:
-    _middlewares = ()
-    _request_func_cache = (None, None)  # a tuple of (all_middlewares, request_func)
+    _middlewares: Tuple[Middleware, ...] = ()
+    # a tuple of (all_middlewares, request_func)
+    _request_func_cache: Tuple[Tuple[Middleware, ...], Callable[..., RPCResponse]] = (None, None)
 
     @property
-    def middlewares(self):
+    def middlewares(self) -> Tuple[Middleware, ...]:
         return self._middlewares
 
+    # web3/manager.py:77: error: "NamedElementOnion" expects 2 type arguments, but 1 given
     @middlewares.setter
-    def middlewares(self, values):
+    def middlewares(
+        self, values: NamedElementOnion[Sequence[Tuple[Middleware, str]]]  # type: ignore
+    ) -> None:
         self._middlewares = tuple(values)
 
-    def request_func(self, web3, outer_middlewares):
+    def request_func(
+        self, web3: "Web3", outer_middlewares: Sequence[Middleware]
+    ) -> Callable[..., RPCResponse]:
         """
         @param outer_middlewares is an iterable of middlewares, ordered by first to execute
         @returns a function that calls all the middleware and eventually self.make_request()
@@ -40,29 +64,31 @@ def request_func(self, web3, outer_middlewares):
             )
         return self._request_func_cache[-1]
 
-    def _generate_request_func(self, web3, middlewares):
+    def _generate_request_func(
+        self, web3: "Web3", middlewares: Sequence[Middleware]
+    ) -> Callable[..., RPCResponse]:
         return combine_middlewares(
             middlewares=middlewares,
             web3=web3,
             provider_request_fn=self.make_request,
         )
 
-    def make_request(self, method, params):
+    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
         raise NotImplementedError("Providers must implement this method")
 
-    def isConnected(self):
+    def isConnected(self) -> bool:
         raise NotImplementedError("Providers must implement this method")
 
 
 class JSONBaseProvider(BaseProvider):
-    def __init__(self):
+    def __init__(self) -> None:
         self.request_counter = itertools.count()
 
-    def decode_rpc_response(self, response):
-        text_response = to_text(response)
+    def decode_rpc_response(self, raw_response: bytes) -> RPCResponse:
+        text_response = to_text(raw_response)
         return FriendlyJsonSerde().json_decode(text_response)
 
-    def encode_rpc_request(self, method, params):
+    def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
         rpc_dict = {
             "jsonrpc": "2.0",
             "method": method,
@@ -72,9 +98,9 @@ def encode_rpc_request(self, method, params):
         encoded = FriendlyJsonSerde().json_encode(rpc_dict)
         return to_bytes(text=encoded)
 
-    def isConnected(self):
+    def isConnected(self) -> bool:
         try:
-            response = self.make_request('web3_clientVersion', [])
+            response = self.make_request(RPCEndpoint('web3_clientVersion'), [])
         except IOError:
             return False
 

web3/providers/eth_tester/defaults.py

@@ -57,7 +57,6 @@ def not_implemented(*args: Any, **kwargs: Any) -> NoReturn:
     raise NotImplementedError("RPC method not implemented")
 
 
-# double check RPCResponse
 @curry
 def call_eth_tester(
     fn_name: str, eth_tester: "EthereumTester", fn_args: Any, fn_kwargs: Any=None

web3/providers/eth_tester/main.py

@@ -44,10 +44,10 @@ async def make_request(
 
 
 class EthereumTesterProvider(BaseProvider):
-    middlewares = [
+    middlewares = (
         default_transaction_fields_middleware,
         ethereum_tester_middleware,
-    ]
+    )
     ethereum_tester = None
     api_endpoints = None