full node stubs

Author: richardkiss

chia/apis/__init__.py

@@ -2,7 +2,8 @@
 
 from chia.apis.api_protocol_registry import ApiProtocolRegistry
 from chia.apis.farmer_stub import FarmerApiStub
+from chia.apis.full_node_stub import FullNodeApiStub
 from chia.apis.harvester_stub import HarvesterApiStub
 from chia.apis.timelord_stub import TimelordApiStub
 
-__all__ = ["ApiProtocolRegistry", "FarmerApiStub", "HarvesterApiStub", "TimelordApiStub"]
+__all__ = ["ApiProtocolRegistry", "FarmerApiStub", "FullNodeApiStub", "HarvesterApiStub", "TimelordApiStub"]

chia/apis/api_protocol_registry.py

@@ -1,16 +1,16 @@
 from __future__ import annotations
 
 from chia.apis.farmer_stub import FarmerApiStub
+from chia.apis.full_node_stub import FullNodeApiStub
 from chia.apis.harvester_stub import HarvesterApiStub
 from chia.apis.timelord_stub import TimelordApiStub
-from chia.full_node.full_node_api import FullNodeAPI
 from chia.introducer.introducer_api import IntroducerAPI
 from chia.protocols.outbound_message import NodeType
 from chia.server.api_protocol import ApiProtocol
 from chia.wallet.wallet_node_api import WalletNodeAPI
 
 ApiProtocolRegistry: dict[NodeType, type[ApiProtocol]] = {
-    NodeType.FULL_NODE: FullNodeAPI,
+    NodeType.FULL_NODE: FullNodeApiStub,
     NodeType.WALLET: WalletNodeAPI,
     NodeType.INTRODUCER: IntroducerAPI,
     NodeType.TIMELORD: TimelordApiStub,

chia/apis/full_node_stub.py

@@ -0,0 +1,350 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, ClassVar, Optional, cast
+
+if TYPE_CHECKING:
+    from chia.server.api_protocol import ApiProtocol
+
+# Minimal imports to avoid circular dependencies
+from chia.protocols import farmer_protocol, full_node_protocol, introducer_protocol, timelord_protocol, wallet_protocol
+from chia.protocols.outbound_message import Message
+from chia.protocols.protocol_message_types import ProtocolMessageTypes
+from chia.server.api_protocol import ApiMetadata
+from chia.server.ws_connection import WSChiaConnection
+
+
+class FullNodeApiStub:
+    """Lightweight API stub for FullNodeAPI to break circular dependencies."""
+
+    if TYPE_CHECKING:
+        _protocol_check: ClassVar[ApiProtocol] = cast("FullNodeApiStub", None)
+
+    log: logging.Logger
+    metadata: ClassVar[ApiMetadata] = ApiMetadata()
+
+    def ready(self) -> bool:
+        """Check if the full node is ready."""
+        return True
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_peers])
+    async def request_peers(
+        self, _request: full_node_protocol.RequestPeers, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle peer request."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def respond_peers(
+        self, request: full_node_protocol.RespondPeers, peer: WSChiaConnection
+    ) -> None:
+        """Handle peer response."""
+
+    @metadata.request(peer_required=True)
+    async def respond_peers_introducer(
+        self, request: introducer_protocol.RespondPeersIntroducer, peer: WSChiaConnection
+    ) -> None:
+        """Handle introducer peer response."""
+
+    @metadata.request(peer_required=True, execute_task=True)
+    async def new_peak(self, request: full_node_protocol.NewPeak, peer: WSChiaConnection) -> None:
+        """Handle new peak."""
+
+    @metadata.request(peer_required=True)
+    async def new_transaction(
+        self, transaction: full_node_protocol.NewTransaction, peer: WSChiaConnection
+    ) -> None:
+        """Handle new transaction."""
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_transaction])
+    async def request_transaction(self, request: full_node_protocol.RequestTransaction) -> Optional[Message]:
+        """Handle transaction request."""
+        return None
+
+    @metadata.request(peer_required=True, bytes_required=True)
+    async def respond_transaction(
+        self,
+        response: full_node_protocol.RespondTransaction,
+        peer: WSChiaConnection,
+        tx_bytes: bytes = b"",
+    ) -> None:
+        """Handle transaction response."""
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_proof_of_weight])
+    async def request_proof_of_weight(self, request: full_node_protocol.RequestProofOfWeight) -> Optional[Message]:
+        """Handle proof of weight request."""
+        return None
+
+    @metadata.request()
+    async def respond_proof_of_weight(self, request: full_node_protocol.RespondProofOfWeight) -> Optional[Message]:
+        """Handle proof of weight response."""
+        return None
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_block, ProtocolMessageTypes.reject_block])
+    async def request_block(self, request: full_node_protocol.RequestBlock) -> Optional[Message]:
+        """Handle block request."""
+        return None
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_blocks, ProtocolMessageTypes.reject_blocks])
+    async def request_blocks(self, request: full_node_protocol.RequestBlocks) -> Optional[Message]:
+        """Handle blocks request."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def reject_block(
+        self,
+        reject_block: full_node_protocol.RejectBlock,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle block rejection."""
+
+    @metadata.request(peer_required=True)
+    async def reject_blocks(
+        self,
+        reject_blocks_request: full_node_protocol.RejectBlocks,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle blocks rejection."""
+
+    @metadata.request(peer_required=True)
+    async def respond_blocks(
+        self,
+        request: full_node_protocol.RespondBlocks,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle blocks response."""
+
+    @metadata.request(peer_required=True)
+    async def respond_block(
+        self,
+        request: full_node_protocol.RespondBlock,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle block response."""
+
+    @metadata.request()
+    async def new_unfinished_block(
+        self, new_unfinished_block: full_node_protocol.NewUnfinishedBlock
+    ) -> Optional[Message]:
+        """Handle new unfinished block."""
+        return None
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_unfinished_block])
+    async def request_unfinished_block(
+        self, request_unfinished_block: full_node_protocol.RequestUnfinishedBlock
+    ) -> Optional[Message]:
+        """Handle unfinished block request."""
+        return None
+
+    @metadata.request()
+    async def new_unfinished_block2(
+        self, new_unfinished_block: full_node_protocol.NewUnfinishedBlock2
+    ) -> Optional[Message]:
+        """Handle new unfinished block v2."""
+        return None
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_unfinished_block])
+    async def request_unfinished_block2(
+        self, request_unfinished_block: full_node_protocol.RequestUnfinishedBlock2
+    ) -> Optional[Message]:
+        """Handle unfinished block request v2."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def respond_unfinished_block(
+        self,
+        respond_unfinished_block: full_node_protocol.RespondUnfinishedBlock,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle unfinished block response."""
+
+    @metadata.request(peer_required=True)
+    async def new_signage_point_or_end_of_sub_slot(
+        self, new_sp: full_node_protocol.NewSignagePointOrEndOfSubSlot, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new signage point or end of sub slot."""
+        return None
+
+    @metadata.request(
+        reply_types=[ProtocolMessageTypes.respond_signage_point, ProtocolMessageTypes.respond_end_of_sub_slot]
+    )
+    async def request_signage_point_or_end_of_sub_slot(
+        self, request: full_node_protocol.RequestSignagePointOrEndOfSubSlot
+    ) -> Optional[Message]:
+        """Handle signage point or end of sub slot request."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def respond_signage_point(
+        self, request: full_node_protocol.RespondSignagePoint, peer: WSChiaConnection
+    ) -> None:
+        """Handle signage point response."""
+
+    @metadata.request(peer_required=True)
+    async def respond_end_of_sub_slot(
+        self, request: full_node_protocol.RespondEndOfSubSlot, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle end of sub slot response."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def request_mempool_transactions(
+        self,
+        request: full_node_protocol.RequestMempoolTransactions,
+        peer: WSChiaConnection,
+    ) -> None:
+        """Handle mempool transactions request."""
+
+    @metadata.request(peer_required=True)
+    async def declare_proof_of_space(
+        self, request: farmer_protocol.DeclareProofOfSpace, peer: WSChiaConnection
+    ) -> None:
+        """Handle proof of space declaration."""
+
+    @metadata.request(peer_required=True)
+    async def signed_values(
+        self, farmer_request: farmer_protocol.SignedValues, peer: WSChiaConnection
+    ) -> None:
+        """Handle signed values."""
+
+    @metadata.request(peer_required=True)
+    async def new_infusion_point_vdf(
+        self, request: timelord_protocol.NewInfusionPointVDF, peer: WSChiaConnection
+    ) -> None:
+        """Handle new infusion point VDF."""
+
+    @metadata.request(peer_required=True)
+    async def new_signage_point_vdf(
+        self, request: timelord_protocol.NewSignagePointVDF, peer: WSChiaConnection
+    ) -> None:
+        """Handle new signage point VDF."""
+
+    @metadata.request(peer_required=True)
+    async def new_end_of_sub_slot_vdf(
+        self, request: timelord_protocol.NewEndOfSubSlotVDF, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle new end of sub slot VDF."""
+        return None
+
+    @metadata.request()
+    async def request_block_header(self, request: wallet_protocol.RequestBlockHeader) -> Optional[Message]:
+        """Handle block header request."""
+        return None
+
+    @metadata.request()
+    async def request_additions(self, request: wallet_protocol.RequestAdditions) -> Optional[Message]:
+        """Handle additions request."""
+        return None
+
+    @metadata.request()
+    async def request_removals(self, request: wallet_protocol.RequestRemovals) -> Optional[Message]:
+        """Handle removals request."""
+        return None
+
+    @metadata.request()
+    async def send_transaction(
+        self, request: wallet_protocol.SendTransaction, *, test: bool = False
+    ) -> Optional[Message]:
+        """Handle send transaction."""
+        return None
+
+    @metadata.request()
+    async def request_puzzle_solution(self, request: wallet_protocol.RequestPuzzleSolution) -> Optional[Message]:
+        """Handle puzzle solution request."""
+        return None
+
+    @metadata.request()
+    async def request_block_headers(self, request: wallet_protocol.RequestBlockHeaders) -> Optional[Message]:
+        """Handle block headers request."""
+        return None
+
+    @metadata.request()
+    async def request_header_blocks(self, request: wallet_protocol.RequestHeaderBlocks) -> Optional[Message]:
+        """Handle header blocks request (deprecated)."""
+        return None
+
+    @metadata.request(bytes_required=True, execute_task=True)
+    async def respond_compact_proof_of_time(
+        self, request: timelord_protocol.RespondCompactProofOfTime, request_bytes: bytes = b""
+    ) -> None:
+        """Handle compact proof of time response."""
+
+    @metadata.request(peer_required=True, bytes_required=True, execute_task=True)
+    async def new_compact_vdf(
+        self, request: full_node_protocol.NewCompactVDF, peer: WSChiaConnection, request_bytes: bytes = b""
+    ) -> None:
+        """Handle new compact VDF."""
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_compact_vdf])
+    async def request_compact_vdf(self, request: full_node_protocol.RequestCompactVDF, peer: WSChiaConnection) -> None:
+        """Handle compact VDF request."""
+
+    @metadata.request(peer_required=True)
+    async def respond_compact_vdf(self, request: full_node_protocol.RespondCompactVDF, peer: WSChiaConnection) -> None:
+        """Handle compact VDF response."""
+
+    @metadata.request(peer_required=True)
+    async def register_for_ph_updates(
+        self, request: wallet_protocol.RegisterForPhUpdates, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle puzzle hash updates registration."""
+        return None
+
+    @metadata.request(peer_required=True)
+    async def register_for_coin_updates(
+        self, request: wallet_protocol.RegisterForCoinUpdates, peer: WSChiaConnection
+    ) -> Optional[Message]:
+        """Handle coin updates registration."""
+        return None
+
+    @metadata.request()
+    async def request_children(self, request: wallet_protocol.RequestChildren) -> Optional[Message]:
+        """Handle children request."""
+        return None
+
+    @metadata.request()
+    async def request_ses_hashes(self, request: wallet_protocol.RequestSESInfo) -> Message:
+        """Handle SES hashes request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_fee_estimates])
+    async def request_fee_estimates(self, request: wallet_protocol.RequestFeeEstimates) -> Message:
+        """Handle fee estimates request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(
+        peer_required=True,
+        reply_types=[ProtocolMessageTypes.respond_remove_puzzle_subscriptions],
+    )
+    async def request_remove_puzzle_subscriptions(
+        self, request: wallet_protocol.RequestRemovePuzzleSubscriptions, peer: WSChiaConnection
+    ) -> Message:
+        """Handle remove puzzle subscriptions request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(
+        peer_required=True,
+        reply_types=[ProtocolMessageTypes.respond_remove_coin_subscriptions],
+    )
+    async def request_remove_coin_subscriptions(
+        self, request: wallet_protocol.RequestRemoveCoinSubscriptions, peer: WSChiaConnection
+    ) -> Message:
+        """Handle remove coin subscriptions request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_puzzle_state])
+    async def request_puzzle_state(
+        self, request: wallet_protocol.RequestPuzzleState, peer: WSChiaConnection
+    ) -> Message:
+        """Handle puzzle state request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(peer_required=True, reply_types=[ProtocolMessageTypes.respond_coin_state])
+    async def request_coin_state(self, request: wallet_protocol.RequestCoinState, peer: WSChiaConnection) -> Message:
+        """Handle coin state request."""
+        raise NotImplementedError("Stub method should not be called")
+
+    @metadata.request(reply_types=[ProtocolMessageTypes.respond_cost_info])
+    async def request_cost_info(self, _request: wallet_protocol.RequestCostInfo) -> Optional[Message]:
+        """Handle cost info request."""
+        return None

tach.toml

@@ -90,7 +90,6 @@ path = "chia.apis"
 depends_on = [
     { path = "chia.introducer", deprecated = false },
     { path = "chia.wallet", deprecated = false },
-    { path = "chia.full_node", deprecated = false },
     { path = "chia.protocols", deprecated = false },
     { path = "chia.server", deprecated = false },
 ]