Async API Client
solana.rpc.async_api
Async API client to interact with the Solana JSON RPC Endpoint.
AsyncClient
Async client class.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
endpoint |
Optional[str] |
URL of the RPC endpoint. |
None |
commitment |
Optional[Commitment] |
Default bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
timeout |
float |
HTTP request timeout in seconds. |
10 |
extra_headers |
Optional[Dict[str, str]] |
Extra headers to pass for HTTP request. |
None |
Source code in solana/rpc/async_api.py
class AsyncClient(_ClientCore): # pylint: disable=too-many-public-methods
"""Async client class.
Args:
endpoint: URL of the RPC endpoint.
commitment: Default bank state to query. It can be either "finalized", "confirmed" or "processed".
timeout: HTTP request timeout in seconds.
extra_headers: Extra headers to pass for HTTP request.
"""
def __init__(
self,
endpoint: Optional[str] = None,
commitment: Optional[Commitment] = None,
timeout: float = 10,
extra_headers: Optional[Dict[str, str]] = None,
) -> None:
"""Init API client."""
super().__init__(commitment)
self._provider = async_http.AsyncHTTPProvider(endpoint, timeout=timeout, extra_headers=extra_headers)
async def __aenter__(self) -> "AsyncClient":
"""Use as a context manager."""
await self._provider.__aenter__()
return self
async def __aexit__(self, _exc_type, _exc, _tb):
"""Exits the context manager."""
await self.close()
async def close(self) -> None:
"""Use this when you are done with the client."""
await self._provider.close()
async def is_connected(self) -> bool:
"""Health check.
>>> solana_client = AsyncClient("http://localhost:8899")
>>> asyncio.run(solana_client.is_connected()) # doctest: +SKIP
True
Returns:
True if the client is connected.
"""
body = self._get_health_body()
response = await self._provider.make_request(body, GetHealthResp)
return response.value == "ok"
async def get_balance(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetBalanceResp:
"""Returns the balance of the account of provided Pubkey.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_balance(Pubkey([0] * 31 + [1]))).value # doctest: +SKIP
0
"""
body = self._get_balance_body(pubkey, commitment)
return await self._provider.make_request(body, GetBalanceResp)
async def get_account_info(
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
encoding: str = "base64",
data_slice: Optional[types.DataSliceOpts] = None,
) -> GetAccountInfoResp:
"""Returns all the account info for the specified public key.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for Account data, either "base58" (slow), "base64", or
"jsonParsed". Default is "base64".
- "base58" is limited to Account data of less than 128 bytes.
- "base64" will return base64 encoded data for Account data of any size.
- "jsonParsed" encoding attempts to use program-specific state parsers to return more human-readable and explicit account state data.
If jsonParsed is requested but a parser cannot be found, the field falls back to base64 encoding,
detectable when the data field is type. (jsonParsed encoding is UNSTABLE).
data_slice: (optional) Option to limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info(Pubkey([0] * 31 + [1]))).value # doctest: +SKIP
Account(
Account {
lamports: 4104230290,
data.len: 0,
owner: 11111111111111111111111111111111,
executable: false,
rent_epoch: 371,
},
)
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_account_info_body(
pubkey=pubkey,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
)
return await self._provider.make_request(body, GetAccountInfoResp)
async def get_account_info_json_parsed(
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
) -> GetAccountInfoMaybeJsonParsedResp:
"""Returns all the account info for the specified public key.
If JSON formatting is not available for this account, base64 is returned.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info_json_parsed(Pubkey([0] * 31 + [1]))).value.owner # doctest: +SKIP
Pubkey(
11111111111111111111111111111111,
)
"""
body = self._get_account_info_body(pubkey=pubkey, commitment=commitment, encoding="jsonParsed", data_slice=None)
return await self._provider.make_request(body, GetAccountInfoMaybeJsonParsedResp)
async def get_block_commitment(self, slot: int) -> GetBlockCommitmentResp:
"""Fetch the commitment for particular block.
Args:
slot: Block, identified by Slot.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_commitment(0)).total_stake # doctest: +SKIP
497717120
"""
body = self._get_block_commitment_body(slot)
return await self._provider.make_request(body, GetBlockCommitmentResp)
async def get_block_time(self, slot: int) -> GetBlockTimeResp:
"""Fetch the estimated production time of a block.
Args:
slot: Block, identified by Slot.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_time(5)).value # doctest: +SKIP
1598400007
"""
body = self._get_block_time_body(slot)
return await self._provider.make_request(body, GetBlockTimeResp)
async def get_cluster_nodes(self) -> GetClusterNodesResp:
"""Returns information about all the nodes participating in the cluster.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_cluster_nodes()).value[0].tpu # doctest: +SKIP
'139.178.65.155:8004'
"""
return await self._provider.make_request(self._get_cluster_nodes, GetClusterNodesResp)
async def get_block(
self,
slot: int,
encoding: str = "json",
max_supported_transaction_version: Union[int, None] = None,
) -> GetBlockResp:
"""Returns identity and transaction information about a confirmed block in the ledger.
Args:
slot: Slot, as u64 integer.
encoding: (optional) Encoding for the returned Transaction, either "json", "jsonParsed",
"base58" (slow), or "base64". If parameter not provided, the default encoding is JSON.
max_supported_transaction_version: (optional) The max transaction version to return in
responses. If the requested transaction is a higher version, an error will be returned
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block(1)).value.blockhash # doctest: +SKIP
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
"""
body = self._get_block_body(slot, encoding, max_supported_transaction_version)
return await self._provider.make_request(body, GetBlockResp)
async def get_recent_performance_samples(self, limit: Optional[int] = None) -> GetRecentPerformanceSamplesResp:
"""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.
Args:
limit: Limit (optional) number of samples to return (maximum 720)
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_recent_performance_samples(1)).value[0] # doctest: +SKIP
RpcPerfSample(
RpcPerfSample {
slot: 168036172,
num_transactions: 7159,
num_slots: 158,
sample_period_secs: 60,
},
)
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_recent_performance_samples_body(limit)
return await self._provider.make_request(body, GetRecentPerformanceSamplesResp)
async def get_block_height(self, commitment: Optional[Commitment] = None) -> GetBlockHeightResp:
"""Returns the current block height of the node.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_height()).value # doctest: +SKIP
1233
"""
body = self._get_block_height_body(commitment)
return await self._provider.make_request(body, GetBlockHeightResp)
async def get_blocks(self, start_slot: int, end_slot: Optional[int] = None) -> GetBlocksResp:
"""Returns a list of confirmed blocks.
Args:
start_slot: Start slot, as u64 integer.
end_slot: (optional) End slot, as u64 integer.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_blocks(5, 10)).value # doctest: +SKIP
[5, 6, 7, 8, 9, 10]
"""
body = self._get_blocks_body(start_slot, end_slot)
return await self._provider.make_request(body, GetBlocksResp)
async def get_signatures_for_address(
self,
account: Pubkey,
before: Optional[Signature] = None,
until: Optional[Signature] = None,
limit: Optional[int] = None,
commitment: Optional[Commitment] = None,
) -> GetSignaturesForAddressResp:
"""Returns confirmed signatures for transactions involving an address.
Signatures are returned backwards in time from the provided signature or
most recent confirmed block.
Args:
account: Account to be queried.
before: (optional) Start searching backwards from this transaction signature.
If not provided the search starts from the top of the highest max confirmed block.
until: (optional) Search until this transaction signature, if found before limit reached.
limit: (optional) Maximum transaction signatures to return (between 1 and 1,000, default: 1,000).
commitment: (optional) Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.pubkey import Pubkey
>>> pubkey = Pubkey.from_string("Vote111111111111111111111111111111111111111")
>>> (await solana_client.get_signatures_for_address(pubkey, limit=1)).value[0].signature # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
body = self._get_signatures_for_address_body(account, before, until, limit, commitment)
return await self._provider.make_request(body, GetSignaturesForAddressResp)
async def get_transaction(
self,
tx_sig: Signature,
encoding: str = "json",
commitment: Optional[Commitment] = None,
max_supported_transaction_version: Optional[int] = None,
) -> GetTransactionResp:
"""Returns transaction details for a confirmed transaction.
Args:
tx_sig: Transaction signature as base-58 encoded string N encoding attempts to use program-specific
instruction parsers to return more human-readable and explicit data in the
`transaction.message.instructions` list.
encoding: (optional) Encoding for the returned Transaction, either "json", "jsonParsed",
"base58" (slow), or "base64". If parameter not provided, the default encoding is JSON.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
max_supported_transaction_version: (optional) The max transaction version to return in responses.
If the requested transaction is a higher version, an error will be returned
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.signature import Signature
>>> sig = Signature.from_string("3PtGYH77LhhQqTXP4SmDVJ85hmDieWsgXCUbn14v7gYyVYPjZzygUQhTk3bSTYnfA48vCM1rmWY7zWL3j1EVKmEy")
>>> (await solana_client.get_transaction(sig)).value.block_time # doctest: +SKIP
1234
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_transaction_body(tx_sig, encoding, commitment, max_supported_transaction_version)
return await self._provider.make_request(body, GetTransactionResp)
async def get_epoch_info(self, commitment: Optional[Commitment] = None) -> GetEpochInfoResp:
"""Returns information about the current epoch.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_info()).value.epoch # doctest: +SKIP
0
"""
body = self._get_epoch_info_body(commitment)
return await self._provider.make_request(body, GetEpochInfoResp)
async def get_epoch_schedule(self) -> GetEpochScheduleResp:
"""Returns epoch schedule information from this cluster's genesis config.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_schedule()).value.slots_per_epoch # doctest: +SKIP
8192
"""
return await self._provider.make_request(self._get_epoch_schedule, GetEpochScheduleResp)
async def get_fee_for_message(
self, message: VersionedMessage, commitment: Optional[Commitment] = None
) -> GetFeeForMessageResp:
"""Returns the fee for a message.
Args:
message: Message that the fee is requested for.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> msg = Message([transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))])
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_fee_for_message(msg)).value # doctest: +SKIP
5000
"""
body = self._get_fee_for_message_body(message, commitment)
return await self._provider.make_request(body, GetFeeForMessageResp)
async def get_first_available_block(self) -> GetFirstAvailableBlockResp:
"""Returns the slot of the lowest confirmed block that has not been purged from the ledger.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_first_available_block()).value # doctest: +SKIP
1
"""
return await self._provider.make_request(self._get_first_available_block, GetFirstAvailableBlockResp)
async def get_genesis_hash(self) -> GetGenesisHashResp:
"""Returns the genesis hash.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_genesis_hash()).value # doctest: +SKIP
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
"""
return await self._provider.make_request(self._get_genesis_hash, GetGenesisHashResp)
async def get_identity(self) -> GetIdentityResp:
"""Returns the identity pubkey for the current node.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_identity()).value.identity # doctest: +SKIP
Pubkey(
2LVtX3Wq5bhqAYYaUYBRknWaYrsfYiXLQBHTxtHWD2mv,
)
"""
return await self._provider.make_request(self._get_identity, GetIdentityResp)
async def get_inflation_governor(self, commitment: Optional[Commitment] = None) -> GetInflationGovernorResp:
"""Returns the current inflation governor.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> await (solana_client.get_inflation_governor()).value.foundation # doctest: +SKIP
0.05
"""
body = self._get_inflation_governor_body(commitment)
return await self._provider.make_request(body, GetInflationGovernorResp)
async def get_inflation_rate(self) -> GetInflationRateResp:
"""Returns the specific inflation values for the current epoch.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_rate()).value.epoch # doctest: +SKIP
1
"""
return await self._provider.make_request(self._get_inflation_rate, GetInflationRateResp)
async def get_inflation_reward(
self, pubkeys: List[Pubkey], epoch: Optional[int] = None, commitment: Optional[Commitment] = None
) -> GetInflationRewardResp:
"""Returns the inflation / staking reward for a list of addresses for an epoch.
Args:
pubkeys: An array of addresses to query, as base-58 encoded strings
epoch: (optional) An epoch for which the reward occurs. If omitted, the previous epoch will be used
commitment: Bank state to query. It can be either "finalized" or "confirmed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_reward()).value.amount # doctest: +SKIP
2500
"""
body = self._get_inflation_reward_body(pubkeys, epoch, commitment)
return await self._provider.make_request(body, GetInflationRewardResp)
async def get_largest_accounts(
self, filter_opt: Optional[str] = None, commitment: Optional[Commitment] = None
) -> GetLargestAccountsResp:
"""Returns the 20 largest accounts, by lamport balance.
Args:
filter_opt: Filter results by account type; currently supported: circulating|nonCirculating.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_largest_accounts()).value[0].lamports # doctest: +SKIP
500000000000000000
"""
body = self._get_largest_accounts_body(filter_opt, commitment)
return await self._provider.make_request(body, GetLargestAccountsResp)
async def get_leader_schedule(
self, epoch: Optional[int] = None, commitment: Optional[Commitment] = None
) -> GetLeaderScheduleResp:
"""Returns the leader schedule for an epoch.
Args:
epoch: Fetch the leader schedule for the epoch that corresponds to the provided slot.
If unspecified, the leader schedule for the current epoch is fetched.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> resp = await solana_client.get_leader_schedule() # doctest: +SKIP
>>> list(resp.value.items())[0] # doctest: +SKIP
(Pubkey(
HMU77m6WSL9Xew9YvVCgz1hLuhzamz74eD9avi4XPdr,
), [346448, 346449, 346450, 346451, 369140, 369141, 369142, 369143, 384204, 384205, 384206, 384207])
"""
body = self._get_leader_schedule_body(epoch, commitment)
return await self._provider.make_request(body, GetLeaderScheduleResp)
async def get_minimum_balance_for_rent_exemption(
self, usize: int, commitment: Optional[Commitment] = None
) -> GetMinimumBalanceForRentExemptionResp:
"""Returns minimum balance required to make account rent exempt.
Args:
usize: Account data length.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_balance_for_rent_exemption(50)).value # doctest: +SKIP
1238880
"""
body = self._get_minimum_balance_for_rent_exemption_body(usize, commitment)
return await self._provider.make_request(body, GetMinimumBalanceForRentExemptionResp)
async def get_multiple_accounts(
self,
pubkeys: List[Pubkey],
commitment: Optional[Commitment] = None,
encoding: str = "base64",
data_slice: Optional[types.DataSliceOpts] = None,
) -> GetMultipleAccountsResp:
"""Returns all the account info for a list of public keys.
Args:
pubkeys: list of Pubkeys to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for Account data, either "base58" (slow) or "base64".
- "base58" is limited to Account data of less than 128 bytes.
- "base64" will return base64 encoded data for Account data of any size.
data_slice: (optional) Option to limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> (await solana_client.get_multiple_accounts(pubkeys)).value[0].lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_multiple_accounts_body(
pubkeys=pubkeys,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
)
return await self._provider.make_request(body, GetMultipleAccountsResp)
async def get_multiple_accounts_json_parsed(
self,
pubkeys: List[Pubkey],
commitment: Optional[Commitment] = None,
) -> GetMultipleAccountsMaybeJsonParsedResp:
"""Returns all the account info for a list of public keys.
Args:
pubkeys: list of Pubkeys to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> asyncio.run(solana_client.get_multiple_accounts(pubkeys)).value[0].lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_multiple_accounts_body(
pubkeys=pubkeys,
commitment=commitment,
encoding="jsonParsed",
data_slice=None,
)
return await self._provider.make_request(body, GetMultipleAccountsMaybeJsonParsedResp)
async def get_program_accounts( # pylint: disable=too-many-arguments
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
encoding: Optional[str] = None,
data_slice: Optional[types.DataSliceOpts] = None,
filters: Optional[Sequence[Union[int, types.MemcmpOpts]]] = None,
) -> GetProgramAccountsResp:
"""Returns all accounts owned by the provided program Pubkey.
Args:
pubkey: Pubkey of program
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for the returned Transaction, either jsonParsed",
"base58" (slow), or "base64".
data_slice: (optional) Limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
filters: (optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a `dataSize` filter.
Example:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_program_accounts_body(
pubkey=pubkey,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
filters=filters,
)
return await self._provider.make_request(body, GetProgramAccountsResp)
async def get_program_accounts_json_parsed( # pylint: disable=too-many-arguments
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
filters: Optional[Sequence[Union[int, types.MemcmpOpts]]] = None,
) -> GetProgramAccountsMaybeJsonParsedResp:
"""Returns all accounts owned by the provided program Pubkey.
Args:
pubkey: Pubkey of program
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
filters: (optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a `dataSize` filter.
Example:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_program_accounts_body(
pubkey=pubkey,
commitment=commitment,
encoding="jsonParsed",
data_slice=None,
filters=filters,
)
return await self._provider.make_request(body, GetProgramAccountsMaybeJsonParsedResp)
async def get_latest_blockhash(self, commitment: Optional[Commitment] = None) -> GetLatestBlockhashResp:
"""Returns the latest block hash from the ledger.
Response also includes the last valid block height.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_latest_blockhash()).value # doctest: +SKIP
RpcBlockhash {
blockhash: Hash(
4TLzN2RAACFnd5TYpHcUi76pC3V1qkggRF29HWk2VLeT,
),
last_valid_block_height: 158286487,
}
"""
body = self._get_latest_blockhash_body(commitment)
return await self._provider.make_request(body, GetLatestBlockhashResp)
async def get_signature_statuses(
self, signatures: List[Signature], search_transaction_history: bool = False
) -> GetSignatureStatusesResp:
"""Returns the statuses of a list of signatures.
Unless the `search_transaction_history` configuration parameter is included, this method only
searches the recent status cache of signatures, which retains statuses for all active slots plus
`MAX_RECENT_BLOCKHASHES` rooted slots.
Args:
signatures: An array of transaction signatures to confirm.
search_transaction_history: If true, a Solana node will search its ledger cache for
any signatures not found in the recent status cache.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> raw_sigs = [
... "5VERv8NMvzbJMEkV8xnrLkEaWRtSz9CosKDYjCJjBRnbJLgp8uirBgmQpjKhoR4tjF3ZpRzrFmBV6UjKdiSZkQUW",
... "5j7s6NiJS3JAkvgkoc18WVAsiSaci2pxB2A6ueCJP4tprA2TFg9wSyTLeYouxPBJEMzJinENTkpA52YStRW5Dia7"]
>>> sigs = [Signature.from_string(sig) for sig in raw_sigs]
>>> (await solana_client.get_signature_statuses(sigs)).value[0].confirmations # doctest: +SKIP
10
"""
body = self._get_signature_statuses_body(signatures, search_transaction_history)
return await self._provider.make_request(body, GetSignatureStatusesResp)
async def get_slot(self, commitment: Optional[Commitment] = None) -> GetSlotResp:
"""Returns the current slot the node is processing.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot()).value # doctest: +SKIP
7515
"""
body = self._get_slot_body(commitment)
return await self._provider.make_request(body, GetSlotResp)
async def get_slot_leader(self, commitment: Optional[Commitment] = None) -> GetSlotLeaderResp:
"""Returns the current slot leader.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot_leader()).value # doctest: +SKIP
Pubkey(
dv2eQHeP4RFrJZ6UeiZWoc3XTtmtZCUKxxCApCDcRNV,
)
"""
body = self._get_slot_leader_body(commitment)
return await self._provider.make_request(body, GetSlotLeaderResp)
async def get_stake_activation(
self,
pubkey: Pubkey,
epoch: Optional[int] = None,
commitment: Optional[Commitment] = None,
) -> GetStakeActivationResp:
"""Returns epoch activation information for a stake account.
Args:
pubkey: Pubkey of stake account to query
epoch: (optional) Epoch for which to calculate activation details. If parameter not provided,
defaults to current epoch.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_stake_activation()).value.active # doctest: +SKIP
124429280
"""
warn("get_stake_activation is deprecated. Use get_account_info instead.", DeprecationWarning, stacklevel=1)
body = self._get_stake_activation_body(pubkey, epoch, commitment)
return await self._provider.make_request(body, GetStakeActivationResp)
async def get_supply(self, commitment: Optional[Commitment] = None) -> GetSupplyResp:
"""Returns information about the current supply.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_supply()).value.circulating # doctest: +SKIP
683635192454157660
"""
body = self._get_supply_body(commitment)
return await self._provider.make_request(body, GetSupplyResp)
async def get_token_account_balance(
self, pubkey: Pubkey, commitment: Optional[Commitment] = None
) -> GetTokenAccountBalanceResp:
"""Returns the token balance of an SPL Token account (UNSTABLE).
Args:
pubkey: Pubkey of Token account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkey = Pubkey.from_string("7fUAJdStEuGbc3sM84cKRL6yYaaSstyLSU4ve5oovLS7")
>>> (await solana_client.get_token_account_balance(pubkey)).value.amount # noqa: E501 # doctest: +SKIP
'9864'
"""
body = self._get_token_account_balance_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenAccountBalanceResp)
async def get_token_accounts_by_delegate(
self,
delegate: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByDelegateResp:
"""Returns all SPL Token accounts by approved Delegate (UNSTABLE).
Args:
delegate: Public key of the delegate owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_delegate_body(delegate, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByDelegateResp)
async def get_token_accounts_by_delegate_json_parsed(
self,
delegate: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByDelegateJsonParsedResp:
"""Returns all SPL Token accounts by approved delegate in JSON format (UNSTABLE).
Args:
delegate: Public key of the delegate owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_delegate_json_parsed_body(delegate, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByDelegateJsonParsedResp)
async def get_token_accounts_by_owner_json_parsed(
self,
owner: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByOwnerJsonParsedResp:
"""Returns all SPL Token accounts by token owner in JSON format (UNSTABLE).
Args:
owner: Public key of the account owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_owner_json_parsed_body(owner, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByOwnerJsonParsedResp)
async def get_token_accounts_by_owner(
self,
owner: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByOwnerResp:
"""Returns all SPL Token accounts by token owner (UNSTABLE).
Args:
owner: Public key of the account owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_owner_body(owner, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByOwnerResp)
async def get_token_largest_accounts(
self, pubkey: Pubkey, commitment: Optional[Commitment] = None
) -> GetTokenLargestAccountsResp:
"""Returns the 20 largest accounts of a particular SPL Token type."""
body = self._get_token_largest_accounts_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenLargestAccountsResp)
async def get_token_supply(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetTokenSupplyResp:
"""Returns the total supply of an SPL Token type."""
body = self._get_token_supply_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenSupplyResp)
async def get_transaction_count(self, commitment: Optional[Commitment] = None) -> GetTransactionCountResp:
"""Returns the current Transaction count from the ledger.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_transaction_count()).value # doctest: +SKIP
4554
"""
body = self._get_transaction_count_body(commitment)
return await self._provider.make_request(body, GetTransactionCountResp)
async def get_minimum_ledger_slot(self) -> MinimumLedgerSlotResp:
"""Returns the lowest slot that the node has information about in its ledger.
This value may increase over time if the node is configured to purge older ledger data.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_ledger_slot()).value # doctest: +SKIP
1234
"""
return await self._provider.make_request(self._minimum_ledger_slot, MinimumLedgerSlotResp)
async def get_version(self) -> GetVersionResp:
"""Returns the current solana versions running on the node.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_version()).value.solana_core # doctest: +SKIP
'1.13.2'
"""
return await self._provider.make_request(self._get_version, GetVersionResp)
async def get_vote_accounts(self, commitment: Optional[Commitment] = None) -> GetVoteAccountsResp:
"""Returns the account info and associated stake for all the voting accounts in the current bank.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_vote_accounts()).value.current[0].commission # doctest: +SKIP
100
"""
body = self._get_vote_accounts_body(commitment)
return await self._provider.make_request(body, GetVoteAccountsResp)
async def request_airdrop(
self, pubkey: Pubkey, lamports: int, commitment: Optional[Commitment] = None
) -> RequestAirdropResp:
"""Requests an airdrop of lamports to a Pubkey.
Args:
pubkey: Pubkey of account to receive lamports, as base-58 encoded string or public key object.
lamports: Amount of lamports.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.request_airdrop(Pubkey([0] * 31 + [1]), 10000)).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
body = self._request_airdrop_body(pubkey, lamports, commitment)
return await self._provider.make_request(body, RequestAirdropResp)
async def send_raw_transaction(self, txn: bytes, opts: Optional[types.TxOpts] = None) -> SendTransactionResp:
"""Send a transaction that has already been signed and serialized into the wire format.
Args:
txn: Transaction bytes.
opts: (optional) Transaction options.
Before submitting, the following preflight checks are performed (unless disabled with the `skip_preflight` option):
- The transaction signatures are verified.
- The transaction is simulated against the latest max confirmed bank and on failure an error
will be returned. Preflight checks may be disabled if desired.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> (await solana_client.send_raw_transaction(bytes.fromhex(full_signed_tx_hex))).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
""" # noqa: E501 # pylint: disable=line-too-long
opts_to_use = types.TxOpts(preflight_commitment=self._commitment) if opts is None else opts
body = self._send_raw_transaction_body(txn, opts_to_use)
resp = await self._provider.make_request(body, SendTransactionResp)
if opts_to_use.skip_confirmation:
return self._post_send(resp)
post_send_args = self._send_raw_transaction_post_send_args(resp, opts_to_use)
return await self.__post_send_with_confirm(*post_send_args)
async def send_legacy_transaction(
self,
txn: LegacyTransaction,
*signers: Keypair,
opts: Optional[types.TxOpts] = None,
recent_blockhash: Optional[Blockhash] = None,
) -> SendTransactionResp:
"""Send a legacy transaction.
Args:
txn: transaction object.
signers: Signers to sign the transaction. Only supported for legacy Transaction.
opts: (optional) Transaction options.
recent_blockhash: (optional) Pass a valid recent blockhash here if you want to
skip fetching the recent blockhash or relying on the cache.
Only supported for legacy Transaction.
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solana.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> txn = Transaction().add(transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000)))
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.send_transaction(txn, sender)).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
warn("send_legacy_transaction is deprecated. Use send_transaction instead.", DeprecationWarning, stacklevel=1)
last_valid_block_height = None
if recent_blockhash is None:
blockhash_resp = await self.get_latest_blockhash(Finalized)
recent_blockhash = self.parse_recent_blockhash(blockhash_resp)
last_valid_block_height = blockhash_resp.value.last_valid_block_height
txn.recent_blockhash = recent_blockhash
txn.sign(*signers)
opts_to_use = (
types.TxOpts(
preflight_commitment=self._commitment,
last_valid_block_height=last_valid_block_height,
)
if opts is None
else opts
)
txn_resp = await self.send_raw_transaction(txn.serialize(), opts=opts_to_use)
return txn_resp
async def send_transaction(
self,
txn: Union[VersionedTransaction, Transaction],
opts: Optional[types.TxOpts] = None,
) -> SendTransactionResp:
"""Send a transaction.
Args:
txn: transaction object.
opts: (optional) Transaction options.
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> from solders.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> ixns = [transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))]
>>> msg = Message(ixns, sender.pubkey())
>>> client = AsyncClient("http://localhost:8899")
>>> (await client.send_transaction(Transaction([sender], msg, (await client.get_latest_blockhash()).value.blockhash))) # doctest: +SKIP
""" # noqa: E501
return await self.send_raw_transaction(bytes(txn), opts=opts)
async def simulate_transaction(
self,
txn: Union[Transaction, VersionedTransaction],
sig_verify: bool = False,
commitment: Optional[Commitment] = None,
) -> SimulateTransactionResp:
"""Simulate sending a transaction.
Args:
txn: A Transaction object, a transaction in wire format, or a transaction as base-64 encoded string
The transaction must have a valid blockhash, but is not required to be signed.
sig_verify: If true the transaction signatures will be verified (default: false).
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> tx = Transaction.from_bytes(bytes.fromhex(full_signed_tx_hex))
>>> (await solana_client.simulate_transaction(tx)).value.logs # doctest: +SKIP
['BPF program 83astBRguLMdt2h5U1Tpdq5tjFoJ6noeGwaY3mDLVcri success']
"""
body = self._simulate_transaction_body(txn, sig_verify, commitment)
return await self._provider.make_request(body, SimulateTransactionResp)
async def validator_exit(self) -> ValidatorExitResp:
"""Request to have the validator exit.
Validator must have booted with RPC exit enabled (`--enable-rpc-exit` parameter).
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.validator_exit()).value # doctest: +SKIP
True
"""
return await self._provider.make_request(self._validator_exit, ValidatorExitResp) # type: ignore
async def __post_send_with_confirm(
self,
resp: SendTransactionResp,
conf_comm: Commitment,
last_valid_block_height: Optional[int],
) -> SendTransactionResp:
resp = self._post_send(resp)
sig = resp.value
self._provider.logger.info("Transaction sent to %s. Signature %s: ", self._provider.endpoint_uri, sig)
await self.confirm_transaction(sig, conf_comm, last_valid_block_height=last_valid_block_height)
return resp
async def confirm_transaction(
self,
tx_sig: Signature,
commitment: Optional[Commitment] = None,
sleep_seconds: float = 0.5,
last_valid_block_height: Optional[int] = None,
) -> GetSignatureStatusesResp:
"""Confirm the transaction identified by the specified signature.
Args:
tx_sig: the transaction signature to confirm.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
sleep_seconds: The number of seconds to sleep when polling the signature status.
last_valid_block_height: The block height by which the transaction would become invalid.
"""
commitment_to_use = _COMMITMENT_TO_SOLDERS[commitment or self._commitment]
commitment_rank = int(commitment_to_use)
if last_valid_block_height: # pylint: disable=no-else-return
current_blockheight = (await self.get_block_height(commitment)).value
while current_blockheight <= last_valid_block_height:
resp = await self.get_signature_statuses([tx_sig])
resp_value = resp.value[0]
if resp_value is not None:
confirmation_status = resp_value.confirmation_status
if confirmation_status is not None:
confirmation_rank = int(confirmation_status)
if confirmation_rank >= commitment_rank:
break
current_blockheight = (await self.get_block_height(commitment)).value
await asyncio.sleep(sleep_seconds)
else:
raise TransactionExpiredBlockheightExceededError(f"{tx_sig} has expired: block height exceeded")
return resp
else:
timeout = time() + 90
while time() < timeout:
resp = await self.get_signature_statuses([tx_sig])
resp_value = resp.value[0]
if resp_value is not None:
confirmation_status = resp_value.confirmation_status
if confirmation_status is not None:
confirmation_rank = int(confirmation_status)
if confirmation_rank >= commitment_rank:
break
await asyncio.sleep(sleep_seconds)
else:
raise UnconfirmedTxError(f"Unable to confirm transaction {tx_sig}")
return resp
__init__(self, endpoint=None, commitment=None, timeout=10, extra_headers=None)
special
Init API client.
Source code in solana/rpc/async_api.py
def __init__(
self,
endpoint: Optional[str] = None,
commitment: Optional[Commitment] = None,
timeout: float = 10,
extra_headers: Optional[Dict[str, str]] = None,
) -> None:
"""Init API client."""
super().__init__(commitment)
self._provider = async_http.AsyncHTTPProvider(endpoint, timeout=timeout, extra_headers=extra_headers)
close(self)
async
Use this when you are done with the client.
Source code in solana/rpc/async_api.py
async def close(self) -> None:
"""Use this when you are done with the client."""
await self._provider.close()
confirm_transaction(self, tx_sig, commitment=None, sleep_seconds=0.5, last_valid_block_height=None)
async
Confirm the transaction identified by the specified signature.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tx_sig |
Signature |
the transaction signature to confirm. |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
sleep_seconds |
float |
The number of seconds to sleep when polling the signature status. |
0.5 |
last_valid_block_height |
Optional[int] |
The block height by which the transaction would become invalid. |
None |
Source code in solana/rpc/async_api.py
async def confirm_transaction(
self,
tx_sig: Signature,
commitment: Optional[Commitment] = None,
sleep_seconds: float = 0.5,
last_valid_block_height: Optional[int] = None,
) -> GetSignatureStatusesResp:
"""Confirm the transaction identified by the specified signature.
Args:
tx_sig: the transaction signature to confirm.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
sleep_seconds: The number of seconds to sleep when polling the signature status.
last_valid_block_height: The block height by which the transaction would become invalid.
"""
commitment_to_use = _COMMITMENT_TO_SOLDERS[commitment or self._commitment]
commitment_rank = int(commitment_to_use)
if last_valid_block_height: # pylint: disable=no-else-return
current_blockheight = (await self.get_block_height(commitment)).value
while current_blockheight <= last_valid_block_height:
resp = await self.get_signature_statuses([tx_sig])
resp_value = resp.value[0]
if resp_value is not None:
confirmation_status = resp_value.confirmation_status
if confirmation_status is not None:
confirmation_rank = int(confirmation_status)
if confirmation_rank >= commitment_rank:
break
current_blockheight = (await self.get_block_height(commitment)).value
await asyncio.sleep(sleep_seconds)
else:
raise TransactionExpiredBlockheightExceededError(f"{tx_sig} has expired: block height exceeded")
return resp
else:
timeout = time() + 90
while time() < timeout:
resp = await self.get_signature_statuses([tx_sig])
resp_value = resp.value[0]
if resp_value is not None:
confirmation_status = resp_value.confirmation_status
if confirmation_status is not None:
confirmation_rank = int(confirmation_status)
if confirmation_rank >= commitment_rank:
break
await asyncio.sleep(sleep_seconds)
else:
raise UnconfirmedTxError(f"Unable to confirm transaction {tx_sig}")
return resp
get_account_info(self, pubkey, commitment=None, encoding='base64', data_slice=None)
async
Returns all the account info for the specified public key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of account to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
encoding |
str |
(optional) Encoding for Account data, either "base58" (slow), "base64", or "jsonParsed". Default is "base64".
If jsonParsed is requested but a parser cannot be found, the field falls back to base64 encoding, detectable when the data field is type. (jsonParsed encoding is UNSTABLE). |
'base64' |
data_slice |
Optional[solana.rpc.types.DataSliceOpts] |
(optional) Option to limit the returned account data using the provided |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info(Pubkey([0] * 31 + [1]))).value
Account(
Account {
lamports: 4104230290,
data.len: 0,
owner: 11111111111111111111111111111111,
executable: false,
rent_epoch: 371,
},
)
Source code in solana/rpc/async_api.py
async def get_account_info(
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
encoding: str = "base64",
data_slice: Optional[types.DataSliceOpts] = None,
) -> GetAccountInfoResp:
"""Returns all the account info for the specified public key.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for Account data, either "base58" (slow), "base64", or
"jsonParsed". Default is "base64".
- "base58" is limited to Account data of less than 128 bytes.
- "base64" will return base64 encoded data for Account data of any size.
- "jsonParsed" encoding attempts to use program-specific state parsers to return more human-readable and explicit account state data.
If jsonParsed is requested but a parser cannot be found, the field falls back to base64 encoding,
detectable when the data field is type. (jsonParsed encoding is UNSTABLE).
data_slice: (optional) Option to limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info(Pubkey([0] * 31 + [1]))).value # doctest: +SKIP
Account(
Account {
lamports: 4104230290,
data.len: 0,
owner: 11111111111111111111111111111111,
executable: false,
rent_epoch: 371,
},
)
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_account_info_body(
pubkey=pubkey,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
)
return await self._provider.make_request(body, GetAccountInfoResp)
get_account_info_json_parsed(self, pubkey, commitment=None)
async
Returns all the account info for the specified public key.
If JSON formatting is not available for this account, base64 is returned.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of account to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info_json_parsed(Pubkey([0] * 31 + [1]))).value.owner
Pubkey(
11111111111111111111111111111111,
)
Source code in solana/rpc/async_api.py
async def get_account_info_json_parsed(
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
) -> GetAccountInfoMaybeJsonParsedResp:
"""Returns all the account info for the specified public key.
If JSON formatting is not available for this account, base64 is returned.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_account_info_json_parsed(Pubkey([0] * 31 + [1]))).value.owner # doctest: +SKIP
Pubkey(
11111111111111111111111111111111,
)
"""
body = self._get_account_info_body(pubkey=pubkey, commitment=commitment, encoding="jsonParsed", data_slice=None)
return await self._provider.make_request(body, GetAccountInfoMaybeJsonParsedResp)
get_balance(self, pubkey, commitment=None)
async
Returns the balance of the account of provided Pubkey.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of account to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_balance(Pubkey([0] * 31 + [1]))).value
0
Source code in solana/rpc/async_api.py
async def get_balance(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetBalanceResp:
"""Returns the balance of the account of provided Pubkey.
Args:
pubkey: Pubkey of account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_balance(Pubkey([0] * 31 + [1]))).value # doctest: +SKIP
0
"""
body = self._get_balance_body(pubkey, commitment)
return await self._provider.make_request(body, GetBalanceResp)
get_block(self, slot, encoding='json', max_supported_transaction_version=None)
async
Returns identity and transaction information about a confirmed block in the ledger.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
slot |
int |
Slot, as u64 integer. |
required |
encoding |
str |
(optional) Encoding for the returned Transaction, either "json", "jsonParsed", "base58" (slow), or "base64". If parameter not provided, the default encoding is JSON. |
'json' |
max_supported_transaction_version |
Optional[int] |
(optional) The max transaction version to return in responses. If the requested transaction is a higher version, an error will be returned |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block(1)).value.blockhash
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
Source code in solana/rpc/async_api.py
async def get_block(
self,
slot: int,
encoding: str = "json",
max_supported_transaction_version: Union[int, None] = None,
) -> GetBlockResp:
"""Returns identity and transaction information about a confirmed block in the ledger.
Args:
slot: Slot, as u64 integer.
encoding: (optional) Encoding for the returned Transaction, either "json", "jsonParsed",
"base58" (slow), or "base64". If parameter not provided, the default encoding is JSON.
max_supported_transaction_version: (optional) The max transaction version to return in
responses. If the requested transaction is a higher version, an error will be returned
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block(1)).value.blockhash # doctest: +SKIP
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
"""
body = self._get_block_body(slot, encoding, max_supported_transaction_version)
return await self._provider.make_request(body, GetBlockResp)
get_block_commitment(self, slot)
async
Fetch the commitment for particular block.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
slot |
int |
Block, identified by Slot. |
required |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_commitment(0)).total_stake
497717120
Source code in solana/rpc/async_api.py
async def get_block_commitment(self, slot: int) -> GetBlockCommitmentResp:
"""Fetch the commitment for particular block.
Args:
slot: Block, identified by Slot.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_commitment(0)).total_stake # doctest: +SKIP
497717120
"""
body = self._get_block_commitment_body(slot)
return await self._provider.make_request(body, GetBlockCommitmentResp)
get_block_height(self, commitment=None)
async
Returns the current block height of the node.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_height()).value
1233
Source code in solana/rpc/async_api.py
async def get_block_height(self, commitment: Optional[Commitment] = None) -> GetBlockHeightResp:
"""Returns the current block height of the node.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_height()).value # doctest: +SKIP
1233
"""
body = self._get_block_height_body(commitment)
return await self._provider.make_request(body, GetBlockHeightResp)
get_block_time(self, slot)
async
Fetch the estimated production time of a block.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
slot |
int |
Block, identified by Slot. |
required |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_time(5)).value
1598400007
Source code in solana/rpc/async_api.py
async def get_block_time(self, slot: int) -> GetBlockTimeResp:
"""Fetch the estimated production time of a block.
Args:
slot: Block, identified by Slot.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_block_time(5)).value # doctest: +SKIP
1598400007
"""
body = self._get_block_time_body(slot)
return await self._provider.make_request(body, GetBlockTimeResp)
get_blocks(self, start_slot, end_slot=None)
async
Returns a list of confirmed blocks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start_slot |
int |
Start slot, as u64 integer. |
required |
end_slot |
Optional[int] |
(optional) End slot, as u64 integer. |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_blocks(5, 10)).value
[5, 6, 7, 8, 9, 10]
Source code in solana/rpc/async_api.py
async def get_blocks(self, start_slot: int, end_slot: Optional[int] = None) -> GetBlocksResp:
"""Returns a list of confirmed blocks.
Args:
start_slot: Start slot, as u64 integer.
end_slot: (optional) End slot, as u64 integer.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_blocks(5, 10)).value # doctest: +SKIP
[5, 6, 7, 8, 9, 10]
"""
body = self._get_blocks_body(start_slot, end_slot)
return await self._provider.make_request(body, GetBlocksResp)
get_cluster_nodes(self)
async
Returns information about all the nodes participating in the cluster.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_cluster_nodes()).value[0].tpu
'139.178.65.155:8004'
Source code in solana/rpc/async_api.py
async def get_cluster_nodes(self) -> GetClusterNodesResp:
"""Returns information about all the nodes participating in the cluster.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_cluster_nodes()).value[0].tpu # doctest: +SKIP
'139.178.65.155:8004'
"""
return await self._provider.make_request(self._get_cluster_nodes, GetClusterNodesResp)
get_epoch_info(self, commitment=None)
async
Returns information about the current epoch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_info()).value.epoch
0
Source code in solana/rpc/async_api.py
async def get_epoch_info(self, commitment: Optional[Commitment] = None) -> GetEpochInfoResp:
"""Returns information about the current epoch.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_info()).value.epoch # doctest: +SKIP
0
"""
body = self._get_epoch_info_body(commitment)
return await self._provider.make_request(body, GetEpochInfoResp)
get_epoch_schedule(self)
async
Returns epoch schedule information from this cluster's genesis config.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_schedule()).value.slots_per_epoch
8192
Source code in solana/rpc/async_api.py
async def get_epoch_schedule(self) -> GetEpochScheduleResp:
"""Returns epoch schedule information from this cluster's genesis config.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_epoch_schedule()).value.slots_per_epoch # doctest: +SKIP
8192
"""
return await self._provider.make_request(self._get_epoch_schedule, GetEpochScheduleResp)
get_fee_for_message(self, message, commitment=None)
async
Returns the fee for a message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message |
Union[solders.message.MessageV0, solders.message.Message] |
Message that the fee is requested for. |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> msg = Message([transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))])
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_fee_for_message(msg)).value
5000
Source code in solana/rpc/async_api.py
async def get_fee_for_message(
self, message: VersionedMessage, commitment: Optional[Commitment] = None
) -> GetFeeForMessageResp:
"""Returns the fee for a message.
Args:
message: Message that the fee is requested for.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> msg = Message([transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))])
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_fee_for_message(msg)).value # doctest: +SKIP
5000
"""
body = self._get_fee_for_message_body(message, commitment)
return await self._provider.make_request(body, GetFeeForMessageResp)
get_first_available_block(self)
async
Returns the slot of the lowest confirmed block that has not been purged from the ledger.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_first_available_block()).value
1
Source code in solana/rpc/async_api.py
async def get_first_available_block(self) -> GetFirstAvailableBlockResp:
"""Returns the slot of the lowest confirmed block that has not been purged from the ledger.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_first_available_block()).value # doctest: +SKIP
1
"""
return await self._provider.make_request(self._get_first_available_block, GetFirstAvailableBlockResp)
get_genesis_hash(self)
async
Returns the genesis hash.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_genesis_hash()).value
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
Source code in solana/rpc/async_api.py
async def get_genesis_hash(self) -> GetGenesisHashResp:
"""Returns the genesis hash.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_genesis_hash()).value # doctest: +SKIP
Hash(
EtWTRABZaYq6iMfeYKouRu166VU2xqa1wcaWoxPkrZBG,
)
"""
return await self._provider.make_request(self._get_genesis_hash, GetGenesisHashResp)
get_identity(self)
async
Returns the identity pubkey for the current node.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_identity()).value.identity
Pubkey(
2LVtX3Wq5bhqAYYaUYBRknWaYrsfYiXLQBHTxtHWD2mv,
)
Source code in solana/rpc/async_api.py
async def get_identity(self) -> GetIdentityResp:
"""Returns the identity pubkey for the current node.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_identity()).value.identity # doctest: +SKIP
Pubkey(
2LVtX3Wq5bhqAYYaUYBRknWaYrsfYiXLQBHTxtHWD2mv,
)
"""
return await self._provider.make_request(self._get_identity, GetIdentityResp)
get_inflation_governor(self, commitment=None)
async
Returns the current inflation governor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> await (solana_client.get_inflation_governor()).value.foundation
0.05
Source code in solana/rpc/async_api.py
async def get_inflation_governor(self, commitment: Optional[Commitment] = None) -> GetInflationGovernorResp:
"""Returns the current inflation governor.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> await (solana_client.get_inflation_governor()).value.foundation # doctest: +SKIP
0.05
"""
body = self._get_inflation_governor_body(commitment)
return await self._provider.make_request(body, GetInflationGovernorResp)
get_inflation_rate(self)
async
Returns the specific inflation values for the current epoch.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_rate()).value.epoch
1
Source code in solana/rpc/async_api.py
async def get_inflation_rate(self) -> GetInflationRateResp:
"""Returns the specific inflation values for the current epoch.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_rate()).value.epoch # doctest: +SKIP
1
"""
return await self._provider.make_request(self._get_inflation_rate, GetInflationRateResp)
get_inflation_reward(self, pubkeys, epoch=None, commitment=None)
async
Returns the inflation / staking reward for a list of addresses for an epoch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkeys |
List[solders.pubkey.Pubkey] |
An array of addresses to query, as base-58 encoded strings |
required |
epoch |
Optional[int] |
(optional) An epoch for which the reward occurs. If omitted, the previous epoch will be used |
None |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized" or "confirmed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_reward()).value.amount
2500
Source code in solana/rpc/async_api.py
async def get_inflation_reward(
self, pubkeys: List[Pubkey], epoch: Optional[int] = None, commitment: Optional[Commitment] = None
) -> GetInflationRewardResp:
"""Returns the inflation / staking reward for a list of addresses for an epoch.
Args:
pubkeys: An array of addresses to query, as base-58 encoded strings
epoch: (optional) An epoch for which the reward occurs. If omitted, the previous epoch will be used
commitment: Bank state to query. It can be either "finalized" or "confirmed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_inflation_reward()).value.amount # doctest: +SKIP
2500
"""
body = self._get_inflation_reward_body(pubkeys, epoch, commitment)
return await self._provider.make_request(body, GetInflationRewardResp)
get_largest_accounts(self, filter_opt=None, commitment=None)
async
Returns the 20 largest accounts, by lamport balance.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filter_opt |
Optional[str] |
Filter results by account type; currently supported: circulating|nonCirculating. |
None |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_largest_accounts()).value[0].lamports
500000000000000000
Source code in solana/rpc/async_api.py
async def get_largest_accounts(
self, filter_opt: Optional[str] = None, commitment: Optional[Commitment] = None
) -> GetLargestAccountsResp:
"""Returns the 20 largest accounts, by lamport balance.
Args:
filter_opt: Filter results by account type; currently supported: circulating|nonCirculating.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_largest_accounts()).value[0].lamports # doctest: +SKIP
500000000000000000
"""
body = self._get_largest_accounts_body(filter_opt, commitment)
return await self._provider.make_request(body, GetLargestAccountsResp)
get_latest_blockhash(self, commitment=None)
async
Returns the latest block hash from the ledger.
Response also includes the last valid block height.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_latest_blockhash()).value
RpcBlockhash {
blockhash: Hash(
4TLzN2RAACFnd5TYpHcUi76pC3V1qkggRF29HWk2VLeT,
),
last_valid_block_height: 158286487,
}
Source code in solana/rpc/async_api.py
async def get_latest_blockhash(self, commitment: Optional[Commitment] = None) -> GetLatestBlockhashResp:
"""Returns the latest block hash from the ledger.
Response also includes the last valid block height.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_latest_blockhash()).value # doctest: +SKIP
RpcBlockhash {
blockhash: Hash(
4TLzN2RAACFnd5TYpHcUi76pC3V1qkggRF29HWk2VLeT,
),
last_valid_block_height: 158286487,
}
"""
body = self._get_latest_blockhash_body(commitment)
return await self._provider.make_request(body, GetLatestBlockhashResp)
get_leader_schedule(self, epoch=None, commitment=None)
async
Returns the leader schedule for an epoch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
epoch |
Optional[int] |
Fetch the leader schedule for the epoch that corresponds to the provided slot. If unspecified, the leader schedule for the current epoch is fetched. |
None |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> resp = await solana_client.get_leader_schedule()
>>> list(resp.value.items())[0]
(Pubkey(
HMU77m6WSL9Xew9YvVCgz1hLuhzamz74eD9avi4XPdr,
), [346448, 346449, 346450, 346451, 369140, 369141, 369142, 369143, 384204, 384205, 384206, 384207])
Source code in solana/rpc/async_api.py
async def get_leader_schedule(
self, epoch: Optional[int] = None, commitment: Optional[Commitment] = None
) -> GetLeaderScheduleResp:
"""Returns the leader schedule for an epoch.
Args:
epoch: Fetch the leader schedule for the epoch that corresponds to the provided slot.
If unspecified, the leader schedule for the current epoch is fetched.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> resp = await solana_client.get_leader_schedule() # doctest: +SKIP
>>> list(resp.value.items())[0] # doctest: +SKIP
(Pubkey(
HMU77m6WSL9Xew9YvVCgz1hLuhzamz74eD9avi4XPdr,
), [346448, 346449, 346450, 346451, 369140, 369141, 369142, 369143, 384204, 384205, 384206, 384207])
"""
body = self._get_leader_schedule_body(epoch, commitment)
return await self._provider.make_request(body, GetLeaderScheduleResp)
get_minimum_balance_for_rent_exemption(self, usize, commitment=None)
async
Returns minimum balance required to make account rent exempt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
usize |
int |
Account data length. |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_balance_for_rent_exemption(50)).value
1238880
Source code in solana/rpc/async_api.py
async def get_minimum_balance_for_rent_exemption(
self, usize: int, commitment: Optional[Commitment] = None
) -> GetMinimumBalanceForRentExemptionResp:
"""Returns minimum balance required to make account rent exempt.
Args:
usize: Account data length.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_balance_for_rent_exemption(50)).value # doctest: +SKIP
1238880
"""
body = self._get_minimum_balance_for_rent_exemption_body(usize, commitment)
return await self._provider.make_request(body, GetMinimumBalanceForRentExemptionResp)
get_minimum_ledger_slot(self)
async
Returns the lowest slot that the node has information about in its ledger.
This value may increase over time if the node is configured to purge older ledger data.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_ledger_slot()).value
1234
Source code in solana/rpc/async_api.py
async def get_minimum_ledger_slot(self) -> MinimumLedgerSlotResp:
"""Returns the lowest slot that the node has information about in its ledger.
This value may increase over time if the node is configured to purge older ledger data.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_minimum_ledger_slot()).value # doctest: +SKIP
1234
"""
return await self._provider.make_request(self._minimum_ledger_slot, MinimumLedgerSlotResp)
get_multiple_accounts(self, pubkeys, commitment=None, encoding='base64', data_slice=None)
async
Returns all the account info for a list of public keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkeys |
List[solders.pubkey.Pubkey] |
list of Pubkeys to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
encoding |
str |
(optional) Encoding for Account data, either "base58" (slow) or "base64".
|
'base64' |
data_slice |
Optional[solana.rpc.types.DataSliceOpts] |
(optional) Option to limit the returned account data using the provided |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> (await solana_client.get_multiple_accounts(pubkeys)).value[0].lamports
1
Source code in solana/rpc/async_api.py
async def get_multiple_accounts(
self,
pubkeys: List[Pubkey],
commitment: Optional[Commitment] = None,
encoding: str = "base64",
data_slice: Optional[types.DataSliceOpts] = None,
) -> GetMultipleAccountsResp:
"""Returns all the account info for a list of public keys.
Args:
pubkeys: list of Pubkeys to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for Account data, either "base58" (slow) or "base64".
- "base58" is limited to Account data of less than 128 bytes.
- "base64" will return base64 encoded data for Account data of any size.
data_slice: (optional) Option to limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> (await solana_client.get_multiple_accounts(pubkeys)).value[0].lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_multiple_accounts_body(
pubkeys=pubkeys,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
)
return await self._provider.make_request(body, GetMultipleAccountsResp)
get_multiple_accounts_json_parsed(self, pubkeys, commitment=None)
async
Returns all the account info for a list of public keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkeys |
List[solders.pubkey.Pubkey] |
list of Pubkeys to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> asyncio.run(solana_client.get_multiple_accounts(pubkeys)).value[0].lamports
1
Source code in solana/rpc/async_api.py
async def get_multiple_accounts_json_parsed(
self,
pubkeys: List[Pubkey],
commitment: Optional[Commitment] = None,
) -> GetMultipleAccountsMaybeJsonParsedResp:
"""Returns all the account info for a list of public keys.
Args:
pubkeys: list of Pubkeys to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkeys = [Pubkey.from_string("6ZWcsUiWJ63awprYmbZgBQSreqYZ4s6opowP4b7boUdh"), Pubkey.from_string("HkcE9sqQAnjJtECiFsqGMNmUho3ptXkapUPAqgZQbBSY")]
>>> asyncio.run(solana_client.get_multiple_accounts(pubkeys)).value[0].lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_multiple_accounts_body(
pubkeys=pubkeys,
commitment=commitment,
encoding="jsonParsed",
data_slice=None,
)
return await self._provider.make_request(body, GetMultipleAccountsMaybeJsonParsedResp)
get_program_accounts(self, pubkey, commitment=None, encoding=None, data_slice=None, filters=None)
async
Returns all accounts owned by the provided program Pubkey.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of program |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
encoding |
Optional[str] |
(optional) Encoding for the returned Transaction, either jsonParsed", "base58" (slow), or "base64". |
None |
data_slice |
Optional[solana.rpc.types.DataSliceOpts] |
(optional) Limit the returned account data using the provided |
None |
filters |
Optional[Sequence[Union[int, solana.rpc.types.MemcmpOpts]]] |
(optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a |
None |
Examples:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports
1
Source code in solana/rpc/async_api.py
async def get_program_accounts( # pylint: disable=too-many-arguments
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
encoding: Optional[str] = None,
data_slice: Optional[types.DataSliceOpts] = None,
filters: Optional[Sequence[Union[int, types.MemcmpOpts]]] = None,
) -> GetProgramAccountsResp:
"""Returns all accounts owned by the provided program Pubkey.
Args:
pubkey: Pubkey of program
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
encoding: (optional) Encoding for the returned Transaction, either jsonParsed",
"base58" (slow), or "base64".
data_slice: (optional) Limit the returned account data using the provided `offset`: <usize> and
`length`: <usize> fields; only available for "base58" or "base64" encoding.
filters: (optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a `dataSize` filter.
Example:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_program_accounts_body(
pubkey=pubkey,
commitment=commitment,
encoding=encoding,
data_slice=data_slice,
filters=filters,
)
return await self._provider.make_request(body, GetProgramAccountsResp)
get_program_accounts_json_parsed(self, pubkey, commitment=None, filters=None)
async
Returns all accounts owned by the provided program Pubkey.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of program |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
filters |
Optional[Sequence[Union[int, solana.rpc.types.MemcmpOpts]]] |
(optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a |
None |
Examples:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports
1
Source code in solana/rpc/async_api.py
async def get_program_accounts_json_parsed( # pylint: disable=too-many-arguments
self,
pubkey: Pubkey,
commitment: Optional[Commitment] = None,
filters: Optional[Sequence[Union[int, types.MemcmpOpts]]] = None,
) -> GetProgramAccountsMaybeJsonParsedResp:
"""Returns all accounts owned by the provided program Pubkey.
Args:
pubkey: Pubkey of program
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
filters: (optional) Options to compare a provided series of bytes with program account data at a particular offset.
Note: an int entry is converted to a `dataSize` filter.
Example:
>>> from typing import List, Union
>>> solana_client = AsyncClient("http://localhost:8899")
>>> memcmp_opts = types.MemcmpOpts(offset=4, bytes="3Mc6vR")
>>> pubkey = Pubkey.from_string("4Nd1mBQtrMJVYVfKf2PJy9NZUZdTAsp7D4xWLs4gDB4T")
>>> filters: List[Union[int, types.MemcmpOpts]] = [17, memcmp_opts]
>>> (await solana_client.get_program_accounts(pubkey, filters=filters)).value[0].account.lamports # doctest: +SKIP
1
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_program_accounts_body(
pubkey=pubkey,
commitment=commitment,
encoding="jsonParsed",
data_slice=None,
filters=filters,
)
return await self._provider.make_request(body, GetProgramAccountsMaybeJsonParsedResp)
get_recent_performance_samples(self, limit=None)
async
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
limit |
Optional[int] |
Limit (optional) number of samples to return (maximum 720) |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_recent_performance_samples(1)).value[0]
RpcPerfSample(
RpcPerfSample {
slot: 168036172,
num_transactions: 7159,
num_slots: 158,
sample_period_secs: 60,
},
)
Source code in solana/rpc/async_api.py
async def get_recent_performance_samples(self, limit: Optional[int] = None) -> GetRecentPerformanceSamplesResp:
"""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.
Args:
limit: Limit (optional) number of samples to return (maximum 720)
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_recent_performance_samples(1)).value[0] # doctest: +SKIP
RpcPerfSample(
RpcPerfSample {
slot: 168036172,
num_transactions: 7159,
num_slots: 158,
sample_period_secs: 60,
},
)
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_recent_performance_samples_body(limit)
return await self._provider.make_request(body, GetRecentPerformanceSamplesResp)
get_signature_statuses(self, signatures, search_transaction_history=False)
async
Returns the statuses of a list of signatures.
Unless the search_transaction_history configuration parameter is included, this method only
searches the recent status cache of signatures, which retains statuses for all active slots plus
MAX_RECENT_BLOCKHASHES rooted slots.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signatures |
List[solders.signature.Signature] |
An array of transaction signatures to confirm. |
required |
search_transaction_history |
bool |
If true, a Solana node will search its ledger cache for any signatures not found in the recent status cache. |
False |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> raw_sigs = [
... "5VERv8NMvzbJMEkV8xnrLkEaWRtSz9CosKDYjCJjBRnbJLgp8uirBgmQpjKhoR4tjF3ZpRzrFmBV6UjKdiSZkQUW",
... "5j7s6NiJS3JAkvgkoc18WVAsiSaci2pxB2A6ueCJP4tprA2TFg9wSyTLeYouxPBJEMzJinENTkpA52YStRW5Dia7"]
>>> sigs = [Signature.from_string(sig) for sig in raw_sigs]
>>> (await solana_client.get_signature_statuses(sigs)).value[0].confirmations
10
Source code in solana/rpc/async_api.py
async def get_signature_statuses(
self, signatures: List[Signature], search_transaction_history: bool = False
) -> GetSignatureStatusesResp:
"""Returns the statuses of a list of signatures.
Unless the `search_transaction_history` configuration parameter is included, this method only
searches the recent status cache of signatures, which retains statuses for all active slots plus
`MAX_RECENT_BLOCKHASHES` rooted slots.
Args:
signatures: An array of transaction signatures to confirm.
search_transaction_history: If true, a Solana node will search its ledger cache for
any signatures not found in the recent status cache.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> raw_sigs = [
... "5VERv8NMvzbJMEkV8xnrLkEaWRtSz9CosKDYjCJjBRnbJLgp8uirBgmQpjKhoR4tjF3ZpRzrFmBV6UjKdiSZkQUW",
... "5j7s6NiJS3JAkvgkoc18WVAsiSaci2pxB2A6ueCJP4tprA2TFg9wSyTLeYouxPBJEMzJinENTkpA52YStRW5Dia7"]
>>> sigs = [Signature.from_string(sig) for sig in raw_sigs]
>>> (await solana_client.get_signature_statuses(sigs)).value[0].confirmations # doctest: +SKIP
10
"""
body = self._get_signature_statuses_body(signatures, search_transaction_history)
return await self._provider.make_request(body, GetSignatureStatusesResp)
get_signatures_for_address(self, account, before=None, until=None, limit=None, commitment=None)
async
Returns confirmed signatures for transactions involving an address.
Signatures are returned backwards in time from the provided signature or most recent confirmed block.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
account |
Pubkey |
Account to be queried. |
required |
before |
Optional[solders.signature.Signature] |
(optional) Start searching backwards from this transaction signature. If not provided the search starts from the top of the highest max confirmed block. |
None |
until |
Optional[solders.signature.Signature] |
(optional) Search until this transaction signature, if found before limit reached. |
None |
limit |
Optional[int] |
(optional) Maximum transaction signatures to return (between 1 and 1,000, default: 1,000). |
None |
commitment |
Optional[Commitment] |
(optional) Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.pubkey import Pubkey
>>> pubkey = Pubkey.from_string("Vote111111111111111111111111111111111111111")
>>> (await solana_client.get_signatures_for_address(pubkey, limit=1)).value[0].signature
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
Source code in solana/rpc/async_api.py
async def get_signatures_for_address(
self,
account: Pubkey,
before: Optional[Signature] = None,
until: Optional[Signature] = None,
limit: Optional[int] = None,
commitment: Optional[Commitment] = None,
) -> GetSignaturesForAddressResp:
"""Returns confirmed signatures for transactions involving an address.
Signatures are returned backwards in time from the provided signature or
most recent confirmed block.
Args:
account: Account to be queried.
before: (optional) Start searching backwards from this transaction signature.
If not provided the search starts from the top of the highest max confirmed block.
until: (optional) Search until this transaction signature, if found before limit reached.
limit: (optional) Maximum transaction signatures to return (between 1 and 1,000, default: 1,000).
commitment: (optional) Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.pubkey import Pubkey
>>> pubkey = Pubkey.from_string("Vote111111111111111111111111111111111111111")
>>> (await solana_client.get_signatures_for_address(pubkey, limit=1)).value[0].signature # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
body = self._get_signatures_for_address_body(account, before, until, limit, commitment)
return await self._provider.make_request(body, GetSignaturesForAddressResp)
get_slot(self, commitment=None)
async
Returns the current slot the node is processing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot()).value
7515
Source code in solana/rpc/async_api.py
async def get_slot(self, commitment: Optional[Commitment] = None) -> GetSlotResp:
"""Returns the current slot the node is processing.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot()).value # doctest: +SKIP
7515
"""
body = self._get_slot_body(commitment)
return await self._provider.make_request(body, GetSlotResp)
get_slot_leader(self, commitment=None)
async
Returns the current slot leader.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot_leader()).value
Pubkey(
dv2eQHeP4RFrJZ6UeiZWoc3XTtmtZCUKxxCApCDcRNV,
)
Source code in solana/rpc/async_api.py
async def get_slot_leader(self, commitment: Optional[Commitment] = None) -> GetSlotLeaderResp:
"""Returns the current slot leader.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_slot_leader()).value # doctest: +SKIP
Pubkey(
dv2eQHeP4RFrJZ6UeiZWoc3XTtmtZCUKxxCApCDcRNV,
)
"""
body = self._get_slot_leader_body(commitment)
return await self._provider.make_request(body, GetSlotLeaderResp)
get_stake_activation(self, pubkey, epoch=None, commitment=None)
async
Returns epoch activation information for a stake account.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of stake account to query |
required |
epoch |
Optional[int] |
(optional) Epoch for which to calculate activation details. If parameter not provided, defaults to current epoch. |
None |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_stake_activation()).value.active
124429280
Source code in solana/rpc/async_api.py
async def get_stake_activation(
self,
pubkey: Pubkey,
epoch: Optional[int] = None,
commitment: Optional[Commitment] = None,
) -> GetStakeActivationResp:
"""Returns epoch activation information for a stake account.
Args:
pubkey: Pubkey of stake account to query
epoch: (optional) Epoch for which to calculate activation details. If parameter not provided,
defaults to current epoch.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_stake_activation()).value.active # doctest: +SKIP
124429280
"""
warn("get_stake_activation is deprecated. Use get_account_info instead.", DeprecationWarning, stacklevel=1)
body = self._get_stake_activation_body(pubkey, epoch, commitment)
return await self._provider.make_request(body, GetStakeActivationResp)
get_supply(self, commitment=None)
async
Returns information about the current supply.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_supply()).value.circulating
683635192454157660
Source code in solana/rpc/async_api.py
async def get_supply(self, commitment: Optional[Commitment] = None) -> GetSupplyResp:
"""Returns information about the current supply.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_supply()).value.circulating # doctest: +SKIP
683635192454157660
"""
body = self._get_supply_body(commitment)
return await self._provider.make_request(body, GetSupplyResp)
get_token_account_balance(self, pubkey, commitment=None)
async
Returns the token balance of an SPL Token account (UNSTABLE).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of Token account to query |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkey = Pubkey.from_string("7fUAJdStEuGbc3sM84cKRL6yYaaSstyLSU4ve5oovLS7")
>>> (await solana_client.get_token_account_balance(pubkey)).value.amount # noqa: E501
'9864'
Source code in solana/rpc/async_api.py
async def get_token_account_balance(
self, pubkey: Pubkey, commitment: Optional[Commitment] = None
) -> GetTokenAccountBalanceResp:
"""Returns the token balance of an SPL Token account (UNSTABLE).
Args:
pubkey: Pubkey of Token account to query
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> pubkey = Pubkey.from_string("7fUAJdStEuGbc3sM84cKRL6yYaaSstyLSU4ve5oovLS7")
>>> (await solana_client.get_token_account_balance(pubkey)).value.amount # noqa: E501 # doctest: +SKIP
'9864'
"""
body = self._get_token_account_balance_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenAccountBalanceResp)
get_token_accounts_by_delegate(self, delegate, opts, commitment=None)
async
Returns all SPL Token accounts by approved Delegate (UNSTABLE).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
delegate |
Pubkey |
Public key of the delegate owner to query. |
required |
opts |
TokenAccountOpts |
Token account option specifying at least one of |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Source code in solana/rpc/async_api.py
async def get_token_accounts_by_delegate(
self,
delegate: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByDelegateResp:
"""Returns all SPL Token accounts by approved Delegate (UNSTABLE).
Args:
delegate: Public key of the delegate owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_delegate_body(delegate, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByDelegateResp)
get_token_accounts_by_delegate_json_parsed(self, delegate, opts, commitment=None)
async
Returns all SPL Token accounts by approved delegate in JSON format (UNSTABLE).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
delegate |
Pubkey |
Public key of the delegate owner to query. |
required |
opts |
TokenAccountOpts |
Token account option specifying at least one of |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Source code in solana/rpc/async_api.py
async def get_token_accounts_by_delegate_json_parsed(
self,
delegate: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByDelegateJsonParsedResp:
"""Returns all SPL Token accounts by approved delegate in JSON format (UNSTABLE).
Args:
delegate: Public key of the delegate owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_delegate_json_parsed_body(delegate, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByDelegateJsonParsedResp)
get_token_accounts_by_owner(self, owner, opts, commitment=None)
async
Returns all SPL Token accounts by token owner (UNSTABLE).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
owner |
Pubkey |
Public key of the account owner to query. |
required |
opts |
TokenAccountOpts |
Token account option specifying at least one of |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Source code in solana/rpc/async_api.py
async def get_token_accounts_by_owner(
self,
owner: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByOwnerResp:
"""Returns all SPL Token accounts by token owner (UNSTABLE).
Args:
owner: Public key of the account owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_owner_body(owner, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByOwnerResp)
get_token_accounts_by_owner_json_parsed(self, owner, opts, commitment=None)
async
Returns all SPL Token accounts by token owner in JSON format (UNSTABLE).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
owner |
Pubkey |
Public key of the account owner to query. |
required |
opts |
TokenAccountOpts |
Token account option specifying at least one of |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Source code in solana/rpc/async_api.py
async def get_token_accounts_by_owner_json_parsed(
self,
owner: Pubkey,
opts: types.TokenAccountOpts,
commitment: Optional[Commitment] = None,
) -> GetTokenAccountsByOwnerJsonParsedResp:
"""Returns all SPL Token accounts by token owner in JSON format (UNSTABLE).
Args:
owner: Public key of the account owner to query.
opts: Token account option specifying at least one of `mint` or `program_id`.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
"""
body = self._get_token_accounts_by_owner_json_parsed_body(owner, opts, commitment)
return await self._provider.make_request(body, GetTokenAccountsByOwnerJsonParsedResp)
get_token_largest_accounts(self, pubkey, commitment=None)
async
Returns the 20 largest accounts of a particular SPL Token type.
Source code in solana/rpc/async_api.py
async def get_token_largest_accounts(
self, pubkey: Pubkey, commitment: Optional[Commitment] = None
) -> GetTokenLargestAccountsResp:
"""Returns the 20 largest accounts of a particular SPL Token type."""
body = self._get_token_largest_accounts_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenLargestAccountsResp)
get_token_supply(self, pubkey, commitment=None)
async
Returns the total supply of an SPL Token type.
Source code in solana/rpc/async_api.py
async def get_token_supply(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetTokenSupplyResp:
"""Returns the total supply of an SPL Token type."""
body = self._get_token_supply_body(pubkey, commitment)
return await self._provider.make_request(body, GetTokenSupplyResp)
get_transaction(self, tx_sig, encoding='json', commitment=None, max_supported_transaction_version=None)
async
Returns transaction details for a confirmed transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tx_sig |
Signature |
Transaction signature as base-58 encoded string N encoding attempts to use program-specific
instruction parsers to return more human-readable and explicit data in the
|
required |
encoding |
str |
(optional) Encoding for the returned Transaction, either "json", "jsonParsed", "base58" (slow), or "base64". If parameter not provided, the default encoding is JSON. |
'json' |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
max_supported_transaction_version |
Optional[int] |
(optional) The max transaction version to return in responses. If the requested transaction is a higher version, an error will be returned |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.signature import Signature
>>> sig = Signature.from_string("3PtGYH77LhhQqTXP4SmDVJ85hmDieWsgXCUbn14v7gYyVYPjZzygUQhTk3bSTYnfA48vCM1rmWY7zWL3j1EVKmEy")
>>> (await solana_client.get_transaction(sig)).value.block_time
1234
Source code in solana/rpc/async_api.py
async def get_transaction(
self,
tx_sig: Signature,
encoding: str = "json",
commitment: Optional[Commitment] = None,
max_supported_transaction_version: Optional[int] = None,
) -> GetTransactionResp:
"""Returns transaction details for a confirmed transaction.
Args:
tx_sig: Transaction signature as base-58 encoded string N encoding attempts to use program-specific
instruction parsers to return more human-readable and explicit data in the
`transaction.message.instructions` list.
encoding: (optional) Encoding for the returned Transaction, either "json", "jsonParsed",
"base58" (slow), or "base64". If parameter not provided, the default encoding is JSON.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
max_supported_transaction_version: (optional) The max transaction version to return in responses.
If the requested transaction is a higher version, an error will be returned
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> from solders.signature import Signature
>>> sig = Signature.from_string("3PtGYH77LhhQqTXP4SmDVJ85hmDieWsgXCUbn14v7gYyVYPjZzygUQhTk3bSTYnfA48vCM1rmWY7zWL3j1EVKmEy")
>>> (await solana_client.get_transaction(sig)).value.block_time # doctest: +SKIP
1234
""" # noqa: E501 # pylint: disable=line-too-long
body = self._get_transaction_body(tx_sig, encoding, commitment, max_supported_transaction_version)
return await self._provider.make_request(body, GetTransactionResp)
get_transaction_count(self, commitment=None)
async
Returns the current Transaction count from the ledger.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_transaction_count()).value
4554
Source code in solana/rpc/async_api.py
async def get_transaction_count(self, commitment: Optional[Commitment] = None) -> GetTransactionCountResp:
"""Returns the current Transaction count from the ledger.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_transaction_count()).value # doctest: +SKIP
4554
"""
body = self._get_transaction_count_body(commitment)
return await self._provider.make_request(body, GetTransactionCountResp)
get_version(self)
async
Returns the current solana versions running on the node.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_version()).value.solana_core
'1.13.2'
Source code in solana/rpc/async_api.py
async def get_version(self) -> GetVersionResp:
"""Returns the current solana versions running on the node.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_version()).value.solana_core # doctest: +SKIP
'1.13.2'
"""
return await self._provider.make_request(self._get_version, GetVersionResp)
get_vote_accounts(self, commitment=None)
async
Returns the account info and associated stake for all the voting accounts in the current bank.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_vote_accounts()).value.current[0].commission
100
Source code in solana/rpc/async_api.py
async def get_vote_accounts(self, commitment: Optional[Commitment] = None) -> GetVoteAccountsResp:
"""Returns the account info and associated stake for all the voting accounts in the current bank.
Args:
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.get_vote_accounts()).value.current[0].commission # doctest: +SKIP
100
"""
body = self._get_vote_accounts_body(commitment)
return await self._provider.make_request(body, GetVoteAccountsResp)
is_connected(self)
async
Health check.
solana_client = AsyncClient("http://localhost:8899") asyncio.run(solana_client.is_connected()) # doctest: +SKIP True
Returns:
| Type | Description |
|---|---|
bool |
True if the client is connected. |
Source code in solana/rpc/async_api.py
async def is_connected(self) -> bool:
"""Health check.
>>> solana_client = AsyncClient("http://localhost:8899")
>>> asyncio.run(solana_client.is_connected()) # doctest: +SKIP
True
Returns:
True if the client is connected.
"""
body = self._get_health_body()
response = await self._provider.make_request(body, GetHealthResp)
return response.value == "ok"
request_airdrop(self, pubkey, lamports, commitment=None)
async
Requests an airdrop of lamports to a Pubkey.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pubkey |
Pubkey |
Pubkey of account to receive lamports, as base-58 encoded string or public key object. |
required |
lamports |
int |
Amount of lamports. |
required |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.request_airdrop(Pubkey([0] * 31 + [1]), 10000)).value
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
Source code in solana/rpc/async_api.py
async def request_airdrop(
self, pubkey: Pubkey, lamports: int, commitment: Optional[Commitment] = None
) -> RequestAirdropResp:
"""Requests an airdrop of lamports to a Pubkey.
Args:
pubkey: Pubkey of account to receive lamports, as base-58 encoded string or public key object.
lamports: Amount of lamports.
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> from solders.pubkey import Pubkey
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.request_airdrop(Pubkey([0] * 31 + [1]), 10000)).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
body = self._request_airdrop_body(pubkey, lamports, commitment)
return await self._provider.make_request(body, RequestAirdropResp)
send_legacy_transaction(self, txn, *signers, *, opts=None, recent_blockhash=None)
async
Send a legacy transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
txn |
Transaction |
transaction object. |
required |
signers |
Keypair |
Signers to sign the transaction. Only supported for legacy Transaction. |
() |
opts |
Optional[solana.rpc.types.TxOpts] |
(optional) Transaction options. |
None |
recent_blockhash |
Optional[solders.hash.Hash] |
(optional) Pass a valid recent blockhash here if you want to skip fetching the recent blockhash or relying on the cache. Only supported for legacy Transaction. |
None |
Examples:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solana.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> txn = Transaction().add(transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000)))
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.send_transaction(txn, sender)).value
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
Source code in solana/rpc/async_api.py
async def send_legacy_transaction(
self,
txn: LegacyTransaction,
*signers: Keypair,
opts: Optional[types.TxOpts] = None,
recent_blockhash: Optional[Blockhash] = None,
) -> SendTransactionResp:
"""Send a legacy transaction.
Args:
txn: transaction object.
signers: Signers to sign the transaction. Only supported for legacy Transaction.
opts: (optional) Transaction options.
recent_blockhash: (optional) Pass a valid recent blockhash here if you want to
skip fetching the recent blockhash or relying on the cache.
Only supported for legacy Transaction.
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solana.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> txn = Transaction().add(transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000)))
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.send_transaction(txn, sender)).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
"""
warn("send_legacy_transaction is deprecated. Use send_transaction instead.", DeprecationWarning, stacklevel=1)
last_valid_block_height = None
if recent_blockhash is None:
blockhash_resp = await self.get_latest_blockhash(Finalized)
recent_blockhash = self.parse_recent_blockhash(blockhash_resp)
last_valid_block_height = blockhash_resp.value.last_valid_block_height
txn.recent_blockhash = recent_blockhash
txn.sign(*signers)
opts_to_use = (
types.TxOpts(
preflight_commitment=self._commitment,
last_valid_block_height=last_valid_block_height,
)
if opts is None
else opts
)
txn_resp = await self.send_raw_transaction(txn.serialize(), opts=opts_to_use)
return txn_resp
send_raw_transaction(self, txn, opts=None)
async
Send a transaction that has already been signed and serialized into the wire format.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
txn |
bytes |
Transaction bytes. |
required |
opts |
Optional[solana.rpc.types.TxOpts] |
(optional) Transaction options. |
None |
Before submitting, the following preflight checks are performed (unless disabled with the skip_preflight option):
- The transaction signatures are verified.
- The transaction is simulated against the latest max confirmed bank and on failure an error
will be returned. Preflight checks may be disabled if desired.
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> (await solana_client.send_raw_transaction(bytes.fromhex(full_signed_tx_hex))).value
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
Source code in solana/rpc/async_api.py
async def send_raw_transaction(self, txn: bytes, opts: Optional[types.TxOpts] = None) -> SendTransactionResp:
"""Send a transaction that has already been signed and serialized into the wire format.
Args:
txn: Transaction bytes.
opts: (optional) Transaction options.
Before submitting, the following preflight checks are performed (unless disabled with the `skip_preflight` option):
- The transaction signatures are verified.
- The transaction is simulated against the latest max confirmed bank and on failure an error
will be returned. Preflight checks may be disabled if desired.
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> (await solana_client.send_raw_transaction(bytes.fromhex(full_signed_tx_hex))).value # doctest: +SKIP
Signature(
1111111111111111111111111111111111111111111111111111111111111111,
)
""" # noqa: E501 # pylint: disable=line-too-long
opts_to_use = types.TxOpts(preflight_commitment=self._commitment) if opts is None else opts
body = self._send_raw_transaction_body(txn, opts_to_use)
resp = await self._provider.make_request(body, SendTransactionResp)
if opts_to_use.skip_confirmation:
return self._post_send(resp)
post_send_args = self._send_raw_transaction_post_send_args(resp, opts_to_use)
return await self.__post_send_with_confirm(*post_send_args)
send_transaction(self, txn, opts=None)
async
Send a transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
txn |
Union[solders.transaction.VersionedTransaction, solders.transaction.Transaction] |
transaction object. |
required |
opts |
Optional[solana.rpc.types.TxOpts] |
(optional) Transaction options. |
None |
Examples:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> from solders.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> ixns = [transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))]
>>> msg = Message(ixns, sender.pubkey())
>>> client = AsyncClient("http://localhost:8899")
>>> (await client.send_transaction(Transaction([sender], msg, (await client.get_latest_blockhash()).value.blockhash)))
Source code in solana/rpc/async_api.py
async def send_transaction(
self,
txn: Union[VersionedTransaction, Transaction],
opts: Optional[types.TxOpts] = None,
) -> SendTransactionResp:
"""Send a transaction.
Args:
txn: transaction object.
opts: (optional) Transaction options.
Example:
>>> from solders.keypair import Keypair
>>> from solders.system_program import TransferParams, transfer
>>> from solders.message import Message
>>> from solders.transaction import Transaction
>>> leading_zeros = [0] * 31
>>> sender, receiver = Keypair.from_seed(leading_zeros + [1]), Keypair.from_seed(leading_zeros + [2])
>>> ixns = [transfer(TransferParams(
... from_pubkey=sender.pubkey(), to_pubkey=receiver.pubkey(), lamports=1000))]
>>> msg = Message(ixns, sender.pubkey())
>>> client = AsyncClient("http://localhost:8899")
>>> (await client.send_transaction(Transaction([sender], msg, (await client.get_latest_blockhash()).value.blockhash))) # doctest: +SKIP
""" # noqa: E501
return await self.send_raw_transaction(bytes(txn), opts=opts)
simulate_transaction(self, txn, sig_verify=False, commitment=None)
async
Simulate sending a transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
txn |
Union[solders.transaction.Transaction, solders.transaction.VersionedTransaction] |
A Transaction object, a transaction in wire format, or a transaction as base-64 encoded string The transaction must have a valid blockhash, but is not required to be signed. |
required |
sig_verify |
bool |
If true the transaction signatures will be verified (default: false). |
False |
commitment |
Optional[Commitment] |
Bank state to query. It can be either "finalized", "confirmed" or "processed". |
None |
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> tx = Transaction.from_bytes(bytes.fromhex(full_signed_tx_hex))
>>> (await solana_client.simulate_transaction(tx)).value.logs
['BPF program 83astBRguLMdt2h5U1Tpdq5tjFoJ6noeGwaY3mDLVcri success']
Source code in solana/rpc/async_api.py
async def simulate_transaction(
self,
txn: Union[Transaction, VersionedTransaction],
sig_verify: bool = False,
commitment: Optional[Commitment] = None,
) -> SimulateTransactionResp:
"""Simulate sending a transaction.
Args:
txn: A Transaction object, a transaction in wire format, or a transaction as base-64 encoded string
The transaction must have a valid blockhash, but is not required to be signed.
sig_verify: If true the transaction signatures will be verified (default: false).
commitment: Bank state to query. It can be either "finalized", "confirmed" or "processed".
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> full_signed_tx_hex = (
... '01b3795ccfaac3eee838bb05c3b8284122c18acedcd645c914fe8e178c3b62640d8616d061cc818b26cab8ecf3855ecc'
... '72fa113f731ecbd0215e88edc0309d6f0a010001031398f62c6d1a457c51ba6a4b5f3dbd2f69fca93216218dc8997e41'
... '6bd17d93ca68ab4677ffb1f2894dd0a6153c231d45ec436ae53ae60149dbe15f32e4b8703f0000000000000000000000'
... '000000000000000000000000000000000000000000839618f701ba7e9ba27ae59825dd6d6bb66d14f6d5d0eae215161d7'
... '1851a106901020200010c0200000040420f0000000000'
... )
>>> tx = Transaction.from_bytes(bytes.fromhex(full_signed_tx_hex))
>>> (await solana_client.simulate_transaction(tx)).value.logs # doctest: +SKIP
['BPF program 83astBRguLMdt2h5U1Tpdq5tjFoJ6noeGwaY3mDLVcri success']
"""
body = self._simulate_transaction_body(txn, sig_verify, commitment)
return await self._provider.make_request(body, SimulateTransactionResp)
validator_exit(self)
async
Request to have the validator exit.
Validator must have booted with RPC exit enabled (--enable-rpc-exit parameter).
Examples:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.validator_exit()).value
True
Source code in solana/rpc/async_api.py
async def validator_exit(self) -> ValidatorExitResp:
"""Request to have the validator exit.
Validator must have booted with RPC exit enabled (`--enable-rpc-exit` parameter).
Example:
>>> solana_client = AsyncClient("http://localhost:8899")
>>> (await solana_client.validator_exit()).value # doctest: +SKIP
True
"""
return await self._provider.make_request(self._validator_exit, ValidatorExitResp) # type: ignore