topic_create_transaction.py

"""
This module defines the TopicCreateTransaction class for creating topics
on the Hedera Consensus Service (HCS) using the Hiero Python SDK.

It provides methods to set properties such as memo, admin key, submit key,
auto-renew period, and auto-renew account, and to build the protobuf
transaction body for submission to the Hedera network .
"""

from __future__ import annotations

from hiero_sdk_python.account.account_id import AccountId
from hiero_sdk_python.channels import _Channel
from hiero_sdk_python.crypto.key import Key
from hiero_sdk_python.Duration import Duration
from hiero_sdk_python.executable import _Method
from hiero_sdk_python.hapi.services import consensus_create_topic_pb2, transaction_pb2
from hiero_sdk_python.hapi.services.schedulable_transaction_body_pb2 import (
    SchedulableTransactionBody,
)
from hiero_sdk_python.tokens.custom_fixed_fee import CustomFixedFee
from hiero_sdk_python.transaction.transaction import Transaction
from hiero_sdk_python.utils.key_utils import key_to_proto


class TopicCreateTransaction(Transaction):
    """
    Represents a transaction to create a new topic in the Hedera
    Consensus Service (HCS).

    This transaction can optionally define an admin key, submit key,
    auto-renew period, auto-renew account, and memo.
    """

    def __init__(
        self,
        memo: str | None = None,
        admin_key: Key | None = None,
        submit_key: Key | None = None,
        auto_renew_period: Duration | None = None,
        auto_renew_account: AccountId | None = None,
        custom_fees: list[CustomFixedFee] | None = None,
        fee_schedule_key: Key | None = None,
        fee_exempt_keys: list[Key] | None = None,
    ) -> None:
        """
        Initializes a new instance of the TopicCreateTransaction class.

        Args:
            memo (str, optional): Optional memo for the topic.
            admin_key (Key, optional): Optional admin key for the topic (PrivateKey or PublicKey).
            submit_key (Key, optional): Optional submit key for the topic (PrivateKey or PublicKey).
            auto_renew_period (Duration, optional): Optional auto-renew period for the topic.
            auto_renew_account (AccountId, optional): Optional account ID for auto-renewal.
            custom_fees (list[CustomFixedFee]): Optional list of custom fees for the topic.
            fee_schedule_key (Key, optional): Optional fee schedule key for the topic (PrivateKey or PublicKey).
            fee_exempt_keys (list[Key], optional): Optional list of fee exempt keys for the topic (PrivateKey or PublicKey).
        """
        super().__init__()
        self.memo: str = memo or ""
        self.admin_key: Key | None = admin_key
        self.submit_key: Key | None = submit_key
        self.auto_renew_period: Duration = auto_renew_period or Duration(7890000)
        self.auto_renew_account: AccountId | None = auto_renew_account
        self.transaction_fee: int | None = 2_000_000_000  # 20 Hbars
        self.custom_fees: list[CustomFixedFee] = custom_fees or []
        self.fee_exempt_keys: list[Key] = fee_exempt_keys or []
        self.fee_schedule_key: Key | None = fee_schedule_key

    def set_memo(self, memo: str) -> TopicCreateTransaction:
        """
        Sets the memo for the topic creation transaction.

        Args:
            memo (str): The memo to set for the topic.

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.memo = memo
        return self

    def set_admin_key(self, key: Key) -> TopicCreateTransaction:
        """
        Sets the admin key for the topic creation transaction.

        Args:
            key (Key): The admin key to set for the topic (PrivateKey or PublicKey).

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.admin_key = key
        return self

    def set_submit_key(self, key: Key) -> TopicCreateTransaction:
        """
        Sets the submit key for the topic creation transaction.

        Args:
            key (Key): The submit key to set for the topic (PrivateKey or PublicKey).

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.submit_key = key
        return self

    def set_auto_renew_period(self, seconds: Duration | int) -> TopicCreateTransaction:
        """
        Sets the auto-renew period for the topic creation transaction.

        Args:
            seconds (Union[Duration, int]): The auto-renew period in seconds or a Duration object.

        Returns:
            TopicCreateTransaction: The current instance for method chaining.

        Raises:
            TypeError: If the provided duration is of an invalid type.
        """
        self._require_not_frozen()
        if isinstance(seconds, int):
            self.auto_renew_period = Duration(seconds)
        elif isinstance(seconds, Duration):
            self.auto_renew_period = seconds
        else:
            raise TypeError("Duration of invalid type")
        return self

    def set_auto_renew_account(self, account_id: AccountId) -> TopicCreateTransaction:
        """
        Sets the account ID for auto-renewal of the topic.

        Args:
            account_id (AccountId): The account ID to set for auto-renewal.

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.auto_renew_account = account_id
        return self

    def set_custom_fees(self, custom_fees: list[CustomFixedFee]) -> TopicCreateTransaction:
        """
        Sets the custom fees for the topic creation transaction.

        Args:
            custom_fees (list[CustomFixedFee]): The custom fees to set for the topic.

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.custom_fees = custom_fees
        return self

    def set_fee_schedule_key(self, key: Key) -> TopicCreateTransaction:
        """
        Sets the fee schedule key for the topic creation transaction.

        Args:
            key (Key): The fee schedule key to set for the topic (PrivateKey or PublicKey).

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.fee_schedule_key = key
        return self

    def set_fee_exempt_keys(self, keys: list[Key]) -> TopicCreateTransaction:
        """
        Sets the fee exempt keys for the topic creation transaction.

        Args:
            keys (list[Key]): The fee exempt keys to set for the topic (PrivateKey or PublicKey).

        Returns:
            TopicCreateTransaction: The current instance for method chaining.
        """
        self._require_not_frozen()
        self.fee_exempt_keys = keys
        return self

    def _to_proto_key(self, key: Key | None):
        """
        Backwards-compatible wrapper around `key_to_proto` for converting SDK keys
        (PrivateKey or PublicKey) to protobuf `Key` messages.

        This exists so that existing unit tests and any external callers that rely
        on `_to_proto_key` continue to work after centralizing the logic in
        `hiero_sdk_python.utils.key_utils.key_to_proto`.
        """
        return key_to_proto(key)

    def _build_proto_body(self) -> consensus_create_topic_pb2.ConsensusCreateTopicTransactionBody:
        """
        Returns the protobuf body for the topic create transaction.

        Returns:
            ConsensusCreateTopicTransactionBody: The protobuf body for this transaction.
        """
        return consensus_create_topic_pb2.ConsensusCreateTopicTransactionBody(
            adminKey=key_to_proto(self.admin_key),
            submitKey=key_to_proto(self.submit_key),
            autoRenewPeriod=(self.auto_renew_period._to_proto() if self.auto_renew_period is not None else None),
            autoRenewAccount=(self.auto_renew_account._to_proto() if self.auto_renew_account is not None else None),
            memo=self.memo,
            custom_fees=[custom_fee._to_topic_fee_proto() for custom_fee in self.custom_fees],
            fee_schedule_key=key_to_proto(self.fee_schedule_key),
            fee_exempt_key_list=[key_to_proto(key) for key in self.fee_exempt_keys],
        )

    def build_transaction_body(self) -> transaction_pb2.TransactionBody:
        """
        Builds and returns the protobuf transaction body for topic creation.

        Returns:
            TransactionBody: The protobuf transaction body containing the topic creation details.
        """
        consensus_create_body = self._build_proto_body()
        transaction_body = self.build_base_transaction_body()
        transaction_body.consensusCreateTopic.CopyFrom(consensus_create_body)
        return transaction_body

    def build_scheduled_body(self) -> SchedulableTransactionBody:
        """
        Builds the scheduled transaction body for this topic create transaction.

        Returns:
            SchedulableTransactionBody: The built scheduled transaction body.
        """
        consensus_create_body = self._build_proto_body()
        schedulable_body = self.build_base_scheduled_body()
        schedulable_body.consensusCreateTopic.CopyFrom(consensus_create_body)
        return schedulable_body

    def _get_method(self, channel: _Channel) -> _Method:
        """
        Returns the method to be used for executing the transaction.

        Args:
            channel (_Channel): The channel to be used for the transaction.

        Returns:
            _Method: The method for executing the transaction.
        """
        return _Method(transaction_func=channel.topic.createTopic, query_func=None)