feat(token): Add getTokenInfo request and response models, including custom fee serialization (#2400)

Signed-off-by: Ntege Daniel <danientege785@gmail.com>

Author: danielmarv

src/hiero_sdk_python/query/token_info_query.py

@@ -57,14 +57,12 @@ def _make_request(self) -> query_pb2.Query:
             Exception: If any other error occurs during request construction.
         """
         try:
-            if not self.token_id:
-                raise ValueError("Token ID must be set before making the request.")
-
             query_header = self._make_request_header()
 
             token_info_query = token_get_info_pb2.TokenGetInfoQuery()
             token_info_query.header.CopyFrom(query_header)
-            token_info_query.token.CopyFrom(self.token_id._to_proto())
+            if self.token_id:
+                token_info_query.token.CopyFrom(self.token_id._to_proto())
 
             query = query_pb2.Query()
             query.tokenGetInfo.CopyFrom(token_info_query)

tck/handlers/registry.py

@@ -6,7 +6,7 @@
 
 import inspect
 from collections.abc import Callable
-from dataclasses import asdict
+from dataclasses import asdict, fields as dc_fields
 from typing import Any, get_type_hints
 
 from tck.errors import JsonRpcError, handle_sdk_errors
@@ -81,6 +81,29 @@ def safe_dispatch(method_name: str, params: Any, request_id: str | int | None) -
         return build_json_rpc_error_response(error, request_id)
 
 
+def _strip_none(obj: Any, nullable_keys: set[str] | None = None) -> Any:
+    """Recursively strip None values from nested dicts and lists."""
+    if isinstance(obj, dict):
+        return {
+            k: _strip_none(v)
+            for k, v in obj.items()
+            if v is not None or (nullable_keys is not None and k in nullable_keys)
+        }
+    if isinstance(obj, list):
+        return [_strip_none(item) for item in obj]
+    return obj
+
+
 def parse_result(result: Any) -> dict:
-    """Parse the result from the methods to dict containing non none key:values"""
-    return {k: v for k, v in asdict(result).items() if v is not None}
+    """Parse the result from the methods to dict containing non none key:values.
+
+    Fields with metadata={"nullable": True} are preserved even when None.
+    Recursively strips None from nested structures.
+    """
+    nullable_fields: set[str] = set()
+    for f in dc_fields(result):
+        if f.metadata.get("nullable"):
+            nullable_fields.add(f.name)
+
+    raw = asdict(result)
+    return _strip_none(raw, nullable_keys=nullable_fields)

tck/handlers/token.py

@@ -4,6 +4,8 @@
 
 from hiero_sdk_python.account.account_id import AccountId
 from hiero_sdk_python.Duration import Duration
+from hiero_sdk_python.hbar import Hbar
+from hiero_sdk_python.query.token_info_query import TokenInfoQuery
 from hiero_sdk_python.response_code import ResponseCode
 from hiero_sdk_python.timestamp import Timestamp
 from hiero_sdk_python.tokens.custom_fee import CustomFee
@@ -19,9 +21,13 @@
 from hiero_sdk_python.tokens.token_associate_transaction import TokenAssociateTransaction
 from hiero_sdk_python.tokens.token_create_transaction import TokenCreateTransaction
 from hiero_sdk_python.tokens.token_delete_transaction import TokenDeleteTransaction
+from hiero_sdk_python.tokens.token_freeze_status import TokenFreezeStatus
 from hiero_sdk_python.tokens.token_freeze_transaction import TokenFreezeTransaction
 from hiero_sdk_python.tokens.token_id import TokenId
+from hiero_sdk_python.tokens.token_info import TokenInfo
+from hiero_sdk_python.tokens.token_kyc_status import TokenKycStatus
 from hiero_sdk_python.tokens.token_mint_transaction import TokenMintTransaction
+from hiero_sdk_python.tokens.token_pause_status import TokenPauseStatus
 from hiero_sdk_python.tokens.token_pause_transaction import TokenPauseTransaction
 from hiero_sdk_python.tokens.token_type import TokenType
 from hiero_sdk_python.transaction.transaction_receipt import TransactionReceipt
@@ -34,6 +40,7 @@
     CreateTokenParams,
     DeleteTokenParams,
     FreezeTokenParams,
+    GetTokenInfoParams,
     MintTokenParams,
     PauseTokenParams,
 )
@@ -42,8 +49,10 @@
     AssociateTokenResponse,
     ClaimTokenResponse,
     CreateTokenResponse,
+    CustomFeeResponse,
     DeleteTokenResponse,
     FreezeTokenResponse,
+    GetTokenInfoResponse,
     MintTokenResponse,
     PauseTokenResponse,
 )
@@ -478,3 +487,157 @@ def claim_token(params: ClaimTokenParams) -> ClaimTokenResponse:
     receipt: TransactionReceipt = response.get_receipt(client, validate_status=True)
 
     return ClaimTokenResponse(status=ResponseCode(receipt.status).name)
+
+
+def _serialize_key(key) -> str | None:
+    """Serialize a key to its DER-encoded hex string representation."""
+    if key is None:
+        return ""
+    return key.to_string_der()
+
+
+def _serialize_custom_fee(fee: CustomFee) -> CustomFeeResponse:
+    """Serialize a CustomFee to a CustomFeeResponse."""
+    response = CustomFeeResponse()
+
+    if fee.fee_collector_account_id is not None:
+        aid = fee.fee_collector_account_id
+        response.feeCollectorAccountId = {
+            "realm": str(aid.realm),
+            "shard": str(aid.shard),
+            "num": str(aid.num),
+        }
+    response.feeCollectorsExempt = fee.all_collectors_are_exempt
+
+    if isinstance(fee, CustomFixedFee):
+        fixed_fee_dict: dict = {"amount": str(fee.amount)}
+        if fee.denominating_token_id is not None:
+            fixed_fee_dict["denominatingTokenId"] = str(fee.denominating_token_id)
+        response.fixedFee = fixed_fee_dict
+    elif isinstance(fee, CustomFractionalFee):
+        response.fractionalFee = {
+            "numerator": str(fee.numerator),
+            "denominator": str(fee.denominator),
+            "minimumAmount": str(fee.min_amount),
+            "maximumAmount": str(fee.max_amount),
+            "assessmentMethod": "inclusive" if fee.assessment_method == FeeAssessmentMethod.INCLUSIVE else "exclusive",
+        }
+    elif isinstance(fee, CustomRoyaltyFee):
+        royalty_fee_dict: dict = {
+            "numerator": str(fee.numerator),
+            "denominator": str(fee.denominator),
+        }
+        if fee.fallback_fee is not None:
+            fallback: dict = {"amount": str(fee.fallback_fee.amount)}
+            if fee.fallback_fee.denominating_token_id is not None:
+                fallback["denominatingTokenId"] = str(fee.fallback_fee.denominating_token_id)
+            royalty_fee_dict["fallbackFee"] = fallback
+        response.royaltyFee = royalty_fee_dict
+
+    return response
+
+
+def _map_pause_status(pause_status: TokenPauseStatus) -> bool | None:
+    """Map TokenPauseStatus enum to TCK boolean representation."""
+    if pause_status == TokenPauseStatus.PAUSED:
+        return True
+    if pause_status == TokenPauseStatus.UNPAUSED:
+        return False
+    return None
+
+
+def _map_token_type(token_type: TokenType | None) -> str | None:
+    """Map TokenType enum to TCK string representation."""
+    if token_type is None:
+        return None
+    mapping = {
+        TokenType.FUNGIBLE_COMMON: "FUNGIBLE_COMMON",
+        TokenType.NON_FUNGIBLE_UNIQUE: "NON_FUNGIBLE_UNIQUE",
+    }
+    return mapping.get(token_type)
+
+
+def _map_supply_type(supply_type: SupplyType | None) -> str | None:
+    """Map SupplyType enum to TCK string representation."""
+    if supply_type is None:
+        return None
+    mapping = {
+        SupplyType.INFINITE: "INFINITE",
+        SupplyType.FINITE: "FINITE",
+    }
+    return mapping.get(supply_type)
+
+
+def _map_freeze_status(freeze_status: TokenFreezeStatus) -> bool | None:
+    """Map TokenFreezeStatus enum to TCK boolean representation."""
+    if freeze_status == TokenFreezeStatus.FROZEN:
+        return True
+    if freeze_status == TokenFreezeStatus.UNFROZEN:
+        return False
+    return None
+
+
+def _map_kyc_status(kyc_status: TokenKycStatus) -> bool | None:
+    """Map TokenKycStatus enum to TCK boolean representation."""
+    if kyc_status == TokenKycStatus.GRANTED:
+        return True
+    if kyc_status == TokenKycStatus.REVOKED:
+        return False
+    return None
+
+
+def _build_token_info_response(info: TokenInfo) -> GetTokenInfoResponse:
+    """Build a GetTokenInfoResponse from a TokenInfo object."""
+    # Serialize custom fees
+    custom_fees = [_serialize_custom_fee(fee) for fee in info.custom_fees] if info.custom_fees else []
+
+    return GetTokenInfoResponse(
+        tokenId=str(info.token_id) if info.token_id else None,
+        name=info.name or "",
+        symbol=info.symbol or "",
+        decimals=info.decimals,
+        totalSupply=str(info.total_supply) if info.total_supply is not None else "0",
+        treasuryAccountId=str(info.treasury) if info.treasury else None,
+        adminKey=_serialize_key(info.admin_key),
+        kycKey=_serialize_key(info.kyc_key),
+        freezeKey=_serialize_key(info.freeze_key),
+        pauseKey=_serialize_key(info.pause_key),
+        wipeKey=_serialize_key(info.wipe_key),
+        supplyKey=_serialize_key(info.supply_key),
+        feeScheduleKey=_serialize_key(info.fee_schedule_key),
+        metadataKey=_serialize_key(info.metadata_key),
+        defaultFreezeStatus=_map_freeze_status(info.default_freeze_status),
+        defaultKycStatus=_map_kyc_status(info.default_kyc_status),
+        pauseStatus=_map_pause_status(info.pause_status),
+        isDeleted=info.is_deleted,
+        autoRenewAccountId=str(info.auto_renew_account) if info.auto_renew_account else None,
+        autoRenewPeriod=str(info.auto_renew_period.seconds) if info.auto_renew_period else None,
+        expirationTime=str(info.expiry.seconds) if info.expiry else None,
+        tokenMemo=info.memo if info.memo is not None else "",
+        customFees=custom_fees,
+        tokenType=_map_token_type(info.token_type),
+        supplyType=_map_supply_type(info.supply_type),
+        maxSupply=str(info.max_supply) if info.max_supply is not None else "0",
+        metadata=info.metadata.hex() if info.metadata else "",
+        ledgerId=info.ledger_id.hex() if info.ledger_id else "",
+    )
+
+
+@rpc_method("getTokenInfo")
+def get_token_info(params: GetTokenInfoParams) -> GetTokenInfoResponse:
+    """Query token info using TCK getTokenInfo parameters."""
+    client = get_client(params.sessionId)
+
+    query = TokenInfoQuery().set_grpc_deadline(DEFAULT_GRPC_TIMEOUT)
+
+    if params.tokenId is not None:
+        query.set_token_id(TokenId.from_string(params.tokenId))
+
+    if params.queryPayment is not None:
+        query.set_query_payment(Hbar.from_tinybars(int(params.queryPayment)))
+
+    if params.maxQueryPayment is not None:
+        query.set_max_query_payment(Hbar.from_tinybars(int(params.maxQueryPayment)))
+
+    info = query.execute(client)
+    return _build_token_info_response(info)

tck/param/token.py

@@ -4,7 +4,7 @@
 
 from dataclasses import dataclass
 
-from tck.param.base import BaseTransactionParams
+from tck.param.base import BaseParams, BaseTransactionParams
 from tck.param.custom_fee import CustomFeeParams
 from tck.util.param_utils import (
     parse_common_transaction_params,
@@ -232,3 +232,22 @@ def parse_json_params(cls, params: dict) -> ClaimTokenParams:
             sessionId=parse_session_id(params),
             commonTransactionParams=parse_common_transaction_params(params),
         )
+
+
+@dataclass
+class GetTokenInfoParams(BaseParams):
+    """Request parameters for the getTokenInfo endpoint."""
+
+    tokenId: str | None = None
+    queryPayment: str | None = None
+    maxQueryPayment: str | None = None
+
+    @classmethod
+    def parse_json_params(cls, params: dict) -> GetTokenInfoParams:
+        """Parse JSON-RPC params into a GetTokenInfoParams instance."""
+        return cls(
+            tokenId=params.get("tokenId"),
+            queryPayment=params.get("queryPayment"),
+            maxQueryPayment=params.get("maxQueryPayment"),
+            sessionId=parse_session_id(params),
+        )

tck/response/token.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from tck.response.base import StatusOnlyResponse
 
@@ -52,3 +52,48 @@ class AirdropTokenResponse(StatusOnlyResponse):
 @dataclass
 class ClaimTokenResponse(StatusOnlyResponse):
     """Response payload for claimToken."""
+
+
+@dataclass
+class CustomFeeResponse:
+    """Nested custom fee details for getTokenInfo."""
+
+    fixedFee: dict | None = None
+    fractionalFee: dict | None = None
+    royaltyFee: dict | None = None
+    feeCollectorAccountId: dict | None = None
+    feeCollectorsExempt: bool | None = None
+
+
+@dataclass
+class GetTokenInfoResponse:
+    """Response payload for getTokenInfo."""
+
+    tokenId: str | None = None
+    name: str | None = None
+    symbol: str | None = None
+    decimals: int | None = None
+    totalSupply: str | None = None
+    treasuryAccountId: str | None = None
+    adminKey: str | None = None
+    kycKey: str | None = None
+    freezeKey: str | None = None
+    pauseKey: str | None = None
+    wipeKey: str | None = None
+    supplyKey: str | None = None
+    feeScheduleKey: str | None = None
+    metadataKey: str | None = None
+    defaultFreezeStatus: bool | None = field(metadata={"nullable": True}, default=None)
+    defaultKycStatus: bool | None = field(metadata={"nullable": True}, default=None)
+    pauseStatus: bool | None = field(metadata={"nullable": True}, default=None)
+    isDeleted: bool | None = None
+    autoRenewAccountId: str | None = None
+    autoRenewPeriod: str | None = None
+    expirationTime: str | None = None
+    tokenMemo: str | None = None
+    customFees: list[CustomFeeResponse] = field(default_factory=list)
+    tokenType: str | None = None
+    supplyType: str | None = None
+    maxSupply: str | None = None
+    metadata: str | None = None
+    ledgerId: str | None = None