tessl/pypi-solana

Comprehensive Python SDK for interacting with the Solana blockchain network including RPC clients, SPL Token support, and WebSocket subscriptions.

Overview

Eval results

Files

RPC Client Operations

Name: tessl/pypi-solana
Author: tessl

Complete Solana JSON-RPC API client with synchronous and asynchronous interfaces. The RPC client provides comprehensive access to the Solana blockchain network, including account queries, transaction operations, network information, and real-time WebSocket subscriptions.

Capabilities

Client Initialization

Creates RPC clients for connecting to Solana clusters with configurable endpoints, commitment levels, timeouts, and connection parameters.

class Client:
    def __init__(self, endpoint: str, commitment: Optional[Commitment] = None, timeout: float = 30, extra_headers: Optional[Dict[str, str]] = None, proxy: Optional[str] = None):
        """
        Initialize synchronous RPC client.
        
        Parameters:
        - endpoint: RPC endpoint URL (e.g., "https://api.mainnet-beta.solana.com")
        - commitment: Default commitment level for requests
        - timeout: Request timeout in seconds
        - extra_headers: Additional HTTP headers
        - proxy: Proxy URL if needed
        """

class AsyncClient:
    def __init__(self, endpoint: str, commitment: Optional[Commitment] = None, timeout: float = 30, extra_headers: Optional[Dict[str, str]] = None, proxy: Optional[str] = None):
        """
        Initialize asynchronous RPC client.
        
        Parameters:
        - endpoint: RPC endpoint URL (e.g., "https://api.mainnet-beta.solana.com")
        - commitment: Default commitment level for requests
        - timeout: Request timeout in seconds
        - extra_headers: Additional HTTP headers
        - proxy: Proxy URL if needed
        """

    def is_connected(self) -> bool:
        """Check if client is connected to the RPC endpoint."""

Account Operations

Retrieve account information, balances, and query multiple accounts efficiently with support for different data encodings and filtering options.

def get_balance(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetBalanceResp:
    """
    Get SOL balance for an account.
    
    Parameters:
    - pubkey: Account public key
    - commitment: Commitment level for the query
    
    Returns:
    GetBalanceResp with value field containing lamports
    """

def get_account_info(self, pubkey: Pubkey, commitment: Optional[Commitment] = None, encoding: str = "base64", data_slice: Optional[DataSliceOpts] = None) -> GetAccountInfoResp:
    """
    Get account information and data.
    
    Parameters:
    - pubkey: Account public key
    - commitment: Commitment level for the query
    - encoding: Data encoding ("base64", "base58", "base64+zstd", "jsonParsed")
    - data_slice: Optional data slice to retrieve partial account data
    
    Returns:
    GetAccountInfoResp with account details and data
    """

def get_account_info_json_parsed(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetAccountInfoMaybeJsonParsedResp:
    """
    Get account information with JSON-parsed data when possible.
    
    Parameters:
    - pubkey: Account public key  
    - commitment: Commitment level for the query
    
    Returns:
    GetAccountInfoMaybeJsonParsedResp with parsed account data
    """

def get_multiple_accounts(self, pubkeys: List[Pubkey], commitment: Optional[Commitment] = None, encoding: str = "base64", data_slice: Optional[DataSliceOpts] = None) -> GetMultipleAccountsResp:
    """
    Get information for multiple accounts in a single request.
    
    Parameters:
    - pubkeys: List of account public keys (max 100)
    - commitment: Commitment level for the query
    - encoding: Data encoding for account data
    - data_slice: Optional data slice for partial account data
    
    Returns:
    GetMultipleAccountsResp with array of account information
    """

def get_multiple_accounts_json_parsed(self, pubkeys: List[Pubkey], commitment: Optional[Commitment] = None) -> GetMultipleAccountsMaybeJsonParsedResp:
    """
    Get multiple accounts with JSON-parsed data when possible.
    
    Parameters:
    - pubkeys: List of account public keys (max 100)
    - commitment: Commitment level for the query
    
    Returns:
    GetMultipleAccountsMaybeJsonParsedResp with parsed account data
    """

Transaction Operations

Send transactions, simulate execution, check transaction status, and confirm transaction completion with comprehensive transaction management capabilities.

def send_transaction(self, txn: Transaction, opts: Optional[TxOpts] = None) -> SendTransactionResp:
    """
    Send a signed transaction to the network.
    
    Parameters:
    - txn: Signed transaction object
    - opts: Transaction options (skip_preflight, encoding, etc.)
    
    Returns:
    SendTransactionResp with transaction signature
    """

def send_raw_transaction(self, txn: bytes, opts: Optional[TxOpts] = None) -> SendTransactionResp:
    """
    Send a raw transaction as bytes.
    
    Parameters:
    - txn: Serialized transaction bytes
    - opts: Transaction options
    
    Returns:
    SendTransactionResp with transaction signature
    """

def simulate_transaction(self, txn: Transaction, sig_verify: bool = False, commitment: Optional[Commitment] = None) -> SimulateTransactionResp:
    """
    Simulate transaction execution without sending to network.
    
    Parameters:
    - txn: Transaction to simulate
    - sig_verify: Whether to verify signatures
    - commitment: Commitment level for simulation
    
    Returns:
    SimulateTransactionResp with simulation results and logs
    """

def get_transaction(self, tx_sig: Signature, encoding: str = "json", commitment: Optional[Commitment] = None, max_supported_transaction_version: Optional[int] = None) -> GetTransactionResp:
    """
    Get confirmed transaction details.
    
    Parameters:
    - tx_sig: Transaction signature
    - encoding: Response encoding ("json", "jsonParsed", "base64", "base58")
    - commitment: Commitment level
    - max_supported_transaction_version: Max transaction version to support
    
    Returns:
    GetTransactionResp with transaction details
    """

def get_signature_statuses(self, signatures: List[Signature], search_transaction_history: bool = False) -> GetSignatureStatusesResp:
    """
    Get status of transaction signatures.
    
    Parameters:
    - signatures: List of transaction signatures (max 256)
    - search_transaction_history: Search transaction history if not found in recent cache
    
    Returns:
    GetSignatureStatusesResp with status for each signature
    """

def confirm_transaction(self, tx_sig: Signature, commitment: Optional[Commitment] = None, sleep_seconds: float = 0.5, last_valid_block_height: Optional[int] = None) -> GetSignatureStatusesResp:
    """
    Wait for transaction confirmation.
    
    Parameters:
    - tx_sig: Transaction signature to confirm
    - commitment: Commitment level to wait for
    - sleep_seconds: Sleep duration between status checks
    - last_valid_block_height: Block height after which to stop waiting
    
    Returns:
    GetSignatureStatusesResp when transaction is confirmed
    """

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:
    """
    Get transaction signatures for an address.
    
    Parameters:
    - account: Account public key
    - before: Start searching backwards from this signature
    - until: Search until this signature
    - limit: Maximum number of signatures to return (max 1000)
    - commitment: Commitment level
    
    Returns:
    GetSignaturesForAddressResp with array of signature information
    """

Block Operations

Access block information, block heights, block time data, and blockchain progression with comprehensive block query capabilities.

def get_block(self, slot: int, encoding: str = "json", max_supported_transaction_version: Optional[int] = None) -> GetBlockResp:
    """
    Get block information for a specific slot.
    
    Parameters:
    - slot: Slot number
    - encoding: Response encoding ("json", "jsonParsed", "base64", "base58")
    - max_supported_transaction_version: Max transaction version to support
    
    Returns:
    GetBlockResp with block details and transactions
    """

def get_block_height(self, commitment: Optional[Commitment] = None) -> GetBlockHeightResp:
    """
    Get current block height.
    
    Parameters:
    - commitment: Commitment level
    
    Returns:
    GetBlockHeightResp with current block height
    """

def get_block_time(self, slot: int) -> GetBlockTimeResp:
    """
    Get estimated production time of a block.
    
    Parameters:
    - slot: Block slot number
    
    Returns:
    GetBlockTimeResp with Unix timestamp
    """

def get_blocks(self, start_slot: int, end_slot: Optional[int] = None) -> GetBlocksResp:
    """
    Get list of confirmed blocks between two slots.
    
    Parameters:
    - start_slot: Start slot (inclusive)
    - end_slot: End slot (inclusive), if None returns up to 500,000 blocks
    
    Returns:
    GetBlocksResp with array of block slot numbers
    """

def get_latest_blockhash(self, commitment: Optional[Commitment] = None) -> GetLatestBlockhashResp:
    """
    Get latest blockhash.
    
    Parameters:
    - commitment: Commitment level
    
    Returns:
    GetLatestBlockhashResp with blockhash and last valid block height
    """

def get_first_available_block() -> GetFirstAvailableBlockResp:
    """
    Get the slot of the lowest confirmed block that has not been purged.
    
    Returns:
    GetFirstAvailableBlockResp with slot number
    """

Network and Cluster Information

Retrieve network statistics, validator information, epoch data, and cluster configuration details.

def get_cluster_nodes() -> GetClusterNodesResp:
    """
    Get information about cluster nodes.
    
    Returns:
    GetClusterNodesResp with array of node information
    """

def get_epoch_info(self, commitment: Optional[Commitment] = None) -> GetEpochInfoResp:
    """
    Get current epoch information.
    
    Parameters:
    - commitment: Commitment level
    
    Returns:
    GetEpochInfoResp with epoch details
    """

def get_slot(self, commitment: Optional[Commitment] = None) -> GetSlotResp:
    """
    Get current slot number.
    
    Parameters:
    - commitment: Commitment level
    
    Returns:
    GetSlotResp with current slot
    """

def get_version() -> GetVersionResp:
    """
    Get current Solana versions running on the node.
    
    Returns:
    GetVersionResp with version information
    """

def get_supply(self, commitment: Optional[Commitment] = None) -> GetSupplyResp:
    """
    Get information about the current supply.
    
    Parameters:
    - commitment: Commitment level
    
    Returns:
    GetSupplyResp with supply information
    """

def get_recent_performance_samples(self, limit: Optional[int] = None) -> GetRecentPerformanceSamplesResp:
    """
    Get recent performance samples.
    
    Parameters:
    - limit: Number of samples to return (max 720)
    
    Returns:
    GetRecentPerformanceSamplesResp with performance samples
    """

def get_vote_accounts(self, vote_pubkey: Optional[Pubkey] = None, commitment: Optional[Commitment] = None, keep_unstaked_delinquents: Optional[bool] = None, delinquent_slot_distance: Optional[int] = None) -> GetVoteAccountsResp:
    """
    Get vote accounts information.
    
    Parameters:
    - vote_pubkey: Specific vote account to query
    - commitment: Commitment level
    - keep_unstaked_delinquents: Include unstaked delinquent validators
    - delinquent_slot_distance: Slot distance for delinquent determination
    
    Returns:
    GetVoteAccountsResp with current and delinquent vote accounts
    """

Program Account Queries

Query accounts owned by specific programs with flexible filtering options and data encoding support.

def get_program_accounts(self, pubkey: Pubkey, commitment: Optional[Commitment] = None, encoding: str = "base64", data_slice: Optional[DataSliceOpts] = None, filters: Optional[List[Union[int, MemcmpOpts]]] = None) -> GetProgramAccountsResp:
    """
    Get accounts owned by a program.
    
    Parameters:
    - pubkey: Program public key
    - commitment: Commitment level
    - encoding: Account data encoding
    - data_slice: Data slice options for partial data
    - filters: Account filters (data size and memcmp filters)
    
    Returns:
    GetProgramAccountsResp with array of accounts
    """

def get_program_accounts_json_parsed(self, pubkey: Pubkey, commitment: Optional[Commitment] = None, filters: Optional[List[Union[int, MemcmpOpts]]] = None) -> GetProgramAccountsMaybeJsonParsedResp:
    """
    Get program accounts with JSON-parsed data when possible.
    
    Parameters:
    - pubkey: Program public key
    - commitment: Commitment level
    - filters: Account filters
    
    Returns:
    GetProgramAccountsMaybeJsonParsedResp with parsed account data
    """

Token-Specific Queries

Specialized queries for SPL Token accounts, balances, and token supply information.

def get_token_account_balance(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetTokenAccountBalanceResp:
    """
    Get token account balance.
    
    Parameters:
    - pubkey: Token account public key
    - commitment: Commitment level
    
    Returns:
    GetTokenAccountBalanceResp with token balance details
    """

def get_token_accounts_by_owner(self, owner: Pubkey, opts: TokenAccountOpts, commitment: Optional[Commitment] = None) -> GetTokenAccountsByOwnerResp:
    """
    Get token accounts owned by an address.
    
    Parameters:
    - owner: Owner public key
    - opts: Query options (mint or program_id filter)
    - commitment: Commitment level
    
    Returns:
    GetTokenAccountsByOwnerResp with token accounts
    """

def get_token_supply(self, pubkey: Pubkey, commitment: Optional[Commitment] = None) -> GetTokenSupplyResp:
    """
    Get token supply information.
    
    Parameters:
    - pubkey: Token mint public key
    - commitment: Commitment level
    
    Returns:
    GetTokenSupplyResp with supply details
    """

Economic Information

Access economic data including fees, inflation rates, rent calculations, and validator performance metrics.

def get_fee_for_message(self, message: Message, commitment: Optional[Commitment] = None) -> GetFeeForMessageResp:
    """
    Get fee required for a message.
    
    Parameters:
    - message: Transaction message
    - commitment: Commitment level
    
    Returns:
    GetFeeForMessageResp with fee in lamports
    """

def get_minimum_balance_for_rent_exemption(self, usize: int, commitment: Optional[Commitment] = None) -> GetMinimumBalanceForRentExemptionResp:
    """
    Get minimum balance needed for rent exemption.
    
    Parameters:
    - usize: Account data size in bytes
    - commitment: Commitment level
    
    Returns:
    GetMinimumBalanceForRentExemptionResp with minimum balance in lamports
    """

def get_inflation_rate() -> GetInflationRateResp:
    """
    Get current inflation rate.
    
    Returns:
    GetInflationRateResp with inflation rate details
    """

def request_airdrop(self, pubkey: Pubkey, lamports: int, commitment: Optional[Commitment] = None) -> RequestAirdropResp:
    """
    Request SOL airdrop (devnet/testnet only).
    
    Parameters:
    - pubkey: Public key to receive airdrop
    - lamports: Amount of lamports to request
    - commitment: Commitment level
    
    Returns:
    RequestAirdropResp with airdrop transaction signature
    """

WebSocket API

Real-time subscriptions for blockchain events and account changes.

# WebSocket subscriptions are available through the websocket_api module
# for real-time account, program, and signature updates

Types

class GetBalanceResp:
    value: int

class GetAccountInfoResp:
    value: Optional[Account]

class Account:
    lamports: int
    data: List[str]  # [data, encoding]
    owner: str
    executable: bool
    rent_epoch: int

class SendTransactionResp:
    value: str  # Transaction signature

class GetLatestBlockhashResp:
    value: RpcBlockhash

class RpcBlockhash:
    blockhash: Hash
    last_valid_block_height: int

class DataSliceOpts(NamedTuple):
    offset: int
    length: int

class MemcmpOpts(NamedTuple):
    offset: int
    bytes_: str

class TokenAccountOpts(NamedTuple):
    mint: Optional[Pubkey] = None
    program_id: Optional[Pubkey] = None

class TxOpts(NamedTuple):
    skip_preflight: bool = False
    preflight_commitment: Optional[Commitment] = None
    encoding: str = "base64"
    max_retries: Optional[int] = None
    skip_confirmation: bool = False

Install with Tessl CLI