AsyncClient

The base async client object used to interact with the blockchain. This is identical to normal sync client, the only difference is that all methods are asynchronous.

class AsyncClient(endpoint: str, local: bool=False)

To initialize an async client, one of the three RPC cluster endpoint URLs must be passed. When running a local cluster, the local argument must be passed as true for the local endpoint to work.

Methods#

• refresh_http
• get_account_info
• get_balance
• get_block
• get_block_height
• get_block_production
• get_block_commitment
• get_blocks
• get_blocks_with_limit
• get_block_time
• get_cluster_nodes
• get_epoch_info
• get_epoch_schedule
• get_fee_for_message
• get_first_available_block
• get_genesis_hash
• get_health
• get_identity
• get_inflation_governor
• get_inflation_rate
• get_inflation_reward
• get_largest_accounts
• get_leader_schedule
• get_max_retransmit_slot
• get_max_shred_insert_slot
• get_minimum_balance_for_rent_exmeption
• get_multiple_accounts
• get_program_accounts
• get_latest_blockhash
• get_recent_performance_samples
• get_supply
• get_token_accounts_by_owner
• get_token_account_balance
• get_transaction
• request_airdrop
• send_transaction
• get_signature_statuses
• get_slot

Attributes#

• endpoint
• http

Methods#

.refresh_http#

Refreshes the HTTP client interacting with the JSON RPC.

async def refresh_http()

.get_account_info#

Returns the account's information through the provided public key argument.

async def get_account_info(public_key: PublicKey | str)

.get_balance#

Returns the account's balance in lamports through the provided public key argument.

async def get_balance(public_key: PublicKey | str)

.get_block#

Returns identity and transaction information about a confirmed block in the ledger by the slot provided.

async def get_block(slot: int)

.get_block_height#

Returns the current block height of the node.

async def get_block_height()

.get_block_production#

Returns recent block production information from the current or previous epoch.

async def get_block_production()

.get_block_commitment#

Returns commitment for particular block identified by slot.

async def get_block_commitment(block: int)

.get_blocks#

Returns a list of confirmed blocks between two slots, end_slot is optional

async def get_blocks(start_slot: int, end_slot: int | None = None)

.get_blocks_with_limit#

Returns a list of confirmed blocks starting from a given slot to a limit.

async def get_blocks_with_limit(start_slot: int, limit)

.get_block_time#

Returns the estimated production time of a block as Unix timestamp.

async def get_block_time(block: int)

.get_cluster_nodes#

Returns information about all the nodes participating in the cluster.

async def get_cluster_nodes()

.get_epoch_info#

Returns information about the current epoch

async def get_epoch_info()

.get_epoch_schedule#

Returns epoch schedule information from this cluster's genesis config.

async def get_epoch_schedule()

.get_fee_for_message#

Retuns the fee which network will charge for a particular message.

async def get_fee_for_message(message: str)

.get_first_available_block#

Returns the slot of the lowest confirmed block that has not been purged from the ledger.

async def get_first_available_block()

.get_genesis_hash#

Returns the genesis hash.

async def get_genesis_hash()

.get_health#

Returns the current health of the node.

async def get_health()

.get_identity#

Returns the identity public key for the current node.

async def get_identity()

.get_inflation_governor#

Returns the current inflation governor.

async def get_inflation_governor()

.get_inflation_rate#

Returns the inflation / staking reward for a list of addresses for an epoch

async def get_inflation_rate()

.get_inflation_reward#

Returns the specific inflation values for the current epoch.

async def get_inflation_reward()

.get_largest_accounts#

Returns the 20 largest accounts, by lamport balance (results may be cached up to two hours). Currently if using get_largest_accounts you will need to nest the method inside of a try/except block and make an exception for a TimeoutError.

async def get_largest_accounts()

.get_leader_schedule#

Returns the leader schedule for an epoch. Lots of data!

async def get_leader_schedule()

.get_max_retransmit_slot#

Get the max slot seen from retransmit stage.

async def get_max_retransmit_slot()

.get_max_shred_insert_slot#

Get the max slot seen from after shred insert.

async def get_max_shred_insert_slot()

.get_minimum_balance_for_rent_exemption#

Returns minimum balance required to make account rent exempt.

async def get_minimum_balance_for_rent_exmeption()

.get_multiple_accounts#

Returns the account information for a list of Pubkeys.

async def get_multiple_accounts()

.get_program_accounts#

Returns all accounts owned by the provided program Pubkey.

async def get_program_accounts()

.get_latest_blockhash#

Returns a recent block hash from the ledger, and a fee schedule that can be used to compute the cost of submitting a transaction using it.

async def get_latest_blockhash()

.get_recent_performance_samples#

Returns a list of recent performance samples, in reverse slot order. Performance samples are taken every 60 seconds and include the number of transactions and slots that occur in a given time window.

async def get_recent_performance_samples()

.get_signatures_for_address#

Returns signatures for confirmed transactions that include the given address in their accountKeys list. Returns signatures backwards in time from the provided signature or most recent confirmed block

async def get_signatures_for_address()

.get_supply#

Returns information about the current supply.

async def get_supply()

.get_token_accounts_by_owner#

Returns all SPL Token accounts of the public key provided. Either mint_id or program_id keyword argument is required, encoding is optional.

async def get_token_accounts_by_owner(public_key: PublicKey | str,  **kwargs)

Example:

import asyncio
from solathon import AsyncClient, PublicKey

def main():
    client = AsyncClient("https://api.mainnet-beta.solana.com")

    public_key = PublicKey("B3BhJ1nvPvEhx3hq3nfK8hx4WYcKZdbhavSobZEA44ai")
    program_id = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" # Token program ID

    tokens = await client.get_token_accounts_by_owner(public_key, program_id=program_id)

    print(tokens)

asyncio.run(main())

.get_token_account_balance#

Returns balance of tokens held in the specified token account.

async def get_token_account_balance(token_account: PublicKey | str, commitment: Optional[Commitment]=None)

Parameters:

• token_account: The public key of the token account
• commitment (optional): The level of commitment desired when querying state

Example:

from solathon import Client, PublicKey

client = Client("https://api.mainnet-beta.solana.com")

token_account = "FhmH5YDUX9gPKUFtNR95sFKsdprz49Ffv1VXjxzPWQti" # Token account public key

tokens = client.get_token_account_balance(token_account)

print(tokens)

.get_transaction#

Returns transaction details for a confirmed transaction signature.

async def get_transaction(signature: Text) -> RPCResponse

Parameters:

• signature: Transaction signature as base-58 encoded string

.request_airdrop#

Requests the amount of lamport specified to be airdropped to the public key.

async def request_airdrop(public_key: PublicKey | lamports: int)

.send_transaction#

Submits a signed Transaction object to the cluster for processing.

async def send_transaction(transaction: Transaction)

Example:

import time
import asyncio
from solathon.core.instructions import transfer
from solathon import AsyncClient, Transaction, PublicKey, Keypair

def main():
    client = AsyncClient("https://api.devnet.solana.com")

    sender = Keypair()
    receiver = PublicKey("B3BhJ1nvPvEhx3hq3nfK8hx4WYcKZdbhavSobZEA44a")
    amount = 10000  # This is the amount in lamports

    # Requesting airdrop to get some devnet tokens for the transaction to succeed
    res = await client.request_airdrop(sender.public_key, amount + amount)
    print("Airdrop response: ", res)
    # 20 second delay to make sure we get enough time to receive airdrop tokens
    time.sleep(20)

    # Creating transfer instruction
    instruction = transfer(
        from_public_key=sender.public_key,
        to_public_key=receiver, 
        lamports=100
    )
    transaction = Transaction(instructions=[instruction], signers=[sender])
    result = await client.send_transaction(transaction)

    print("Transaction response: ", result)

asyncio.run(main())

.get_signature_statuses#

Returns signature statuses for confirmed transactions that include the given address in their accountKeys list. Returns signatures backwards in time from the provided signature or most recent confirmed block

async def get_signature_statuses()

.get_slot#

Returns the current slot of the node.

async def get_slot()

.build_and_send_request_async#

Internal method used to construct and send RPC requests to the Solana network.

async def build_and_send_request_async(method: Text, params: List[Any]) -> RPCResponse

Parameters:

• method: The RPC method name to call
• params: List of parameters to pass to the RPC method

Attributes#

.endpoint#

Returns the endpoint which the client is currently using.

.http#

Returns the http object being used to interact with the JSON RPC.