Skip to content

cosmpy.aerial.client.__init__

Client functionality.

Account Objects

@dataclass
class Account()

Account.

StakingPosition Objects

@dataclass
class StakingPosition()

Staking positions.

UnbondingPositions Objects

@dataclass
class UnbondingPositions()

Unbonding positions.

Validator Objects

@dataclass
class Validator()

Validator.

Coin Objects

@dataclass
class Coin()

Coins.

StakingSummary Objects

@dataclass
class StakingSummary()

Get the staking summary.

total_staked

@property
def total_staked() -> int

Get the total staked amount.

total_rewards

@property
def total_rewards() -> int

Get the total rewards.

total_unbonding

@property
def total_unbonding() -> int

total unbonding.

LedgerClient Objects

class LedgerClient()

Ledger client.

__init__

def __init__(cfg: NetworkConfig,
             query_interval_secs: int = DEFAULT_QUERY_INTERVAL_SECS,
             query_timeout_secs: int = DEFAULT_QUERY_TIMEOUT_SECS)

Init ledger client.

Arguments:

  • cfg: Network configurations
  • query_interval_secs: int. optional interval int seconds
  • query_timeout_secs: int. optional interval int seconds

network_config

@property
def network_config() -> NetworkConfig

Get the network config.

Returns:

network config

gas_strategy

@property
def gas_strategy() -> GasStrategy

Get gas strategy.

Returns:

gas strategy

gas_strategy

@gas_strategy.setter
def gas_strategy(strategy: GasStrategy)

Set gas strategy.

Arguments:

  • strategy: strategy

Raises:

  • RuntimeError: Invalid strategy must implement GasStrategy interface

query_account

def query_account(address: Address) -> Account

Query account.

Arguments:

  • address: address

Raises:

  • RuntimeError: Unexpected account type returned from query

Returns:

account details

query_params

def query_params(subspace: str, key: str) -> Any

Query Prams.

Arguments:

  • subspace: subspace
  • key: key

Returns:

Query params

query_bank_balance

def query_bank_balance(address: Address, denom: Optional[str] = None) -> int

Query bank balance.

Arguments:

  • address: address
  • denom: denom, defaults to None

Returns:

bank balance

query_bank_all_balances

def query_bank_all_balances(address: Address) -> List[Coin]

Query bank all balances.

Arguments:

  • address: address

Returns:

bank all balances

send_tokens

def send_tokens(destination: Address,
                amount: int,
                denom: str,
                sender: Wallet,
                memo: Optional[str] = None,
                gas_limit: Optional[int] = None) -> SubmittedTx

Send tokens.

Arguments:

  • destination: destination address
  • amount: amount
  • denom: denom
  • sender: sender
  • memo: memo, defaults to None
  • gas_limit: gas limit, defaults to None

Returns:

prepare and broadcast the transaction and transaction details

query_validators

def query_validators(
        status: Optional[ValidatorStatus] = None) -> List[Validator]

Query validators.

Arguments:

  • status: validator status, defaults to None

Returns:

List of validators

query_staking_summary

def query_staking_summary(address: Address) -> StakingSummary

Query staking summary.

Arguments:

  • address: address

Returns:

staking summary

delegate_tokens

def delegate_tokens(validator: Address,
                    amount: int,
                    sender: Wallet,
                    memo: Optional[str] = None,
                    gas_limit: Optional[int] = None) -> SubmittedTx

Delegate tokens.

Arguments:

  • validator: validator address
  • amount: amount
  • sender: sender
  • memo: memo, defaults to None
  • gas_limit: gas limit, defaults to None

Returns:

prepare and broadcast the transaction and transaction details

redelegate_tokens

def redelegate_tokens(current_validator: Address,
                      next_validator: Address,
                      amount: int,
                      sender: Wallet,
                      memo: Optional[str] = None,
                      gas_limit: Optional[int] = None) -> SubmittedTx

Redelegate tokens.

Arguments:

  • current_validator: current validator address
  • next_validator: next validator address
  • amount: amount
  • sender: sender
  • memo: memo, defaults to None
  • gas_limit: gas limit, defaults to None

Returns:

prepare and broadcast the transaction and transaction details

undelegate_tokens

def undelegate_tokens(validator: Address,
                      amount: int,
                      sender: Wallet,
                      memo: Optional[str] = None,
                      gas_limit: Optional[int] = None) -> SubmittedTx

Undelegate tokens.

Arguments:

  • validator: validator
  • amount: amount
  • sender: sender
  • memo: memo, defaults to None
  • gas_limit: gas limit, defaults to None

Returns:

prepare and broadcast the transaction and transaction details

claim_rewards

def claim_rewards(validator: Address,
                  sender: Wallet,
                  memo: Optional[str] = None,
                  gas_limit: Optional[int] = None) -> SubmittedTx

claim rewards.

Arguments:

  • validator: validator
  • sender: sender
  • memo: memo, defaults to None
  • gas_limit: gas limit, defaults to None

Returns:

prepare and broadcast the transaction and transaction details

estimate_gas_for_tx

def estimate_gas_for_tx(tx: Transaction) -> int

Estimate gas for transaction.

Arguments:

  • tx: transaction

Returns:

Estimated gas for transaction

estimate_fee_from_gas

def estimate_fee_from_gas(gas_limit: int) -> str

Estimate fee from gas.

Arguments:

  • gas_limit: gas limit

Returns:

Estimated fee for transaction

estimate_gas_and_fee_for_tx

def estimate_gas_and_fee_for_tx(tx: Transaction) -> Tuple[int, str]

Estimate gas and fee for transaction.

Arguments:

  • tx: transaction

Returns:

estimate gas, fee for transaction

wait_for_query_tx

def wait_for_query_tx(tx_hash: str,
                      timeout: Optional[timedelta] = None,
                      poll_period: Optional[timedelta] = None) -> TxResponse

Wait for query transaction.

Arguments:

  • tx_hash: transaction hash
  • timeout: timeout, defaults to None
  • poll_period: poll_period, defaults to None

Raises:

  • QueryTimeoutError: timeout

Returns:

transaction response

query_tx

def query_tx(tx_hash: str) -> TxResponse

query transaction.

Arguments:

  • tx_hash: transaction hash

Raises:

  • NotFoundError: Tx details not found
  • grpc.RpcError: RPC connection issue

Returns:

query response

simulate_tx

def simulate_tx(tx: Transaction) -> int

simulate transaction.

Arguments:

  • tx: transaction

Raises:

  • RuntimeError: Unable to simulate non final transaction

Returns:

gas used in transaction

broadcast_tx

def broadcast_tx(tx: Transaction) -> SubmittedTx

Broadcast transaction.

Arguments:

  • tx: transaction

Returns:

Submitted transaction