feat(mypy): fix async web3 method typing

Author: BobTheBuidler

faster_web3/eth/async_eth.py

@@ -612,7 +612,6 @@ async def modify_transaction(
             current_transaction
         )
         new_transaction = merge(current_transaction_params, transaction_params)
-
         return await async_replace_transaction(w3, current_transaction, new_transaction)
 
     # eth_sign
@@ -730,7 +729,8 @@ async def subscribe(
         label: Optional[str] = None,
         parallelize: Optional[bool] = None,
     ) -> HexStr:
-        if not isinstance(self.w3.provider, PersistentConnectionProvider):
+        w3 = self.w3
+        if not isinstance(w3.provider, PersistentConnectionProvider):
             raise MethodNotSupported(
                 "eth_subscribe is only supported with providers that support "
                 "persistent connections."
@@ -743,21 +743,22 @@ async def subscribe(
             label=label,
             parallelize=parallelize,
         )
-        return await self.w3.subscription_manager.subscribe(sub)
+        return await w3.subscription_manager.subscribe(sub)
 
     _unsubscribe: Method[Callable[[HexStr], Awaitable[bool]]] = Method(
         RPC.eth_unsubscribe,
         mungers=[default_root_munger],
     )
 
     async def unsubscribe(self, subscription_id: HexStr) -> bool:
-        if not isinstance(self.w3.provider, PersistentConnectionProvider):
+        w3 = self.w3
+        if not isinstance(w3.provider, PersistentConnectionProvider):
             raise MethodNotSupported(
                 "eth_unsubscribe is only supported with providers that support "
                 "persistent connections."
             )
 
-        for sub in self.w3.subscription_manager.subscriptions:
+        for sub in w3.subscription_manager.subscriptions:
             if sub._id == subscription_id:
                 return await sub.unsubscribe()
 

faster_web3/providers/persistent/subscription_container.py

@@ -54,6 +54,4 @@ def get_handler_subscription_by_id(
         self, sub_id: HexStr
     ) -> Optional[EthSubscription[Any]]:
         sub = self.get_by_id(sub_id)
-        if sub and sub._handler:
-            return sub
-        return None
+        return sub if sub and sub._handler else None

faster_web3/providers/persistent/utils.py

@@ -1,44 +1,65 @@
 import functools
 from typing import (
     TYPE_CHECKING,
-    Any,
     Callable,
     Optional,
 )
 
+from typing_extensions import (
+    Concatenate,
+    ParamSpec,
+)
+
 from faster_web3.exceptions import (
     Web3ValidationError,
 )
 from faster_web3.providers import (
     PersistentConnectionProvider,
 )
+from faster_web3.types import (
+    TReturn,
+)
 
 if TYPE_CHECKING:
     from faster_web3.main import (  # noqa: F401
+        AsyncProviderT,
         AsyncWeb3,
     )
 
 
+P = ParamSpec("P")
+
+AsyncWeb3Method = Callable[Concatenate["AsyncWeb3[AsyncProviderT]", P], TReturn]
+
+
 def persistent_connection_provider_method(
     message: Optional[str] = None,
-) -> Callable[..., Any]:
+) -> Callable[
+    [AsyncWeb3Method[AsyncProviderT, P, TReturn]],
+    AsyncWeb3Method[AsyncProviderT, P, TReturn],
+]:
     """
     Decorator that raises an exception if the provider is not an instance of
     ``PersistentConnectionProvider``.
     """
 
-    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
-        @functools.wraps(func)
-        def inner(self: "AsyncWeb3[Any]", *args: Any, **kwargs: Any) -> Any:
-            nonlocal message
-            if message is None:
-                message = (
-                    f"``{func.__name__}`` can only be called on a "
-                    "``PersistentConnectionProvider`` instance."
-                )
+    def decorator(
+        func: AsyncWeb3Method[AsyncProviderT, P, TReturn],
+    ) -> AsyncWeb3Method[AsyncProviderT, P, TReturn]:
+        if message is None:
+            message_actual = (
+                f"``{func.__name__}`` can only be called on a "
+                "``PersistentConnectionProvider`` instance."
+            )
+        else:
+            message_actual = message
 
+        @functools.wraps(func)
+        def inner(
+            self: "AsyncWeb3[AsyncProviderT]", *args: P.args, **kwargs: P.kwargs
+        ) -> TReturn:
             if not isinstance(self.provider, PersistentConnectionProvider):
-                raise Web3ValidationError(message)
+                raise Web3ValidationError(message_actual)
             return func(self, *args, **kwargs)
 
         return inner