fix: more cyclo complexity

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

Author: exploreriii

src/hiero_sdk_python/tokens/token_create_transaction.py

@@ -23,6 +23,7 @@
 from hiero_sdk_python.account.account_id import AccountId
 from hiero_sdk_python.crypto.private_key import PrivateKey
 from hiero_sdk_python.tokens.custom_fee import CustomFee
+from hiero_sdk_python.crypto.public_key import PublicKey
 
 DEFAULT_TRANSACTION_FEE = 3_000_000_000
 
@@ -401,29 +402,21 @@ def build_transaction_body(self) -> transaction_body_pb2.TransactionBody:
         # Validate freeze status
         TokenCreateValidator._validate_token_freeze_status(self._keys, self._token_params)
 
-        admin_key_proto:    Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.admin_key)
-        supply_key_proto:   Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.supply_key)
-        freeze_key_proto:   Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.freeze_key)
-        wipe_key_proto:     Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.wipe_key)
-        metadata_key_proto: Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.metadata_key)
-        pause_key_proto:    Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.pause_key)
-        kyc_key_proto:      Optional[basic_types_pb2.Key] = self._to_proto_key(self._keys.kyc_key)
-
-        # Ensure token type is correctly set with default to fungible
-        if self._token_params.token_type is None:
-            token_type_value = 0  # default FUNGIBLE_COMMON
-        elif isinstance(self._token_params.token_type, TokenType):
-            token_type_value = self._token_params.token_type.value
-        else:
-            token_type_value = self._token_params.token_type
-
-        # Ensure supply type is correctly set with default to infinite
-        if self._token_params.supply_type is None:
-            supply_type_value = 0  # default INFINITE
-        elif isinstance(self._token_params.supply_type, SupplyType):
-            supply_type_value = self._token_params.supply_type.value
-        else:
-            supply_type_value = self._token_params.supply_type
+        # --- small helpers to keep this method simple ---
+        def _enum_value(val, enum_cls, default_int: int) -> int:
+            """Return enum int for None/enum/int input, with a default."""
+            if val is None:
+                return default_int
+            if isinstance(val, enum_cls):
+                return val.value
+            return int(val)
+
+        def _key_proto(pk: Optional[PublicKey]) -> Optional[basic_types_pb2.Key]:
+            return self._to_proto_key(pk)
+
+        # Resolve enum-like params with defaults (FUNGIBLE_COMMON=0, INFINITE=0)
+        token_type_value  = _enum_value(self._token_params.token_type,  TokenType,  0)
+        supply_type_value = _enum_value(self._token_params.supply_type, SupplyType, 0)
 
         # Construct the TokenCreateTransactionBody
         token_create_body = token_create_pb2.TokenCreateTransactionBody(
@@ -438,29 +431,27 @@ def build_transaction_body(self) -> transaction_body_pb2.TransactionBody:
             treasury=self._token_params.treasury_account_id._to_proto(),
         )
 
-        # Conditionally attach each optional sub-message
-        if admin_key_proto is not None:
-            token_create_body.adminKey.CopyFrom(admin_key_proto)
-        if supply_key_proto is not None:
-            token_create_body.supplyKey.CopyFrom(supply_key_proto)
-        if freeze_key_proto is not None:
-            token_create_body.freezeKey.CopyFrom(freeze_key_proto)
-        if wipe_key_proto is not None:
-            token_create_body.wipeKey.CopyFrom(wipe_key_proto)
-        if metadata_key_proto is not None:
-            token_create_body.metadata_key.CopyFrom(metadata_key_proto)
-        if pause_key_proto is not None:
-            token_create_body.pause_key.CopyFrom(pause_key_proto)
-        if kyc_key_proto is not None:
-            token_create_body.kycKey.CopyFrom(kyc_key_proto)
+        # Conditionally attach each optional sub-message (data-driven)
+        key_map = (
+            (_key_proto(self._keys.admin_key),    ("adminKey",)),
+            (_key_proto(self._keys.supply_key),   ("supplyKey",)),
+            (_key_proto(self._keys.freeze_key),   ("freezeKey",)),
+            (_key_proto(self._keys.wipe_key),     ("wipeKey",)),
+            (_key_proto(self._keys.metadata_key), ("metadata_key",)),  # snake_case in recent protos
+            (_key_proto(self._keys.pause_key),    ("pause_key",)),     # snake_case
+            (_key_proto(self._keys.kyc_key),      ("kycKey",)),
+        )
+        for key_proto, path in key_map:
+            if key_proto is not None:
+                getattr(token_create_body, path[0]).CopyFrom(key_proto)
+
+        # Custom fees (append if present)
         if self._token_params.custom_fees:
-            for fee in self._token_params.custom_fees:
-                token_create_body.custom_fees.append(fee._to_proto())
+            token_create_body.custom_fees.extend(f._to_proto() for f in self._token_params.custom_fees)
 
         # Build the base transaction body and attach the token creation details
         transaction_body: transaction_body_pb2.TransactionBody = self.build_base_transaction_body()
         transaction_body.tokenCreation.CopyFrom(token_create_body)
-
         return transaction_body
 
     def _get_method(self, channel: _Channel) -> _Method:

src/hiero_sdk_python/tokens/token_info.py

@@ -333,6 +333,52 @@ def _from_proto(cls, proto_obj: hapi_pb.TokenInfo) -> "TokenInfo":
 
         return tokenInfoObject
 
+    # === helpers ===
+    @staticmethod
+    def _copy_key_if_present(dst: "TokenInfo", setter: str, key_msg) -> None:
+        # In proto3, keys are a oneof; check presence via WhichOneof
+        if key_msg is not None and hasattr(key_msg, "WhichOneof") and key_msg.WhichOneof("key"):
+            getattr(dst, setter)(PublicKey._from_proto(key_msg))
+
+    @staticmethod
+    def _parse_custom_fees(proto_obj) -> List[CustomFee]:
+        out: List[CustomFee] = []
+        for fee_proto in getattr(proto_obj, "custom_fees", []):  # snake_case matches your generated proto
+            if fee_proto.HasField("fixed_fee"):
+                out.append(CustomFixedFee._from_proto(fee_proto))
+            elif fee_proto.HasField("fractional_fee"):
+                out.append(CustomFractionalFee._from_proto(fee_proto))
+            elif fee_proto.HasField("royalty_fee"):
+                out.append(CustomRoyaltyFee._from_proto(fee_proto))
+        return out
+
+    @staticmethod
+    def _copy_msg_to_proto(src_obj, dst_proto, src_attr: str, dst_field: str) -> None:
+        """Copy a submessage if present, using _to_proto or _to_protobuf as available."""
+        val = getattr(src_obj, src_attr)
+        if not val:
+            return
+        to_fn = getattr(val, "_to_proto", None) or getattr(val, "_to_protobuf", None)
+        if to_fn is None:
+            raise AttributeError(f"{type(val).__name__} has neither _to_proto nor _to_protobuf")
+        getattr(dst_proto, dst_field).CopyFrom(to_fn())
+
+    @staticmethod
+    def _set_bool(dst_proto, field_name: str, value: Optional[bool], default: bool = False) -> None:
+        """Assign a boolean, using default when value is None."""
+        setattr(dst_proto, field_name, default if value is None else bool(value))
+
+    @staticmethod
+    def _set_enum(dst_proto, field_name: str, enum_val) -> None:
+        """Assign an enum value or 0 if None."""
+        setattr(dst_proto, field_name, (enum_val.value if enum_val is not None else 0))
+
+    @staticmethod
+    def _append_custom_fees(dst_proto, fees: Optional[List[Any]]) -> None:
+        """Append serialized custom fees if any."""
+        if fees:
+            dst_proto.custom_fees.extend(fee._to_proto() for fee in fees)
+
     def _to_proto(self) -> hapi_pb.TokenInfo:
         """
         Converts the TokenInfo instance to a protobuf TokenInfo object.
@@ -352,10 +398,10 @@ def _to_proto(self) -> hapi_pb.TokenInfo:
             metadata=self.metadata,
         )
 
-        for fee in self.custom_fees or []:
-            proto.custom_fees.append(fee._to_proto())
+        # Custom fees
+        self._append_custom_fees(proto, self.custom_fees)
 
-        for src_attr, dst_field in [
+        msg_fields = [
             ("admin_key",          "adminKey"),
             ("kyc_key",            "kycKey"),
             ("freeze_key",         "freezeKey"),
@@ -365,20 +411,16 @@ def _to_proto(self) -> hapi_pb.TokenInfo:
             ("fee_schedule_key",   "fee_schedule_key"),
             ("pause_key",          "pause_key"),
             ("auto_renew_account", "autoRenewAccount"),
-        ]:
+            ("auto_renew_period",  "autoRenewPeriod"),
+            ("expiry",             "expiry"),
+        ]
+        for src_attr, dst_field in msg_fields:
             self._copy_msg_to_proto(self, proto, src_attr, dst_field)
 
-        if self.is_deleted is not None:
-            proto.deleted = self.is_deleted
-
-        proto.defaultFreezeStatus = self.default_freeze_status.value
-        proto.defaultKycStatus   = self.default_kyc_status.value
-        proto.pause_status       = self.pause_status.value
-
-        if self.auto_renew_period:
-            proto.autoRenewPeriod.CopyFrom(self.auto_renew_period._to_proto())
-        if self.expiry:
-            proto.expiry.CopyFrom(self.expiry._to_protobuf())
+        self._set_bool(proto, "deleted", self.is_deleted, default=False)
+        self._set_enum(proto, "defaultFreezeStatus", self.default_freeze_status)
+        self._set_enum(proto, "defaultKycStatus", self.default_kyc_status)
+        self._set_enum(proto, "pause_status", self.pause_status)
 
         return proto
 

tests/unit/test_token_create_transaction.py

@@ -34,6 +34,7 @@
 from hiero_sdk_python.exceptions import PrecheckError
 from hiero_sdk_python.crypto.private_key import PrivateKey
 from hiero_sdk_python.hapi.services import basic_types_pb2
+from hiero_sdk_python.crypto.public_key import PublicKey
 
 pytestmark = pytest.mark.unit