starred/chia-blockchain
1
0
Fork 0
mirror of https://github.com/Chia-Network/chia-blockchain.git synced 2026-08-24 10:05:29 -05:00
Code Issues Packages Projects Releases Wiki Activity

[LABS-471] Remove contravariant from WalletProtocol (#21041)

Browse Source 
* [LABS-470] Fix mismatching Wallets with WalletProtocol

* [LABS-471] Remove contravariant from WalletProtocol
This commit is contained in:
Matt Hauff
2026-07-06 11:46:24 -05:00
committed by GitHub
parent 318588a7e9
commit 777811688d
16 changed files with 31 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chia/_tests/wallet/rpc/test_wallet_rpc.py
+1 -1
View File
@@ -249,7 +249,7 @@ async def assert_push_tx_error(node_rpc: FullNodeRpcClient, tx: TransactionRecor
raise ValueError from error
async def assert_get_balance(rpc_client: WalletRpcClient, wallet_node: WalletNode, wallet: WalletProtocol[Any]) -> None:
async def assert_get_balance(rpc_client: WalletRpcClient, wallet_node: WalletNode, wallet: WalletProtocol) -> None:
expected_balance = await wallet_node.get_balance(wallet.id())
expected_balance_dict = expected_balance.to_json_dict()
expected_balance_dict.setdefault("pending_approval_balance", None)
chia/data_layer/data_layer_wallet.py
+1 -1
View File
@@ -108,7 +108,7 @@ class DataLayerSummary(Streamable):
class DataLayerWallet:
if TYPE_CHECKING:
# TODO Create DataLayer coin data model if necessary
_protocol_check: ClassVar[WalletProtocol[object]] = cast("DataLayerWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("DataLayerWallet", None)
wallet_state_manager: WalletStateManager
log: logging.Logger
chia/pools/pool_wallet.py
+1 -1
View File
@@ -63,7 +63,7 @@ class PoolWallet:
if TYPE_CHECKING:
from chia.wallet.wallet_protocol import WalletProtocol
_protocol_check: ClassVar[WalletProtocol[object]] = cast("PoolWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("PoolWallet", None)
MINIMUM_INITIAL_BALANCE: ClassVar[int] = 1
MINIMUM_RELATIVE_LOCK_HEIGHT: ClassVar[int] = 5
chia/wallet/cat_wallet/cat_wallet.py
+4 -4
View File
@@ -83,7 +83,7 @@ def not_ephemeral_additions(sp: WalletSpendBundle) -> list[Coin]:
class CATWallet:
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[CATCoinData]] = cast("CATWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("CATWallet", None)
wallet_state_manager: WalletStateManager
log: logging.Logger
@@ -356,11 +356,11 @@ class CATWallet:
def get_asset_id(self) -> bytes32:
return self.cat_info.limitations_program_hash
async def coin_added(
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: CATCoinData | None
) -> None:
async def coin_added(self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: object | None) -> None:
"""Notification from wallet state manager that wallet has been received."""
self.log.info(f"CAT wallet has been notified that {coin.name().hex()} was added")
if coin_data is not None:
assert isinstance(coin_data, CATCoinData)
inner_puzzle = await self.inner_puzzle_for_cat_puzhash(coin.puzzle_hash)
lineage_proof = LineageProof(coin.parent_coin_info, inner_puzzle.get_tree_hash(), uint64(coin.amount))
chia/wallet/cat_wallet/r_cat_wallet.py
+2 -2
View File
@@ -14,7 +14,7 @@ from chia.types.blockchain_format.coin import Coin
from chia.types.blockchain_format.program import Program
from chia.util.streamable import Streamable, streamable
from chia.wallet.cat_wallet.cat_constants import DEFAULT_CATS
from chia.wallet.cat_wallet.cat_info import CATCoinData, RCATInfo
from chia.wallet.cat_wallet.cat_info import RCATInfo
from chia.wallet.cat_wallet.cat_utils import (
CAT_MOD,
CAT_MOD_HASH,
@@ -56,7 +56,7 @@ class RCATMetadata(Streamable):
class RCATWallet(CATWallet):
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[CATCoinData]] = cast("RCATWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("RCATWallet", None)
wallet_state_manager: WalletStateManager
log: logging.Logger
chia/wallet/did_wallet/did_wallet.py
+2 -4
View File
@@ -58,7 +58,7 @@ if TYPE_CHECKING:
class DIDWallet:
if TYPE_CHECKING:
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[DIDCoinData]] = cast("DIDWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("DIDWallet", None)
wallet_state_manager: WalletStateManager
log: logging.Logger
@@ -348,9 +348,7 @@ class DIDWallet:
# We can improve this interface by passing in the CoinSpend, as well
# We need to change DID Wallet coin_added to expect p2 spends as well as recovery spends,
# or only call it in the recovery spend case
async def coin_added(
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: DIDCoinData | None
) -> None:
async def coin_added(self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: object | None) -> None:
"""Notification from wallet state manager that wallet has been received."""
parent = self.get_parent_for_coin(coin)
if coin_data is not None:
chia/wallet/nft_wallet/nft_wallet.py
+2 -4
View File
@@ -70,7 +70,7 @@ def compute_royalty_amount(offered_amount: int, royalty_split: int, percentage:
class NFTWallet:
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[NFTCoinData]] = cast("NFTWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("NFTWallet", None)
wallet_state_manager: Any
log: logging.Logger
@@ -168,9 +168,7 @@ class NFTWallet:
raise KeyError(f"Couldn't find coin with id: {nft_coin_id}")
return nft_coin
async def coin_added(
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: NFTCoinData | None
) -> None:
async def coin_added(self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: object | None) -> None:
"""Notification from wallet state manager that wallet has been received."""
self.log.info(f"NFT wallet %s has been notified that {coin} was added", self.get_name())
if await self.nft_store.exists(coin.name()):
chia/wallet/plotnft_wallet/plotnft_wallet.py
+1 -1
View File
@@ -35,7 +35,7 @@ class PlotNFT2Wallet:
if TYPE_CHECKING:
from chia.wallet.wallet_protocol import WalletProtocol
_protocol_check: ClassVar[WalletProtocol[object]] = cast("PlotNFT2Wallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("PlotNFT2Wallet", None)
wallet_state_manager: WalletStateManager
xch_wallet: Wallet
chia/wallet/remote_wallet/remote_wallet.py
+1 -1
View File
@@ -22,7 +22,7 @@ from chia.wallet.wallet_protocol import WalletProtocol
# Furthermore the wallet will act mainly as a sentinel for CoinRecords that are related to the remote wallet.
class RemoteWallet:
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[object]] = cast("RemoteWallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("RemoteWallet", None)
wallet_state_manager: Any
log: logging.Logger
chia/wallet/trade_manager.py
+1 -1
View File
@@ -825,7 +825,7 @@ class TradeManager:
for asset_id, amount in arbitrage.items():
if asset_id is None:
wallet: WalletProtocol[Any] | None = self.wallet_state_manager.main_wallet
wallet: WalletProtocol | None = self.wallet_state_manager.main_wallet
assert wallet is not None
key: bytes32 | int = int(wallet.id())
else:
chia/wallet/util/wallet_types.py
+1 -1
View File
@@ -62,7 +62,7 @@ class WalletIdentifier:
type: WalletType
@classmethod
def create(cls, wallet: WalletProtocol[T_contra]) -> WalletIdentifier:
def create(cls, wallet: WalletProtocol) -> WalletIdentifier:
return cls(wallet.id(), wallet.type())
chia/wallet/vc_wallet/cr_cat_wallet.py
+3 -5
View File
@@ -16,7 +16,7 @@ from chia.types.blockchain_format.program import Program
from chia.util.byte_types import hexstr_to_bytes
from chia.util.hash import std_hash
from chia.util.streamable import VersionedBlob
from chia.wallet.cat_wallet.cat_info import CATCoinData, CRCATInfo
from chia.wallet.cat_wallet.cat_info import CRCATInfo
from chia.wallet.cat_wallet.cat_utils import CAT_MOD_HASH, CAT_MOD_HASH_HASH, construct_cat_puzzle
from chia.wallet.cat_wallet.cat_wallet import CATWallet
from chia.wallet.coin_selection import select_coins
@@ -205,9 +205,7 @@ class CRCATWallet(CATWallet):
async def set_tail_program(self, tail_program: str) -> None: # pragma: no cover
raise NotImplementedError("set_tail_program is a legacy method and is not available on CR-CAT wallets")
async def coin_added(
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: CATCoinData | None
) -> None:
async def coin_added(self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: object | None) -> None:
"""Notification from wallet state manager that wallet has been received."""
self.log.info(f"CR-CAT wallet has been notified that {coin.name().hex()} was added")
try:
@@ -868,4 +866,4 @@ class CRCATWallet(CATWallet):
if TYPE_CHECKING:
_dummy: WalletProtocol[CATCoinData] = CRCATWallet()
_dummy: WalletProtocol = CRCATWallet()
chia/wallet/vc_wallet/vc_wallet.py
+1 -1
View File
@@ -630,4 +630,4 @@ class VCWallet:
if TYPE_CHECKING:
_dummy: WalletProtocol[VerifiedCredential] = VCWallet() # pragma: no cover
_dummy: WalletProtocol = VCWallet() # pragma: no cover
chia/wallet/wallet.py
+1 -2
View File
@@ -27,7 +27,6 @@ from chia.wallet.derive_keys import (
_derive_path_unhardened,
master_sk_to_singleton_owner_sk,
)
from chia.wallet.puzzles.clawback.metadata import ClawbackMetadata
from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import (
DEFAULT_HIDDEN_PUZZLE_HASH,
calculate_synthetic_offset,
@@ -66,7 +65,7 @@ if TYPE_CHECKING:
class Wallet:
if TYPE_CHECKING:
_protocol_check: ClassVar[WalletProtocol[ClawbackMetadata]] = cast("Wallet", None)
_protocol_check: ClassVar[WalletProtocol] = cast("Wallet", None)
wallet_info: WalletInfo
wallet_state_manager: WalletStateManager
chia/wallet/wallet_protocol.py
+2 -2
View File
@@ -24,14 +24,14 @@ if TYPE_CHECKING:
T_contra = TypeVar("T_contra", contravariant=True)
class WalletProtocol(Protocol[T_contra]):
class WalletProtocol(Protocol):
@classmethod
def type(cls) -> WalletType: ...
def id(self) -> uint32: ...
async def coin_added(
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: T_contra | None
self, coin: Coin, height: uint32, peer: WSChiaConnection, coin_data: object | None
) -> None: ...
async def select_coins(
chia/wallet/wallet_state_manager.py
+7 -7
View File
@@ -150,7 +150,7 @@ from chia.wallet.wallet_transaction_store import WalletTransactionStore
from chia.wallet.wallet_user_store import WalletUserStore
from chia.wallet.wsm_apis import CreateMorePuzzleHashesResult, GetUnusedDerivationRecordResult
TWalletType = TypeVar("TWalletType", bound=WalletProtocol[Any])
TWalletType = TypeVar("TWalletType", bound=WalletProtocol)
if TYPE_CHECKING:
from chia.wallet.wallet_node import WalletNode
@@ -195,7 +195,7 @@ class WalletStateManager:
db_wrapper: DBWrapper2
main_wallet: Wallet
wallets: dict[uint32, WalletProtocol[Any]]
wallets: dict[uint32, WalletProtocol]
private_key: PrivateKey | None
root_pubkey: G1Element
@@ -308,7 +308,7 @@ class WalletStateManager:
pool_config.perform_migration_from_old_config(root_path=self.root_path)
wallet: WalletProtocol[Any] | None = None
wallet: WalletProtocol | None = None
for wallet_info in await self.get_all_wallet_info_entries():
wallet_type = WalletType(wallet_info.type)
if wallet_type == WalletType.STANDARD_WALLET:
@@ -2566,7 +2566,7 @@ class WalletStateManager:
result = await self.coin_store.get_coin_records(**kwargs)
return [await self.get_coin_record_by_wallet_record(record) for record in result.records]
async def get_wallet_for_coin(self, coin_id: bytes32) -> WalletProtocol[Any] | None:
async def get_wallet_for_coin(self, coin_id: bytes32) -> WalletProtocol | None:
coin_record = await self.coin_store.get_coin_record(coin_id)
if coin_record is None:
return None
@@ -2622,7 +2622,7 @@ class WalletStateManager:
async def get_all_wallet_info_entries(self, wallet_type: WalletType | None = None) -> list[WalletInfo]:
return await self.user_store.get_all_wallet_info_entries(wallet_type)
async def get_wallet_for_asset_id(self, asset_id: bytes32) -> WalletProtocol[Any] | None:
async def get_wallet_for_asset_id(self, asset_id: bytes32) -> WalletProtocol | None:
for wallet_id, wallet in self.wallets.items():
if wallet.type() in {WalletType.CAT, WalletType.CRCAT, WalletType.RCAT}:
assert isinstance(wallet, CATWallet)
@@ -2639,7 +2639,7 @@ class WalletStateManager:
return wallet
return None
async def get_wallet_for_puzzle_info(self, puzzle_driver: PuzzleInfo) -> WalletProtocol[Any] | None:
async def get_wallet_for_puzzle_info(self, puzzle_driver: PuzzleInfo) -> WalletProtocol | None:
for wallet in self.wallets.values():
match_function = getattr(wallet, "match_puzzle_info", None)
if match_function is not None and callable(match_function):
@@ -2660,7 +2660,7 @@ class WalletStateManager:
},
)
async def add_new_wallet(self, wallet: WalletProtocol[Any]) -> None:
async def add_new_wallet(self, wallet: WalletProtocol) -> None:
self.wallets[wallet.id()] = wallet
result = await self.create_more_puzzle_hashes()
await result.commit(self)