HTTP API Client¶

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

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

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

    ret = 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")

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 = self.get_account_asset_balances(addr)
    return balances.get(str(token_id), 0)

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

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

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

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 = 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),
    }

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

    ret = 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)

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:
        return ValueError("either query by id or issuer")
    if asset_id:
        return self.provider.make_request("wallet/getassetissuebyid", {"value": asset_id, "visible": True})
    return self.provider.make_request(
        "wallet/getassetissuebyaccount",
        {"address": keys.to_base58check_address(issuer), "visible": True},
    )

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 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

def get_bandwidth(self, addr: TAddress) -> int:
    """Query the bandwidth of the account"""
    ret = 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")

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 = self.provider.make_request("wallet/getblockbynum", {"num": id_or_num, "visible": visible})
    elif isinstance(id_or_num, (str,)):
        block = self.provider.make_request("wallet/getblockbyid", {"value": id_or_num, "visible": visible})
    elif id_or_num is None:
        block = 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

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 self.provider.make_request(
        "wallet/getcandelegatedmaxsize",
        {"owner_address": keys.to_base58check_address(address), "type": 1 if resource == "ENERGY" else 0, "visible": True},
    )

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

def get_contract(self, addr: TAddress) -> Contract:
    """Get a contract object."""
    addr = keys.to_base58check_address(addr)
    info = 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 Contract(
        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,
    )

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

def get_contract_info(self, addr: TAddress) -> dict:
    """Queries a contract's information from the blockchain"""
    addr = keys.to_base58check_address(addr)
    info = 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

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

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 self.provider.make_request(
        "wallet/getdelegatedresourcev2",
        {
            "fromAddress": keys.to_base58check_address(fromAddr),
            "toAddress": keys.to_base58check_address(toAddr),
            "visible": True,
        },
    )

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

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 = self.provider.make_request("wallet/estimateenergy", params)
    self._handle_api_error(ret)
    return ret["energy_required"]

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

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

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

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

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

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

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

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

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

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

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 = self.provider.make_request("walletsolidity/gettransactioninfobyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

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 = self.provider.make_request("wallet/gettransactionbyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

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 = self.provider.make_request("wallet/gettransactioninfobyid", {"value": txn_id, "visible": True})
    self._handle_api_error(ret)
    if ret:
        return ret
    raise TransactionNotFound

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 = self.provider.make_request("wallet/getexpandedspendingkey", {"value": sk})
    ask = esk["ask"]
    nsk = esk["nsk"]
    ovk = esk["ovk"]
    ak = self.provider.make_request("wallet/getakfromask", {"value": ask})["value"]
    nk = self.provider.make_request("wallet/getnkfromnsk", {"value": nsk})["value"]

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

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

    ret = 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,
    }

def list_assets(self) -> list:
    """List all TRC10 tokens(assets)."""
    ret = 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

def list_nodes(self) -> list:
    """List all nodes that current API node is connected to."""
    # NOTE: visible parameter is ignored
    ret = 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

def list_witnesses(self) -> list:
    """List all witnesses, including SR, SRP, and SRC."""
    # NOTE: visible parameter is ignored
    ret = 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