The Async API¶

def generate_address(self, priv_key=None) -> dict:
    """Generate a random address."""
    if priv_key is None:
        priv_key = PrivateKey.random()
    return {
        "base58check_address": priv_key.public_key.to_base58check_address(),
        "hex_address": priv_key.public_key.to_hex_address(),
        "private_key": priv_key.hex(),
        "public_key": priv_key.public_key.hex(),
    }

def generate_address_from_mnemonic(self, mnemonic: str, passphrase: str = "", account_path: str = TRON_DEFAULT_PATH):
    """
    Generate address from a mnemonic.

    :param str mnemonic: space-separated list of BIP39 mnemonic seed words
    :param str passphrase: Optional passphrase used to encrypt the mnemonic
    :param str account_path: Specify an alternate HD path for deriving the seed using
        BIP32 HD wallet key derivation.
    """
    seed = seed_from_mnemonic(mnemonic, passphrase)
    key = key_from_seed(seed, account_path)
    priv_key = PrivateKey(key)
    return {
        "base58check_address": priv_key.public_key.to_base58check_address(),
        "hex_address": priv_key.public_key.to_hex_address(),
        "private_key": priv_key.hex(),
        "public_key": priv_key.public_key.hex(),
    }

def generate_address_with_mnemonic(
    self, passphrase: str = "", num_words: int = 12, language: str = "english", account_path: str = TRON_DEFAULT_PATH
):
    r"""
    Create a new address and related mnemonic.

    Creates a new address, and returns it alongside the mnemonic that can be used to regenerate it using any BIP39-compatible wallet.

    :param str passphrase: Extra passphrase to encrypt the seed phrase
    :param int num_words: Number of words to use with seed phrase. Default is 12 words.
                          Must be one of [12, 15, 18, 21, 24].
    :param str language: Language to use for BIP39 mnemonic seed phrase.
    :param str account_path: Specify an alternate HD path for deriving the seed using
        BIP32 HD wallet key derivation.
    """  # noqa: E501
    mnemonic = generate_mnemonic(num_words, language)
    return self.generate_address_from_mnemonic(mnemonic, passphrase, account_path), mnemonic

async def generate_zkey(self) -> dict:
    """Generate a random shielded address."""
    return await self.provider.make_request("wallet/getnewshieldedaddress")

async def get_account(self, addr: TAddress) -> dict:
    """Get account info from an address."""

    ret = await self.provider.make_request(
        "wallet/getaccount", {"address": keys.to_base58check_address(addr), "visible": True}
    )
    if ret:
        return ret
    raise AddressNotFound("account not found on-chain")

async def get_account_asset_balance(self, addr: TAddress, token_id: int | str) -> int:
    """Get TRC10 token balance of an account. Result is in raw amount."""
    if int(token_id) < 1000000 or int(token_id) > 1999999:
        raise ValueError("invalid token_id range")

    balances = await self.get_account_asset_balances(addr)
    return balances.get(str(token_id), 0)

async def get_account_asset_balances(self, addr: TAddress) -> dict:
    """Get all TRC10 token balances of an account."""
    info = await self.get_account(addr)
    return {p["key"]: p["value"] for p in info.get("assetV2", {}) if p["value"] > 0}

async def get_account_balance(self, addr: TAddress) -> Decimal:
    """Get TRX balance of an account. Result in `TRX`."""

    info = await self.get_account(addr)
    return Decimal(info.get("balance", 0)) / 1_000_000

async def get_account_permission(self, addr: TAddress) -> dict:
    """Get account's permission info from an address. Can be used in `account_permission_update`."""

    addr = keys.to_base58check_address(addr)
    # will check account existence
    info = await self.get_account(addr)
    # For old accounts prior to AccountPermissionUpdate, these fields are not set.
    # So default permission is for backward compatibility.
    default_witness = None
    if info.get("is_witness", None):
        default_witness = {
            "type": "Witness",
            "id": 1,
            "permission_name": "witness",
            "threshold": 1,
            "keys": [{"address": addr, "weight": 1}],
        }
    return {
        "owner": info.get(
            "owner_permission",
            {"permission_name": "owner", "threshold": 1, "keys": [{"address": addr, "weight": 1}]},
        ),
        "actives": info.get(
            "active_permission",
            [
                {
                    "type": "Active",
                    "id": 2,
                    "permission_name": "active",
                    "threshold": 1,
                    "operations": "7fff1fc0033e0100000000000000000000000000000000000000000000000000",
                    "keys": [{"address": addr, "weight": 1}],
                }
            ],
        ),
        "witness": info.get("witness_permission", default_witness),
    }

async def get_account_resource(self, addr: TAddress) -> dict:
    """Get resource info of an account."""

    ret = await self.provider.make_request(
        "wallet/getaccountresource",
        {"address": keys.to_base58check_address(addr), "visible": True},
    )
    if ret:
        return ret
    raise AddressNotFound("account not found on-chain")

def get_address_from_passphrase(self, passphrase: str) -> dict:
    """Get an address from a passphrase, compatiable with `wallet/createaddress`."""
    priv_key = PrivateKey.from_passphrase(passphrase.encode())
    return self.generate_address(priv_key)

async def get_asset(self, asset_id: int = None, issuer: TAddress = None) -> dict:
    """Get TRC10(asset) info by asset's id or issuer."""
    if asset_id and issuer:
        raise ValueError("either query by id or issuer")
    if asset_id:
        return await self.provider.make_request("wallet/getassetissuebyid", {"value": asset_id, "visible": True})
    return await self.provider.make_request(
        "wallet/getassetissuebyaccount",
        {"address": keys.to_base58check_address(issuer), "visible": True},
    )

async def get_asset_from_name(self, name: str) -> dict:
    """Get asset info from its abbr name, might fail if there're duplicates."""
    assets = [asset for asset in await self.list_assets() if asset["abbr"] == name]
    if assets:
        if len(assets) == 1:
            return assets[0]
        raise ValueError("duplicated assets with the same name", [asset["id"] for asset in assets])
    raise AssetNotFound

async def get_bandwidth(self, addr: TAddress) -> int:
    """Query the bandwidth of the account"""

    ret = await self.provider.make_request(
        "wallet/getaccountnet", {"address": keys.to_base58check_address(addr), "visible": True}
    )
    if ret:
        # (freeNetLimit - freeNetUsed) + (NetLimit - NetUsed)
        return ret["freeNetLimit"] - ret.get("freeNetUsed", 0) + ret.get("NetLimit", 0) - ret.get("NetUsed", 0)
    raise AddressNotFound("account not found on-chain")

async def get_block(self, id_or_num: None | str | int = None, *, visible: bool = True) -> dict:
    """Get block from a block id or block number.

    :param id_or_num: Block number, or Block hash(id), or ``None`` (default) to get the latest block.
    :param visible: Use ``visible=False`` to get non-base58check addresses and strings instead of hex strings.
    """

    if isinstance(id_or_num, (int,)):
        block = await self.provider.make_request("wallet/getblockbynum", {"num": id_or_num, "visible": visible})
    elif isinstance(id_or_num, (str,)):
        block = await self.provider.make_request("wallet/getblockbyid", {"value": id_or_num, "visible": visible})
    elif id_or_num is None:
        block = await self.provider.make_request("wallet/getnowblock", {"visible": visible})
    else:
        raise TypeError(f"can not infer type of {id_or_num}")

    if "Error" in (block or {}):
        raise BugInJavaTron(block)
    if block:
        return block
    raise BlockNotFound

async def get_can_delegated_max_size(self, address: TAddress, resource: str = "ENERGY") -> dict:
    """Query the amount of delegatable resources share of the specified resource type for an address

    Args:
        address (TAddress):
        resource (str, optional): Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``. Defaults to "ENERGY".

    Returns:
        dict: Response data
    """
    return await self.provider.make_request(
        "wallet/getcandelegatedmaxsize",
        {"owner_address": keys.to_base58check_address(address), "type": 1 if resource == "ENERGY" else 0, "visible": True},
    )

async def get_chain_parameters(self) -> dict:
    """List all chain parameters, values that can be changed via proposal."""
    params = await self.provider.make_request("wallet/getchainparameters", {"visible": True})
    return params.get("chainParameter", [])

async def get_contract(self, addr: TAddress) -> AsyncContract:
    """Get a contract object."""
    addr = keys.to_base58check_address(addr)
    info = await self.provider.make_request("wallet/getcontract", {"value": addr, "visible": True})

    try:
        self._handle_api_error(info)
    except ApiError as e:
        # your java's null pointer exception sucks
        raise AddressNotFound("contract address not found") from e

    return AsyncContract(
        addr=addr,
        bytecode=info.get("bytecode", ""),
        name=info.get("name", ""),
        abi=info.get("abi", {}).get("entrys", []),
        origin_energy_limit=info.get("origin_energy_limit", 0),
        user_resource_percent=info.get("consume_user_resource_percent", 100),
        origin_address=info.get("origin_address", ""),
        code_hash=info.get("code_hash", ""),
        client=self,
    )

async def get_contract_as_shielded_trc20(self, addr: TAddress) -> ShieldedTRC20:
    """Get a Shielded TRC20 Contract object."""
    contract = await self.get_contract(addr)
    return ShieldedTRC20(contract)

async def get_contract_info(self, addr: TAddress) -> dict:
    """Queries a contract's information from the blockchain"""
    addr = keys.to_base58check_address(addr)
    info = await self.provider.make_request("wallet/getcontractinfo", {"value": addr, "visible": True})

    try:
        self._handle_api_error(info)
    except ApiError as e:
        raise AddressNotFound("contract address not found") from e

    return info

async def get_delegated_resource_account_index_v2(self, addr: TAddress) -> dict:
    """Query the resource delegation index by an account"""
    return await self.provider.make_request(
        "wallet/getdelegatedresourceaccountindexv2",
        {
            "value": keys.to_base58check_address(addr),
            "visible": True,
        },
    )

async def get_delegated_resource_v2(self, fromAddr: TAddress, toAddr: TAddress) -> dict:
    """Query the amount of delegatable resources share of the specified resource type for an address"""
    return await self.provider.make_request(
        "wallet/getdelegatedresourcev2",
        {
            "fromAddress": keys.to_base58check_address(fromAddr),
            "toAddress": keys.to_base58check_address(toAddr),
            "visible": True,
        },
    )

async def get_energy(self, address: str) -> int:
    """Query the energy of the account"""
    account_info = await self.get_account_resource(address)
    energy_limit = account_info.get("EnergyLimit", 0)
    energy_used = account_info.get("EnergyUsed", 0)
    return energy_limit - energy_used

async def get_estimated_energy(
    self,
    owner_address: TAddress,
    contract_address: TAddress,
    function_selector: str,
    parameter: str,
) -> int:
    """Returns an estimated energy of calling a contract from the chain."""
    params = {
        "owner_address": keys.to_base58check_address(owner_address),
        "contract_address": keys.to_base58check_address(contract_address),
        "function_selector": function_selector,
        "parameter": parameter,
        "visible": True,
    }
    ret = await self.provider.make_request("wallet/estimateenergy", params)
    self._handle_api_error(ret)
    return ret["energy_required"]

async def get_latest_block(self) -> dict:
    """Get latest block."""
    return await self.provider.make_request("wallet/getnowblock", {"visible": True})

async def get_latest_block_id(self) -> str:
    """Get latest block id in hex."""

    info = await self.provider.make_request("wallet/getnodeinfo")
    return info["block"].split(",ID:", 1)[-1]

async def get_latest_block_number(self) -> int:
    """Get latest block number. Implemented via `wallet/getnodeinfo`, which is faster than `wallet/getnowblock`."""

    info = await self.provider.make_request("wallet/getnodeinfo")
    return int(info["block"].split(",ID:", 1)[0].replace("Num:", "", 1))

async def get_latest_solid_block_id(self) -> str:
    """Get latest solid block id in hex."""

    try:
        info = await self.provider.make_request("wallet/getnodeinfo")
        return info["solidityBlock"].split(",ID:", 1)[-1]
    except Exception:
        info = await self.get_latest_solid_block()
        return info["blockID"]

async def get_latest_solid_block_number(self) -> int:
    """Get latest solid block number. Implemented via `wallet/getnodeinfo`,
    which is faster than `walletsolidity/getnowblock`."""
    info = await self.provider.make_request("wallet/getnodeinfo")
    return int(info["solidityBlock"].split(",ID:", 1)[0].replace("Num:", "", 1))

async def get_node_info(self) -> dict:
    """Get current API node' info."""

    return await self.provider.make_request("wallet/getnodeinfo", {"visible": True})

async def get_solid_transaction(self, txn_id: str) -> dict:
    """Get transaction from a transaction id, must be in solid block."""

    if len(txn_id) != 64:
        raise BadHash("wrong transaction hash length")

    ret = await self.provider.make_request("walletsolidity/gettransactionbyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

async def get_solid_transaction_info(self, txn_id: str) -> dict:
    """Get transaction receipt info from a transaction id, must be in solid block."""

    if len(txn_id) != 64:
        raise BadHash("wrong transaction hash length")

    ret = await self.provider.make_request("walletsolidity/gettransactioninfobyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

async def get_transaction(self, txn_id: str) -> dict:
    """Get transaction from a transaction id."""

    if len(txn_id) != 64:
        raise BadHash("wrong transaction hash length")

    ret = await self.provider.make_request("wallet/gettransactionbyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

async def get_transaction_info(self, txn_id: str) -> dict:
    """Get transaction receipt info from a transaction id."""

    if len(txn_id) != 64:
        raise BadHash("wrong transaction hash length")

    ret = await self.provider.make_request("wallet/gettransactioninfobyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

async def get_zkey_from_sk(self, sk: str, d: str = None) -> dict:
    """Get the shielded address from sk(spending key) and d(diversifier)."""
    if len(sk) != 64:
        raise BadKey("32 byte sk required")
    if d and len(d) != 22:
        raise BadKey("11 byte d required")

    esk = await self.provider.make_request("wallet/getexpandedspendingkey", {"value": sk})
    ask = esk["ask"]
    nsk = esk["nsk"]
    ovk = esk["ovk"]
    ak = (await self.provider.make_request("wallet/getakfromask", {"value": ask}))["value"]
    nk = (await self.provider.make_request("wallet/getnkfromnsk", {"value": nsk}))["value"]

    ivk = (await self.provider.make_request("wallet/getincomingviewingkey", {"ak": ak, "nk": nk}))["ivk"]

    if d is None:
        d = (await self.provider.make_request("wallet/getdiversifier"))["d"]

    ret = await self.provider.make_request("wallet/getzenpaymentaddress", {"ivk": ivk, "d": d})
    pkD = ret["pkD"]
    payment_address = ret["payment_address"]

    return {
        "sk": sk,
        "ask": ask,
        "nsk": nsk,
        "ovk": ovk,
        "ak": ak,
        "nk": nk,
        "ivk": ivk,
        "d": d,
        "pkD": pkD,
        "payment_address": payment_address,
    }

async def list_assets(self) -> list:
    """List all TRC10 tokens(assets)."""
    ret = await self.provider.make_request("wallet/getassetissuelist", {"visible": True})
    assets = ret["assetIssue"]
    for asset in assets:
        asset["id"] = int(asset["id"])
        asset["owner_address"] = keys.to_base58check_address(asset["owner_address"])
        asset["name"] = bytes.fromhex(asset["name"]).decode()
        if "abbr" in asset:
            asset["abbr"] = bytes.fromhex(asset["abbr"]).decode()
        else:
            asset["abbr"] = ""
        if "description" in asset:
            asset["description"] = bytes.fromhex(asset["description"]).decode("utf8", "replace")
        else:
            asset["description"] = ""
        asset["url"] = bytes.fromhex(asset["url"]).decode()
    return assets

async def list_nodes(self) -> list:
    """List all nodes that current API node is connected to."""
    # NOTE: visible parameter is ignored
    ret = await self.provider.make_request("wallet/listnodes", {"visible": True})
    nodes = ret.get("nodes", [])
    for node in nodes:
        node["address"]["host"] = bytes.fromhex(node["address"]["host"]).decode()
    return nodes

async def list_witnesses(self) -> list:
    """List all witnesses, including SR, SRP, and SRC."""
    # NOTE: visible parameter is ignored
    ret = await self.provider.make_request("wallet/listwitnesses", {"visible": True})
    witnesses = ret.get("witnesses", [])
    for witness in witnesses:
        witness["address"] = keys.to_base58check_address(witness["address"])

    return witnesses

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

    :param owner: Address of 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) -> "AsyncTransactionBuilder":
    """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,
) -> AsyncTransactionBuilder:
    """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) -> AsyncTransactionBuilder:
    """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) -> "AsyncTransactionBuilder":
    """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,
) -> "AsyncTransactionBuilder":
    """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: AsyncContract) -> "AsyncTransactionBuilder":
    """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") -> "AsyncTransactionBuilder":
    """Freeze balance to get energy or bandwidth, for 3 days.

    :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) -> AsyncTransactionBuilder:
    """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"
) -> "AsyncTransactionBuilder":
    """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
) -> "AsyncTransactionBuilder":
    """Unfreeze balance to get TRX back.

    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    """
    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
) -> "AsyncTransactionBuilder":
    """Unfreeze balance to get TRX back.

    :param resource: Resource type, can be ``"ENERGY"`` or ``"BANDWIDTH"``
    """
    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]) -> "AsyncTransactionBuilder":
    """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) -> "AsyncTransactionBuilder":
    """Withdraw voting rewards."""
    payload = {"owner_address": keys.to_hex_address(owner)}
    return self._build_transaction("WithdrawBalanceContract", payload)