Cosmos client

class xchainpy2_cosmos.cosm_client.CosmosGaiaClient(network=NetworkType.MAINNET, phrase: str | None = None, private_key: str | bytes | callable | None = None, fee_bound: FeeBounds | None = None, root_derivation_paths: Dict[NetworkType, str] | None = None, client_urls={NetworkType.MAINNET: 'https://lcd-cosmoshub.keplr.app', NetworkType.STAGENET: 'https://lcd-cosmoshub.keplr.app', NetworkType.TESTNET: 'https://rest.sentry-02.theta-testnet.polypore.xyz'}, chain_ids={NetworkType.MAINNET: 'cosmoshub-4', NetworkType.STAGENET: 'cosmoshub-4', NetworkType.TESTNET: 'theta-testnet-001'}, explorer_providers={NetworkType.MAINNET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.STAGENET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.TESTNET: ('https://explorer.theta-testnet.polypore.xyz/', 'https://explorer.theta-testnet.polypore.xyz/accounts/{address}', 'https://explorer.theta-testnet.polypore.xyz/transactions/{tx_id}')}, wallet_index=0)

Bases: XChainClient

async broadcast_tx(tx_hex: str) → str: Broadcast the transaction to the network. :param tx_hex: The transaction content in hex format. :return: The transaction identifier (or hash).

build_transfer_tx(what: CryptoAmount, recipient: str) → Transaction

async check_balance(address, amount: CryptoAmount)

convert_coin_to_amount(c: Coin)

async fetch_chain_id(server='') → str: Helper to get Cosmos’ chain id :return:

async get_account(address: str | None = None) → Account | None: Get the account information for the given address: number and sequence. It there is no account, it will return None. It will throw an exception for special addresses (like Reserve) :param address: By default, it will return the account of the current wallet. (optional) :return:

get_address() → str: Get the address for the given wallet index. :return: string address

get_amount_string(amount)

async get_balance(address: str = '') → List[CryptoAmount]: Get the balance of a given address. :param address: By default, it will return the balance of the current wallet. (optional) :return:

get_chain_id()

get_client() → LedgerClient: Please use this getter to obtain LedgerClient for specific Cosmos calls like delegation, staking etc. Underlying library is cosmpy (https://github.com/fetchai/cosmpy) :return: LedgerClient

property get_client_urls

get_denom(asset: Asset) → str

async get_fees() → Fees

get_network()

get_private_key() → str: Get the private key for the given wallet index. :return:

get_private_key_cosmos() → PrivateKey

get_public_key() → PublicKey

async get_transaction_data(tx_id: str, our_address: str = '') → XcTx | None: Get the transaction data for the given transaction id. :param tx_id: :return:

async get_transaction_data_cosmos(tx_id: str) → dict: Get the transaction data for the given transaction id. :param tx_id: :return: raw dict object

async get_transaction_data_raw(tx_id: str) → dict

async get_transactions(address: str = '', offset: int = 0, limit: int = 10, start_time: datetime | None = None, end_time: datetime | None = None, asset: Asset | None = None, filter_function: Callable[[object], bool] | None = None, batch_size=10, batch_delay_sec=2) → TxPage

parse_denom_to_asset(denom: str) → Asset

patch_client(user_agent='Mozilla/5.0 (Windows NT 10.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36', identifier9r='xchainpy-client')

property prefix: str

purge_client()

property rest_session: ClientSession

property rpc_url

property sdk_client: LedgerClient

async search_tx(message_action=None, message_sender=None, offset=0, limit=50)

async search_tx_from_rpc(message_action=None, message_sender=None, transfer_sender=None, transfer_recipient=None, page=None, limit=None, tx_min_height=None, tx_max_height=None, rpc_endpoint=None)

property server_url

set_chain_id(chain_id: str): Set/update the current chain id. :param chain_id: :return: None

set_network(network: NetworkType): Set/update the current network. :param network: :return: None

set_phrase(phrase: str, wallet_index: int = 0)

async transfer(what: CryptoAmount, recipient: str, memo: str | None = None, fee_rate: int | None = None, check_balance: bool = True) → str: Transfer coins. :param check_balance: Check balance before transfer. Default is True. :param what: CryptoAmount (amount and asset to transfer) :param recipient: str recipient address :param memo: str :param fee_rate: int :return: str tx hash

url_to_fetch_tx_data(tx_id, server_url=None)

validate_address(address: str) → bool: Validates a Cosmos address. :param address: Address string :return: True if valid, False otherwise.

Cosmos constants

xchainpy2_cosmos.const.get_default_fees() → Fees

xchainpy2_cosmos.const.make_explorer(path, acc_subpath, tx_subpath)

xchainpy2_cosmos.const.make_explorer_mint_scan()

xchainpy2_cosmos.const.make_explorer_std(path)

Cosmos Utils

xchainpy2_cosmos.utils.convert_address_for_msg(address, prefix: str) → bytes

xchainpy2_cosmos.utils.get_asset(denom: str, native_denom='uatom', native_asset=('GAIA', 'ATOM', '', AssetKind.NORMAL)) → Asset | None: Get Asset from denomination - currently ATOM supported only :param denom: :param native_asset: :param native_denom: :return: The asset of the given denomination.

xchainpy2_cosmos.utils.get_coin_amount(coins, decimals) → Amount

xchainpy2_cosmos.utils.get_coins_by_asset(coins, search_asset: Asset, native_denom: str, native_asset: Asset)

xchainpy2_cosmos.utils.get_denom(asset: Asset) → str: Get denomination from Asset - currently ATOM supported only :param asset: :return: The denomination of the given asset.

xchainpy2_cosmos.utils.is_msg_multi_send(msg)

xchainpy2_cosmos.utils.is_msg_send(msg)

xchainpy2_cosmos.utils.parse_cosmos_amount(amount: str) → Tuple[int, str]: Parse COSMOS amount into integer amount and string denomination :param amount: a string with format “123asset”, like “100rune” :return: (int, str)

xchainpy2_cosmos.utils.parse_cosmos_amounts(value: str) → List[Tuple[int, str]]: Parse COSMOS amounts from string like “1234rune,1234bnb” :param value: Cosmos SDK amount string :return: List of (value, asset) pairs

xchainpy2_cosmos.utils.parse_transfer_log(log: TxLog, decimals, filter_address, native_denom: str, native_asset: Asset) → List[TokenTransfer]

xchainpy2_cosmos.utils.parse_tx_response_json(j: dict, tx_id: str, address: str, decimals: int, native_denom: str, native_asset: Asset) → XcTx

Cosmos models

class xchainpy2_cosmos.models.RPCResponse(jsonrpc, id, result)

Bases: NamedTuple

classmethod from_rpc_json(j)

id: int: Alias for field number 1

jsonrpc: str: Alias for field number 0

result: dict: Alias for field number 2

class xchainpy2_cosmos.models.TxEvent(type, attributes)

Bases: NamedTuple

property as_dict

property as_dict_of_list

attributes: List[TxEventAttribute]: Alias for field number 1

find_attr_value_first(key, default=None) → str | None

find_attributes(key)

classmethod from_rpc_json(j)

type: str: Alias for field number 0

class xchainpy2_cosmos.models.TxEventAttribute(key, value)

Bases: NamedTuple

classmethod from_rpc_json(j)

key: str: Alias for field number 0

value: str: Alias for field number 1

class xchainpy2_cosmos.models.TxHistoryResponse(page_number, page_total, limit, total, tx_responses, txs)

Bases: NamedTuple

classmethod from_rpc_json(j)

limit: int: Alias for field number 2

page_number: int: Alias for field number 0

page_total: int: Alias for field number 1

total: int: Alias for field number 3

tx_responses: List[TxResponse]: Alias for field number 4

txs: List[dict]: Alias for field number 5

exception xchainpy2_cosmos.models.TxInternalException: Bases: Exception

exception xchainpy2_cosmos.models.TxLoadException: Bases: Exception

class xchainpy2_cosmos.models.TxLog(msg_index, log, events)

Bases: NamedTuple

events: List[TxEvent]: Alias for field number 2

find_event(event_name)

find_events(event_name) → List[TxEvent]

classmethod from_rpc_json(j)

log: str: Alias for field number 1

msg_index: int: Alias for field number 0

class xchainpy2_cosmos.models.TxResponse(height, txhash, data, raw_log, logs, gas_wanted, gas_used, tx, timestamp)

Bases: NamedTuple

data: str: Alias for field number 2

classmethod from_rpc_json(j)

gas_used: str: Alias for field number 6

gas_wanted: str: Alias for field number 5

height: int: Alias for field number 0

logs: list: Alias for field number 4

raw_log: str: Alias for field number 3

timestamp: str: Alias for field number 8

tx: dict | None: Alias for field number 7

txhash: str: Alias for field number 1

xchainpy2_cosmos.models.load_logs(j_logs)