Annotate chia/rpc/util.py.

chia/rpc/util.py

@@ -34,6 +34,12 @@
 # This definition is weaker than that one however because the arguments can be anything
 RpcEndpoint = Callable[..., Awaitable[dict[str, Any]]]
 MarshallableRpcEndpoint = Callable[..., Awaitable[Streamable]]
+if TYPE_CHECKING:
+    from chia.rpc.rpc_server import EndpointResult
+    from chia.rpc.wallet_rpc_api import WalletRpcApi
+else:
+    EndpointResult = dict[str, Any]
+    WalletRpcApi = object
 
 
 ALL_TRANSLATION_LAYERS: dict[str, TranslationLayer] = {"CHIP-0028": BLIND_SIGNER_TRANSLATION}
@@ -45,7 +51,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 +86,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 +115,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