Sending Transaction¶

def account_permission_update(self, owner: TAddress, perm: dict) -> "TransactionBuilder":
    """Update account permission.

    :param owner:
    :param perm: Permission dict from [`get_account_permission`][tronpy.Tron.get_account_permission]
    """

    if "owner" in perm:
        for key in perm["owner"]["keys"]:
            key["address"] = keys.to_hex_address(key["address"])
    if "actives" in perm:
        for act in perm["actives"]:
            for key in act["keys"]:
                key["address"] = keys.to_hex_address(key["address"])
    if perm.get("witness"):
        for key in perm["witness"]["keys"]:
            key["address"] = keys.to_hex_address(key["address"])

    return self._build_transaction(
        "AccountPermissionUpdateContract",
        dict(owner_address=keys.to_hex_address(owner), **perm),
    )

def account_update(self, owner: TAddress, name: str) -> "TransactionBuilder":
    """Update account name. An account can only set name once."""
    return self._build_transaction(
        "UpdateAccountContract",
        {"owner_address": keys.to_hex_address(owner), "account_name": name.encode().hex()},
    )

def asset_issue(
    self,
    owner: TAddress,
    abbr: str,
    total_supply: int,
    *,
    url: str,
    name: str = None,
    description: str = "",
    start_time: int = None,
    end_time: int = None,
    precision: int = 6,
    frozen_supply: list = None,
    trx_num: int = 1,
    num: int = 1,
) -> TransactionBuilder:
    """Issue a TRC10 token.

    Almost all parameters have resonable defaults.
    """
    if name is None:
        name = abbr

    if start_time is None:
        # use default expiration
        start_time = current_timestamp() + 60_000

    if end_time is None:
        # use default expiration
        end_time = current_timestamp() + 60_000 + 1

    if frozen_supply is None:
        frozen_supply = []

    return self._build_transaction(
        "AssetIssueContract",
        {
            "owner_address": keys.to_hex_address(owner),
            "abbr": abbr.encode().hex(),
            "name": name.encode().hex(),
            "total_supply": total_supply,
            "precision": precision,
            "url": url.encode().hex(),
            "description": description.encode().hex(),
            "start_time": start_time,
            "end_time": end_time,
            "frozen_supply": frozen_supply,
            "trx_num": trx_num,
            "num": num,
            "public_free_asset_net_limit": 0,
            "free_asset_net_limit": 0,
        },
    )

def asset_transfer(self, from_: TAddress, to: TAddress, amount: int, token_id: int) -> TransactionBuilder:
    """Transfer TRC10 tokens."""
    return self._build_transaction(
        "TransferAssetContract",
        {
            "owner_address": keys.to_hex_address(from_),
            "to_address": keys.to_hex_address(to),
            "amount": amount,
            "asset_name": str(token_id).encode().hex(),
        },
    )

def create_witness(self, owner: TAddress, url: str) -> "TransactionBuilder":
    """Create a new witness, will consume 1_000 TRX."""
    payload = {"owner_address": keys.to_hex_address(owner), "url": url.encode().hex()}
    return self._build_transaction("WitnessCreateContract", payload)

def delegate_resource(
    self,
    owner: TAddress,
    receiver: TAddress,
    balance: int,
    resource: str = "BANDWIDTH",
    lock: bool = False,
    lock_period: int = None,
) -> "TransactionBuilder":
    """Delegate bandwidth or energy resources to other accounts in Stake2.0.

    :param owner:
    :param receiver:
    :param balance:
    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    :param lock: Optionally lock delegated resources for 3 days.
    :param lock_period: Optionally lock delegated resources for a specific period. Default: 3 days.
    """

    payload = {
        "owner_address": keys.to_hex_address(owner),
        "receiver_address": keys.to_hex_address(receiver),
        "balance": balance,
        "resource": resource,
        "lock": lock,
    }
    if lock_period is not None:
        payload["lock_period"] = lock_period

    return self._build_transaction("DelegateResourceContract", payload)

def deploy_contract(self, owner: TAddress, contract: Contract) -> "TransactionBuilder":
    """Deploy a new contract on chain."""
    contract._client = self.client
    contract.owner_address = owner
    contract.origin_address = owner
    contract.contract_address = None

    return contract.deploy()

def freeze_balance(self, owner: TAddress, amount: int, resource: str = "ENERGY") -> "TransactionBuilder":
    """Freeze balance to get energy or bandwidth, for 3 days.

    :param owner:
    :param amount:
    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    """
    payload = {
        "owner_address": keys.to_hex_address(owner),
        "frozen_balance": amount,
        "resource": resource,
    }
    return self._build_transaction("FreezeBalanceV2Contract", payload)

def transfer(self, from_: TAddress, to: TAddress, amount: int) -> TransactionBuilder:
    """Transfer TRX. ``amount`` in `SUN`."""
    return self._build_transaction(
        "TransferContract",
        {"owner_address": keys.to_hex_address(from_), "to_address": keys.to_hex_address(to), "amount": amount},
    )

def undelegate_resource(
    self, owner: TAddress, receiver: TAddress, balance: int, resource: str = "BANDWIDTH"
) -> "TransactionBuilder":
    """Cancel the delegation of bandwidth or energy resources to other accounts in Stake2.0

    :param owner:
    :param receiver:
    :param balance:
    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    """

    payload = {
        "owner_address": keys.to_hex_address(owner),
        "receiver_address": keys.to_hex_address(receiver),
        "balance": balance,
        "resource": resource,
    }

    return self._build_transaction("UnDelegateResourceContract", payload)

def unfreeze_balance(self, owner: TAddress, resource: str = "ENERGY", *, unfreeze_balance: int) -> "TransactionBuilder":
    """Unfreeze balance to get TRX back.

    :param owner:
    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    :param unfreeze_balance:
    """
    payload = {
        "owner_address": keys.to_hex_address(owner),
        "unfreeze_balance": unfreeze_balance,
        "resource": resource,
    }
    return self._build_transaction("UnfreezeBalanceV2Contract", payload)

def unfreeze_balance_legacy(
    self, owner: TAddress, resource: str = "ENERGY", receiver: TAddress = None
) -> "TransactionBuilder":
    """Unfreeze Stake 1.0 balance to get TRX back.

    :param owner:
    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    :param receiver:
    """
    payload = {
        "owner_address": keys.to_hex_address(owner),
        "resource": resource,
    }
    if receiver is not None:
        payload["receiver_address"] = keys.to_hex_address(receiver)
    return self._build_transaction("UnfreezeBalanceContract", payload)

def vote_witness(self, owner: TAddress, *votes: tuple[TAddress, int]) -> "TransactionBuilder":
    """Vote for witnesses. Empty ``votes`` to clean voted."""
    votes = [{"vote_address": keys.to_hex_address(addr), "vote_count": count} for addr, count in votes]
    payload = {"owner_address": keys.to_hex_address(owner), "votes": votes}
    return self._build_transaction("VoteWitnessContract", payload)

def withdraw_rewards(self, owner: TAddress) -> "TransactionBuilder":
    """Withdraw voting rewards."""
    payload = {"owner_address": keys.to_hex_address(owner)}
    return self._build_transaction("WithdrawBalanceContract", payload)

def withdraw_stake_balance(self, owner: TAddress) -> "TransactionBuilder":
    """Withdraw all stake v2 balance after waiting for 14 days since unfreeze_balance call.

    :param owner:
    """
    payload = {
        "owner_address": keys.to_hex_address(owner),
    }
    return self._build_transaction("WithdrawExpireUnfreezeContract", payload)

def build(self, *, offline: bool = False, ref_block_id: str = None, **kwargs) -> Transaction:
    """Build the transaction."""
    if offline:
        if not ref_block_id:
            raise ValueError("ref_block_id is required when building offline transactions")
        if proto is None:
            raise ImportError(PROTOBUF_NOT_INSTALLED_ERROR_MESSAGE)
    else:
        ref_block_id = self._client.get_latest_solid_block_id()
    self._raw_data["ref_block_bytes"] = ref_block_id[12:16]
    self._raw_data["ref_block_hash"] = ref_block_id[16:32]
    if offline:
        txid = proto.calculate_txid_from_raw_data(self._raw_data)
        return Transaction(
            self._raw_data,
            client=None,
            method=self._method,
            txid=txid,
            permission=None,
        )
    return Transaction(self._raw_data, client=self._client, method=self._method)

def fee_limit(self, value: int) -> "TransactionBuilder":
    """Set fee_limit of the transaction, in `SUN`."""
    self._raw_data["fee_limit"] = value
    return self

def memo(self, memo: str | bytes) -> "TransactionBuilder":
    """Set memo of the transaction."""
    data = memo.encode() if isinstance(memo, (str,)) else memo
    self._raw_data["data"] = data.hex()
    return self

def permission_id(self, perm_id: int) -> "TransactionBuilder":
    """Set permission_id of the transaction."""
    self._raw_data["contract"][0]["Permission_id"] = perm_id
    return self

def with_owner(self, addr: TAddress) -> "TransactionBuilder":
    """Set owner of the transaction."""
    if "owner_address" in self._raw_data["contract"][0]["parameter"]["value"]:
        self._raw_data["contract"][0]["parameter"]["value"]["owner_address"] = keys.to_hex_address(addr)
    else:
        raise TypeError("can not set owner")
    return self

def broadcast(self) -> TransactionRet:
    """Broadcast the transaction to TRON network."""
    return TransactionRet(self._client.broadcast(self), client=self._client, method=self._method)

def set_signature(self, signature: list) -> "Transaction":
    """set transaction signature"""
    self._signature = signature
    return self

def sign(self, priv_key: PrivateKey) -> "Transaction":
    """Sign the transaction with a private key."""

    if not self.txid:
        raise ValueError("txID not calculated")
    if self.is_expired:
        raise ValueError("expired")

    if self._permission is not None:
        addr_of_key = priv_key.public_key.to_hex_address()
        for key in self._permission["keys"]:
            if key["address"] == addr_of_key:
                break
        else:
            raise BadKey(
                "provided private key is not in the permission list",
                f"provided {priv_key.public_key.to_base58check_address()}",
                f"required {self._permission}",
            )
    sig = priv_key.sign_msg_hash(bytes.fromhex(self.txid))
    self._signature.append(sig.hex())
    return self

def update(self):
    """update Transaction, change ref_block and txID, remove all signature"""
    self._raw_data["timestamp"] = current_timestamp()
    self._raw_data["expiration"] = self._raw_data["timestamp"] + 60_000
    ref_block_id = self._client.get_latest_solid_block_id()
    # last 2 byte of block number part
    self._raw_data["ref_block_bytes"] = ref_block_id[12:16]
    # last half part of block hash
    self._raw_data["ref_block_hash"] = ref_block_id[16:32]

    self.txid = ""
    self._permission = None
    self._signature = []
    sign_weight = self._client.get_sign_weight(self)
    if "transaction" not in sign_weight:
        self._client._handle_api_error(sign_weight)
        return  # unreachable
    self.txid = sign_weight["transaction"]["transaction"]["txID"]

    # when account not exist on-chain
    self._permission = sign_weight.get("permission", None)
    # remove all _signature
    self._signature = []

def result(self, timeout=30, interval=1.6, solid=False) -> dict:
    """Wait the contract calling result.

    :returns: Result of contract method
    """
    if self._method is None:
        raise TypeError("Not a smart contract call")

    receipt = self.wait(timeout, interval, solid)

    if receipt.get("result", None) == "FAILED":
        msg = receipt.get("resMessage", receipt["result"])

        if receipt["receipt"]["result"] == "REVERT":
            with contextlib.suppress(Exception):
                result = receipt.get("contractResult", [])
                if result and len(result[0]) > (4 + 32) * 2:
                    error_msg = tron_abi.decode_single("string", bytes.fromhex(result[0])[4 + 32 :])
                    msg = f"{msg}: {error_msg}"
        raise TvmError(msg)

    return self._method.parse_output(receipt["contractResult"][0])

def wait(self, timeout=30, interval=1.6, solid=False) -> dict:
    """Wait the transaction to be on chain.

    :returns: TransactionInfo
    """

    get_transaction_info = self._client.get_transaction_info
    if solid:
        get_transaction_info = self._client.get_solid_transaction_info

    end_time = time.time() + timeout
    while time.time() < end_time:
        try:
            return get_transaction_info(self._txid)
        except TransactionNotFound:
            time.sleep(interval)

    raise TransactionNotFound("timeout and can not find the transaction")