explicitly check RpcApiProtocol classes (#18786)

chia/rpc/crawler_rpc_api.py

@@ -1,14 +1,19 @@
 from __future__ import annotations
 
 import ipaddress
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, cast
 
 from chia.rpc.rpc_server import Endpoint, EndpointResult
 from chia.seeder.crawler import Crawler
 from chia.util.ws_message import WsRpcMessage, create_payload_dict
 
 
 class CrawlerRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("CrawlerRpcApi", None)
+
     def __init__(self, crawler: Crawler):
         self.service = crawler
         self.service_name = "chia_crawler"

chia/rpc/data_layer_rpc_api.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, Union, cast
 
 from chia.data_layer.data_layer_errors import OfferIntegrityError
 from chia.data_layer.data_layer_util import (
@@ -89,6 +89,11 @@ def get_fee(config: dict[str, Any], request: dict[str, Any]) -> uint64:
 
 
 class DataLayerRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("DataLayerRpcApi", None)
+
     # TODO: other RPC APIs do not accept a wallet and the service start does not expect to provide one
     def __init__(self, data_layer: DataLayer):  # , wallet: DataLayerWallet):
         self.service: DataLayer = data_layer

chia/rpc/farmer_rpc_api.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 import operator
-from typing import Any, Callable, ClassVar, Optional
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Optional, cast
 
 from typing_extensions import Protocol
 
@@ -78,6 +78,11 @@ def plot_matches_filter(plot: Plot, filter_item: FilterItem) -> bool:
 
 
 class FarmerRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("FarmerRpcApi", None)
+
     def __init__(self, farmer: Farmer):
         self.service = farmer
         self.service_name = "chia_farmer"

chia/rpc/full_node_rpc_api.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, cast
 
 from chia.consensus.block_record import BlockRecord
 from chia.consensus.blockchain import Blockchain, BlockchainMutexPriority
@@ -82,6 +82,11 @@ async def get_average_block_time(
 
 
 class FullNodeRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("FullNodeRpcApi", None)
+
     def __init__(self, service: FullNode) -> None:
         self.service = service
         self.service_name = "chia_full_node"

chia/rpc/harvester_rpc_api.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, cast
 
 from chia.harvester.harvester import Harvester
 from chia.rpc.rpc_server import Endpoint, EndpointResult
@@ -9,6 +9,11 @@
 
 
 class HarvesterRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("HarvesterRpcApi", None)
+
     def __init__(self, harvester: Harvester):
         self.service = harvester
         self.service_name = "chia_harvester"

chia/rpc/timelord_rpc_api.py

@@ -1,13 +1,18 @@
 from __future__ import annotations
 
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, cast
 
 from chia.rpc.rpc_server import Endpoint
 from chia.timelord.timelord import Timelord
 from chia.util.ws_message import WsRpcMessage, create_payload_dict
 
 
 class TimelordRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("TimelordRpcApi", None)
+
     def __init__(self, timelord: Timelord):
         self.service = timelord
         self.service_name = "chia_timelord"

chia/rpc/wallet_rpc_api.py

@@ -5,7 +5,7 @@
 import logging
 import zlib
 from pathlib import Path
-from typing import Any, ClassVar, Optional, Union
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, Union, cast
 
 from chia_rs import AugSchemeMPL, G1Element, G2Element, PrivateKey
 from clvm_tools.binutils import assemble
@@ -165,6 +165,11 @@
 
 
 class WalletRpcApi:
+    if TYPE_CHECKING:
+        from chia.rpc.rpc_server import RpcApiProtocol
+
+        _protocol_check: ClassVar[RpcApiProtocol] = cast("WalletRpcApi", None)
+
     max_get_coin_records_limit: ClassVar[uint32] = uint32(1000)
     max_get_coin_records_filter_items: ClassVar[uint32] = uint32(1000)