feat: implement TokenWipeTransaction (#90)

* feat: add wipe_key attribute to TokenCreateTransaction

Signed-off-by: dosi <[email protected]>

* feat: implement class TokenWipeTransaction

Signed-off-by: dosi <[email protected]>

* test: add unit and integration tests

Signed-off-by: dosi <[email protected]>

* docs: add comprehensive token wipe transaction example

Signed-off-by: dosi <[email protected]>

* docs: add token wipe transaction documentation to README

Signed-off-by: dosi <[email protected]>

* docs: split token_wipe_transaction example into smaller functions

Signed-off-by: dosi <[email protected]>

* chore: address PR review comments for files in examples folder

Signed-off-by: dosi <[email protected]>

* chore: rename token_wipe_transaction to token_wipe

Signed-off-by: dosi <[email protected]>

* style: remove spaces around equals in function parameters

Signed-off-by: dosi <[email protected]>

---------

Signed-off-by: dosi <[email protected]>

Author: Dosik13

examples/README.md

@@ -20,6 +20,7 @@ You can choose either syntax or even mix both styles in your projects.
   - [Associating a Token](#associating-a-token)
   - [Dissociating a Token](#dissociating-a-token)
   - [Transferring Tokens](#transferring-tokens)
+  - [Wiping Tokens](#wiping-tokens)
   - [Deleting a Token](#deleting-a-token)
   - [Freezing a Token](#freezing-a-token)
   - [Unfreezing a Token](#unfreezing-a-token)
@@ -261,6 +262,32 @@ transaction.execute(client)
     transaction.execute(client)
 ```
 
+### Wiping tokens
+
+#### Pythonic Syntax:
+```
+transaction = TokenWipeTransaction(
+    token_id=token_id,
+    account_id=account_id,
+    amount=amount
+).freeze_with(client)
+
+transaction.execute(client)
+
+```
+#### Method Chaining:
+```
+    transaction = (
+        TokenWipeTransaction()
+        .set_token_id(token_id)
+        .set_account_id(account_id)
+        .set_amount(amount)
+        .freeze_with(client)
+    )
+
+    transaction.execute(client)
+```
+
 ### Deleting a Token
 
 #### Pythonic Syntax:

examples/token_wipe.py

@@ -0,0 +1,176 @@
+import os
+import sys
+from dotenv import load_dotenv
+
+from hiero_sdk_python import (
+    Client,
+    AccountId,
+    PrivateKey,
+    Network,
+    TransferTransaction,
+    TokenAssociateTransaction,
+)
+from hiero_sdk_python.account.account_create_transaction import AccountCreateTransaction
+from hiero_sdk_python.hbar import Hbar
+from hiero_sdk_python.response_code import ResponseCode
+from hiero_sdk_python.tokens.supply_type import SupplyType
+from hiero_sdk_python.tokens.token_create_transaction import TokenCreateTransaction
+from hiero_sdk_python.tokens.token_type import TokenType
+from hiero_sdk_python.tokens.token_wipe_transaction import TokenWipeTransaction
+
+load_dotenv()
+
+def setup_client():
+    """Initialize and set up the client with operator account"""
+    # Initialize network and client
+    network = Network(network='testnet')
+    client = Client(network)
+
+    # Set up operator account
+    operator_id = AccountId.from_string(os.getenv('OPERATOR_ID'))
+    operator_key = PrivateKey.from_string(os.getenv('OPERATOR_KEY'))
+    client.set_operator(operator_id, operator_key)
+    
+    return client, operator_id, operator_key
+
+def create_test_account(client):
+    """Create a new account for testing"""
+    # Generate private key for new account
+    new_account_private_key = PrivateKey.generate()
+    new_account_public_key = new_account_private_key.public_key()
+    
+    # Create new account with initial balance of 1 HBAR
+    transaction = (
+        AccountCreateTransaction()
+        .set_key(new_account_public_key)
+        .set_initial_balance(Hbar(1))
+        .freeze_with(client)
+    )
+    
+    receipt = transaction.execute(client)
+    
+    # Check if account creation was successful
+    if receipt.status != ResponseCode.SUCCESS:
+        print(f"Account creation failed with status: {ResponseCode.get_name(receipt.status)}")
+        sys.exit(1)
+    
+    # Get account ID from receipt
+    account_id = receipt.accountId
+    print(f"New account created with ID: {account_id}")
+    
+    return account_id, new_account_private_key
+
+def create_token(client, operator_id, operator_key):
+    """Create a fungible token"""
+    # Create fungible token
+    # Note: The wipe key is required to perform token wipe operations
+    transaction = (
+        TokenCreateTransaction()
+        .set_token_name("Token123")
+        .set_token_symbol("T123")
+        .set_decimals(2)
+        .set_initial_supply(10)
+        .set_treasury_account_id(operator_id)
+        .set_token_type(TokenType.FUNGIBLE_COMMON)
+        .set_supply_type(SupplyType.FINITE)
+        .set_max_supply(100)
+        .set_admin_key(operator_key)      # For token management
+        .set_supply_key(operator_key)     # For minting/burning
+        .set_freeze_key(operator_key)     # For freezing accounts
+        .set_wipe_key(operator_key)       # Required for wiping tokens
+        .freeze_with(client)
+    )
+    
+    receipt = transaction.execute(client)
+    
+    # Check if token creation was successful
+    if receipt.status != ResponseCode.SUCCESS:
+        print(f"Token creation failed with status: {ResponseCode.get_name(receipt.status)}")
+        sys.exit(1)
+    
+    # Get token ID from receipt
+    token_id = receipt.tokenId
+    print(f"Token created with ID: {token_id}")
+    
+    return token_id
+
+def associate_token(client, account_id, token_id, account_private_key):
+    """Associate a token with an account"""
+    # Associate the token with the new account
+    # Note: Accounts must be associated with tokens before they can receive them
+    associate_transaction = (
+        TokenAssociateTransaction()
+        .set_account_id(account_id)
+        .add_token_id(token_id)
+        .freeze_with(client)
+        .sign(account_private_key) # Has to be signed by new account's key
+    )
+    
+    receipt = associate_transaction.execute(client)
+    
+    if receipt.status != ResponseCode.SUCCESS:
+        print(f"Token association failed with status: {ResponseCode.get_name(receipt.status)}")
+        sys.exit(1)
+    
+    print("Token successfully associated with account")
+
+def transfer_tokens(client, token_id, operator_id, account_id, amount):
+    """Transfer tokens from operator to the specified account"""
+    # Transfer tokens to the new account
+    # Note: Negative amount for sender, positive for receiver
+    transfer_transaction = (
+        TransferTransaction()
+        .add_token_transfer(token_id, operator_id, -amount)  # From operator
+        .add_token_transfer(token_id, account_id, amount)    # To new account
+        .freeze_with(client)
+    )
+    
+    receipt = transfer_transaction.execute(client)
+    
+    # Check if token transfer was successful
+    if receipt.status != ResponseCode.SUCCESS:
+        print(f"Token transfer failed with status: {ResponseCode.get_name(receipt.status)}")
+        sys.exit(1)
+    
+    print(f"Successfully transferred {amount} tokens to account {account_id}")
+
+def wipe_tokens(client, token_id, account_id, amount):
+    """Wipe tokens from the specified account"""
+    # Wipe the tokens from the account
+    # Note: This requires the wipe key that was specified during token creation
+    transaction = (
+        TokenWipeTransaction()
+        .set_token_id(token_id)
+        .set_account_id(account_id)
+        .set_amount(amount)
+        .freeze_with(client)
+    )
+    
+    receipt = transaction.execute(client)
+    
+    if receipt.status != ResponseCode.SUCCESS:
+        print(f"Token wipe failed with status: {ResponseCode.get_name(receipt.status)}")
+        sys.exit(1)
+    
+    print(f"Successfully wiped {amount} tokens from account {account_id}")
+
+def token_wipe():
+    """
+    Demonstrates the token wipe functionality by:
+    1. Creating a new account
+    2. Creating a fungible token with wipe capability
+    3. Associating the token with the new account
+    4. Transferring tokens to the new account
+    5. Wiping the tokens from the account
+    """
+    client, operator_id, operator_key = setup_client()
+    account_id, new_account_private_key = create_test_account(client)
+    token_id = create_token(client, operator_id, operator_key)
+    associate_token(client, account_id, token_id, new_account_private_key)
+    
+    amount = 10
+    transfer_tokens(client, token_id, operator_id, account_id, amount)
+    wipe_tokens(client, token_id, account_id, amount)
+
+if __name__ == "__main__":
+    token_wipe()

src/hiero_sdk_python/tokens/token_create_transaction.py

@@ -177,11 +177,13 @@ class TokenKeys:
         admin_key: The admin key for the token to update and delete.
         supply_key: The supply key for the token to mint and burn.
         freeze_key: The freeze key for the token to freeze and unfreeze.
+        wipe_key: The wipe key for the token to wipe tokens from an account.
     """
 
     admin_key: Optional[PrivateKey] = None
     supply_key: Optional[PrivateKey] = None
     freeze_key: Optional[PrivateKey] = None
+    wipe_key: Optional[PrivateKey] = None
 
 class TokenCreateTransaction(Transaction):
     """
@@ -312,6 +314,11 @@ def set_freeze_key(self, key):
         self._require_not_frozen()
         self._keys.freeze_key = key
         return self
+    
+    def set_wipe_key(self, key):
+        self._require_not_frozen()
+        self._keys.wipe_key = key
+        return self
 
     def build_transaction_body(self):
         """
@@ -344,7 +351,11 @@ def build_transaction_body(self):
         if self._keys.freeze_key:
             freeze_public_key_bytes = self._keys.freeze_key.public_key().to_bytes_raw()
             freeze_key_proto = basic_types_pb2.Key(ed25519=freeze_public_key_bytes)
-
+            
+        wipe_key_proto = None
+        if self._keys.wipe_key:
+            wipe_public_key_bytes = self._keys.wipe_key.public_key().to_bytes_raw()
+            wipe_key_proto = basic_types_pb2.Key(ed25519=wipe_public_key_bytes)
 
         # Ensure token type is correctly set with default to fungible
         if self._token_params.token_type is None:
@@ -376,6 +387,7 @@ def build_transaction_body(self):
             adminKey=admin_key_proto,
             supplyKey=supply_key_proto,
             freezeKey=freeze_key_proto,
+            wipeKey=wipe_key_proto
         )
         # Build the base transaction body and attach the token creation details
         transaction_body = self.build_base_transaction_body()

src/hiero_sdk_python/tokens/token_wipe_transaction.py

@@ -0,0 +1,99 @@
+from hiero_sdk_python.tokens.token_id import TokenId
+from hiero_sdk_python.account.account_id import AccountId
+from hiero_sdk_python.transaction.transaction import Transaction
+from hiero_sdk_python.hapi.services.token_wipe_account_pb2 import TokenWipeAccountTransactionBody
+from hiero_sdk_python.channels import _Channel
+from hiero_sdk_python.executable import _Method
+
+class TokenWipeTransaction(Transaction):
+    """
+    Represents a token wipe transaction on the Hedera network.
+    
+    This transaction wipes (removes) tokens from an account.
+    
+    Inherits from the base Transaction class and implements the required methods
+    to build and execute a token wipe transaction.
+    """
+    def __init__(self, token_id=None, account_id=None, amount=None, serial=[]):
+        """
+        Initializes a new TokenWipeTransaction instance with optional token_id and account_id.
+
+        Args:
+            token_id (TokenId, optional): The ID of the token to be wiped.
+            account_id (AccountId, optional): The ID of the account to have their tokens wiped.
+            amount (int, optional): The amount of tokens to wipe.
+            serial (list[int], optional): The serial numbers of NFTs to wipe.
+        """
+        super().__init__()
+        self.token_id : TokenId = token_id
+        self.account_id : AccountId = account_id
+        self.amount : int = amount
+        self.serial : list[int] = serial
+        
+    def set_token_id(self, token_id):
+        """
+        Sets the ID of the token to be wiped.
+
+        Args:
+            token_id (TokenId): The ID of the token to be wiped.
+
+        Returns:
+            TokenWipeTransaction: Returns self for method chaining.
+        """
+        self._require_not_frozen()
+        self.token_id = token_id
+        return self
+    
+    def set_account_id(self, account_id):
+        self._require_not_frozen()
+        self.account_id = account_id
+        return self
+    
+    def set_amount(self, amount):
+        self._require_not_frozen()
+        self.amount = amount
+        return self
+    
+    def set_serial(self, serial):
+        self._require_not_frozen()
+        self.serial = serial
+        return self
+    
+    def build_transaction_body(self):
+        """
+        Builds and returns the protobuf transaction body for token wipe.
+
+        Returns:
+            TransactionBody: The protobuf transaction body containing the token wipe details.
+        """
+        token_wipe_body = TokenWipeAccountTransactionBody(
+            token=self.token_id and self.token_id.to_proto(),
+            account=self.account_id and self.account_id.to_proto(),
+            amount=self.amount,
+            serialNumbers=self.serial
+        )
+        transaction_body = self.build_base_transaction_body()
+        transaction_body.tokenWipe.CopyFrom(token_wipe_body)
+        return transaction_body
+    
+    def _get_method(self, channel: _Channel) -> _Method:
+        return _Method(
+            transaction_func=channel.token.wipeTokenAccount,
+            query_func=None
+        )
+    
+    def _from_proto(self, proto: TokenWipeAccountTransactionBody):
+        """
+        Deserializes a TokenWipeAccountTransactionBody from a protobuf object.
+
+        Args:
+            proto (TokenWipeAccountTransactionBody): The protobuf object to deserialize.
+
+        Returns:
+            TokenWipeTransaction: Returns self for method chaining.
+        """
+        self.token_id = TokenId.from_proto(proto.token)
+        self.account_id = AccountId.from_proto(proto.account)
+        self.amount = proto.amount
+        self.serial = proto.serialNumbers
+        return self