Annotate chia/rpc/util.py.

chia/rpc/util.py

@@ -4,11 +4,13 @@
 import logging
 import traceback
 from collections.abc import Awaitable
-from typing import TYPE_CHECKING, Any, Callable, Optional, get_type_hints
+from typing import Any, Callable, Optional, get_type_hints
 
 import aiohttp
 from chia_rs import AugSchemeMPL
 
+from chia.rpc.rpc_server import EndpointResult
+from chia.rpc.wallet_rpc_api import WalletRpcApi
 from chia.types.blockchain_format.coin import Coin
 from chia.types.coin_spend import CoinSpend
 from chia.util.json_util import obj_to_response
@@ -45,7 +47,9 @@ def marshal(func: MarshallableRpcEndpoint) -> RpcEndpoint:
     assert issubclass(request_hint, Streamable)
     request_class = request_hint
 
-    async def rpc_endpoint(self, request: dict[str, Any], *args: object, **kwargs: object) -> dict[str, Any]:
+    async def rpc_endpoint(
+        self: WalletRpcApi, request: dict[str, Any], *args: object, **kwargs: object
+    ) -> dict[str, Any]:
         response_obj: Streamable = await func(
             self,
             (
@@ -78,8 +82,10 @@ async def rpc_endpoint(self, request: dict[str, Any], *args: object, **kwargs: o
     return rpc_endpoint
 
 
-def wrap_http_handler(f) -> Callable:
-    async def inner(request) -> aiohttp.web.Response:
+def wrap_http_handler(
+    f: Callable[[dict[str, Any]], Awaitable[EndpointResult]]
+) -> Callable[[aiohttp.web.Request], Awaitable[aiohttp.web.StreamResponse]]:
+    async def inner(request: aiohttp.web.Request) -> aiohttp.web.StreamResponse:
         request_data = await request.json()
         try:
             res_object = await f(request_data)
@@ -105,11 +111,9 @@ def tx_endpoint(
     merge_spends: bool = True,
 ) -> Callable[[RpcEndpoint], RpcEndpoint]:
     def _inner(func: RpcEndpoint) -> RpcEndpoint:
-        async def rpc_endpoint(self, request: dict[str, Any], *args, **kwargs) -> dict[str, Any]:
-            if TYPE_CHECKING:
-                from chia.rpc.wallet_rpc_api import WalletRpcApi
-
-                assert isinstance(self, WalletRpcApi)
+        async def rpc_endpoint(
+            self: WalletRpcApi, request: dict[str, Any], *args: object, **kwargs: object
+        ) -> dict[str, Any]:
             assert self.service.logged_in_fingerprint is not None
             tx_config_loader: TXConfigLoader = TXConfigLoader.from_json_dict(request)
 

mypy-exclusions.txt

@@ -10,7 +10,6 @@ chia.plotters.plotters_util
 chia.plotting.manager
 chia.plotting.util
 chia.rpc.rpc_client
-chia.rpc.util
 chia.simulator.full_node_simulator
 chia.simulator.keyring
 chia.simulator.wallet_tools