tessl/pypi-solders
Python bindings for Solana Rust tools providing high-performance blockchain development primitives, RPC functionality, and testing infrastructure.
—
Pending
transaction-status.mddocs/
Transaction Status
Transaction metadata, execution results, and status tracking including error handling, parsed transaction data, and confirmation information. This provides comprehensive tools for monitoring transaction lifecycle and analyzing execution outcomes.
Capabilities
Transaction Status and Confirmation
Core transaction status information including confirmation levels and execution metadata.
class TransactionStatus:
"""
Transaction confirmation status and metadata.
"""
def __init__(
self,
slot: int,
confirmations: Optional[int],
status: Optional['TransactionErrorType'],
confirmation_status: Optional['TransactionConfirmationStatus']
):
"""
Create transaction status.
Parameters:
- slot: int, slot where transaction was processed
- confirmations: Optional[int], number of confirmations (None if not confirmed)
- status: Optional[TransactionErrorType], error status (None if successful)
- confirmation_status: Optional[TransactionConfirmationStatus], confirmation level
"""
@property
def slot(self) -> int:
"""Slot where transaction was processed."""
@property
def confirmations(self) -> Optional[int]:
"""Number of confirmations (None if not confirmed)."""
@property
def status(self) -> Optional['TransactionErrorType']:
"""Error status (None if successful)."""
@property
def confirmation_status(self) -> Optional['TransactionConfirmationStatus']:
"""Current confirmation level."""
def is_successful(self) -> bool:
"""
Check if transaction was successful.
Returns:
bool, True if transaction succeeded
"""
def is_confirmed(self) -> bool:
"""
Check if transaction is confirmed.
Returns:
bool, True if transaction has confirmations
"""
class TransactionConfirmationStatus:
"""Transaction confirmation level enumeration."""
Processed: 'TransactionConfirmationStatus' # Transaction processed
Confirmed: 'TransactionConfirmationStatus' # Transaction confirmed
Finalized: 'TransactionConfirmationStatus' # Transaction finalized
def __str__(self) -> str:
"""String representation of confirmation status."""Transaction Metadata
Comprehensive transaction execution metadata including fees, logs, and state changes.
class UiTransactionStatusMeta:
"""
Transaction execution metadata in UI-friendly format.
"""
def __init__(
self,
err: Optional['TransactionErrorType'],
fee: int,
pre_balances: List[int],
post_balances: List[int],
inner_instructions: Optional[List['UiInnerInstructions']],
log_messages: Optional[List[str]],
pre_token_balances: Optional[List['UiTransactionTokenBalance']],
post_token_balances: Optional[List['UiTransactionTokenBalance']],
rewards: Optional[List['Reward']],
loaded_addresses: Optional['UiLoadedAddresses'],
return_data: Optional['TransactionReturnData'],
compute_units_consumed: Optional[int]
):
"""
Create transaction metadata.
Parameters:
- err: Optional[TransactionErrorType], transaction error (None if successful)
- fee: int, transaction fee in lamports
- pre_balances: List[int], account balances before execution
- post_balances: List[int], account balances after execution
- inner_instructions: Optional[List], cross-program invocation instructions
- log_messages: Optional[List[str]], program log messages
- pre_token_balances: Optional[List], token balances before execution
- post_token_balances: Optional[List], token balances after execution
- rewards: Optional[List[Reward]], epoch rewards (if any)
- loaded_addresses: Optional[UiLoadedAddresses], addresses loaded from lookup tables
- return_data: Optional[TransactionReturnData], program return data
- compute_units_consumed: Optional[int], compute units used
"""
@property
def err(self) -> Optional['TransactionErrorType']:
"""Transaction error (None if successful)."""
@property
def fee(self) -> int:
"""Transaction fee in lamports."""
@property
def pre_balances(self) -> List[int]:
"""Account balances before execution."""
@property
def post_balances(self) -> List[int]:
"""Account balances after execution."""
@property
def inner_instructions(self) -> Optional[List['UiInnerInstructions']]:
"""Cross-program invocation instructions."""
@property
def log_messages(self) -> Optional[List[str]]:
"""Program log messages."""
@property
def pre_token_balances(self) -> Optional[List['UiTransactionTokenBalance']]:
"""Token balances before execution."""
@property
def post_token_balances(self) -> Optional[List['UiTransactionTokenBalance']]:
"""Token balances after execution."""
@property
def rewards(self) -> Optional[List['Reward']]:
"""Epoch rewards (if any)."""
@property
def loaded_addresses(self) -> Optional['UiLoadedAddresses']:
"""Addresses loaded from lookup tables."""
@property
def return_data(self) -> Optional['TransactionReturnData']:
"""Program return data."""
@property
def compute_units_consumed(self) -> Optional[int]:
"""Compute units used."""
def is_successful(self) -> bool:
"""Check if transaction was successful."""
def get_balance_changes(self) -> List[int]:
"""
Calculate balance changes for each account.
Returns:
List[int], balance change for each account (post - pre)
"""
def get_fee_payer_balance_change(self) -> int:
"""
Get balance change for fee payer (first account).
Returns:
int, fee payer balance change including fees
"""Token Balance Changes
Token-specific balance tracking and change analysis.
class UiTransactionTokenBalance:
"""
Token balance information for a transaction account.
"""
def __init__(
self,
account_index: int,
mint: str,
ui_token_amount: 'UiTokenAmount',
owner: Optional[str],
program_id: Optional[str]
):
"""
Create token balance information.
Parameters:
- account_index: int, index of account in transaction
- mint: str, token mint address
- ui_token_amount: UiTokenAmount, token amount with decimals
- owner: Optional[str], token account owner
- program_id: Optional[str], token program ID
"""
@property
def account_index(self) -> int:
"""Index of account in transaction."""
@property
def mint(self) -> str:
"""Token mint address."""
@property
def ui_token_amount(self) -> 'UiTokenAmount':
"""Token amount with decimals."""
@property
def owner(self) -> Optional[str]:
"""Token account owner."""
@property
def program_id(self) -> Optional[str]:
"""Token program ID."""
def calculate_token_balance_changes(
pre_balances: List['UiTransactionTokenBalance'],
post_balances: List['UiTransactionTokenBalance']
) -> List[tuple[str, str, float]]:
"""
Calculate token balance changes across transaction.
Parameters:
- pre_balances: List, token balances before transaction
- post_balances: List, token balances after transaction
Returns:
List[tuple[str, str, float]], (mint, owner, change_amount) for each change
"""Inner Instructions and Cross-Program Invocations
Detailed tracking of cross-program invocations and nested instruction execution.
class UiInnerInstructions:
"""
Inner instructions from cross-program invocations.
"""
def __init__(self, index: int, instructions: List['UiInstruction']):
"""
Create inner instructions container.
Parameters:
- index: int, index of parent instruction that generated these
- instructions: List[UiInstruction], list of inner instructions
"""
@property
def index(self) -> int:
"""Index of parent instruction."""
@property
def instructions(self) -> List['UiInstruction']:
"""List of inner instructions."""
class UiInstruction:
"""
UI representation of an instruction (union type).
"""
# This is a union of UiParsedInstruction and UiCompiledInstruction
class UiParsedInstruction:
"""
Union of parsed and partially decoded instructions.
"""
# This is a union of ParsedInstruction and UiPartiallyDecodedInstruction
class ParsedInstruction:
"""
Fully parsed instruction with program-specific interpretation.
"""
def __init__(
self,
program: str,
program_id: Pubkey,
parsed: dict
):
"""
Create parsed instruction.
Parameters:
- program: str, program name (e.g., "spl-token")
- program_id: Pubkey, program ID that was invoked
- parsed: dict, program-specific parsed data
"""
@property
def program(self) -> str:
"""Program name."""
@property
def program_id(self) -> Pubkey:
"""Program ID that was invoked."""
@property
def parsed(self) -> dict:
"""Program-specific parsed data."""
class UiPartiallyDecodedInstruction:
"""
Partially decoded instruction with raw accounts and data.
"""
def __init__(
self,
program_id: Pubkey,
accounts: List[Pubkey],
data: str,
stack_height: Optional[int]
):
"""
Create partially decoded instruction.
Parameters:
- program_id: Pubkey, program that was invoked
- accounts: List[Pubkey], accounts passed to instruction
- data: str, base58-encoded instruction data
- stack_height: Optional[int], call stack depth
"""
@property
def program_id(self) -> Pubkey:
"""Program that was invoked."""
@property
def accounts(self) -> List[Pubkey]:
"""Accounts passed to instruction."""
@property
def data(self) -> str:
"""Base58-encoded instruction data."""
@property
def stack_height(self) -> Optional[int]:
"""Call stack depth."""
class UiCompiledInstruction:
"""
Compiled instruction with account indexes.
"""
def __init__(self, program_id_index: int, accounts: List[int], data: str):
"""
Create compiled instruction.
Parameters:
- program_id_index: int, index of program ID in account keys
- accounts: List[int], account indexes
- data: str, base58-encoded instruction data
"""
@property
def program_id_index(self) -> int:
"""Program ID index in account keys."""
@property
def accounts(self) -> List[int]:
"""Account indexes."""
@property
def data(self) -> str:
"""Base58-encoded instruction data."""Rewards and Incentives
Epoch reward distribution and staking incentive tracking.
class Reward:
"""
Individual reward from epoch distribution.
"""
def __init__(
self,
pubkey: str,
lamports: int,
post_balance: int,
reward_type: Optional['RewardType'],
commission: Optional[int]
):
"""
Create reward information.
Parameters:
- pubkey: str, recipient account address
- lamports: int, reward amount in lamports
- post_balance: int, account balance after reward
- reward_type: Optional[RewardType], type of reward
- commission: Optional[int], validator commission (if applicable)
"""
@property
def pubkey(self) -> str:
"""Recipient account address."""
@property
def lamports(self) -> int:
"""Reward amount in lamports."""
@property
def post_balance(self) -> int:
"""Account balance after reward."""
@property
def reward_type(self) -> Optional['RewardType']:
"""Type of reward."""
@property
def commission(self) -> Optional[int]:
"""Validator commission (if applicable)."""
class RewardType:
"""Reward type enumeration."""
Fee: 'RewardType' # Transaction fees
Rent: 'RewardType' # Rent collection
Staking: 'RewardType' # Staking rewards
Voting: 'RewardType' # Voting rewards
def __str__(self) -> str:
"""String representation of reward type."""Transaction Encoding and Details
Transaction representation options and detail levels for different use cases.
class TransactionDetails:
"""Transaction detail level for responses."""
Full: 'TransactionDetails' # Include all transaction details
Accounts: 'TransactionDetails' # Include only account keys
Signatures: 'TransactionDetails' # Include only signatures
None_: 'TransactionDetails' # Exclude transaction details
class UiTransactionEncoding:
"""Transaction encoding format for responses."""
Json: 'UiTransactionEncoding' # JSON format with parsed data
JsonParsed: 'UiTransactionEncoding' # JSON with program-specific parsing
Base58: 'UiTransactionEncoding' # Base58 encoded
Base64: 'UiTransactionEncoding' # Base64 encoded
class TransactionBinaryEncoding:
"""Binary transaction encoding format."""
Base58: 'TransactionBinaryEncoding' # Base58 encoded
Base64: 'TransactionBinaryEncoding' # Base64 encodedTransaction Return Data
Program return data and execution results.
class TransactionReturnData:
"""
Return data from program execution.
"""
def __init__(self, program_id: Pubkey, data: bytes):
"""
Create transaction return data.
Parameters:
- program_id: Pubkey, program that returned the data
- data: bytes, returned data
"""
@property
def program_id(self) -> Pubkey:
"""Program that returned the data."""
@property
def data(self) -> bytes:
"""Returned data."""
def decode_string(self) -> str:
"""
Decode return data as UTF-8 string.
Returns:
str, decoded string data
Raises:
- UnicodeDecodeError: if data is not valid UTF-8
"""
def decode_json(self) -> dict:
"""
Decode return data as JSON.
Returns:
dict, parsed JSON data
Raises:
- JSONDecodeError: if data is not valid JSON
"""Address Lookup Table Support
Address lookup table metadata and loaded address information.
class UiLoadedAddresses:
"""
Addresses loaded from lookup tables in versioned transactions.
"""
def __init__(self, writable: List[str], readonly: List[str]):
"""
Create loaded addresses information.
Parameters:
- writable: List[str], writable addresses from lookup tables
- readonly: List[str], readonly addresses from lookup tables
"""
@property
def writable(self) -> List[str]:
"""Writable addresses from lookup tables."""
@property
def readonly(self) -> List[str]:
"""Readonly addresses from lookup tables."""
def total_loaded(self) -> int:
"""
Get total number of loaded addresses.
Returns:
int, total loaded addresses (writable + readonly)
"""
class UiAddressTableLookup:
"""
Address table lookup reference in UI format.
"""
def __init__(
self,
account_key: str,
writable_indexes: List[int],
readonly_indexes: List[int]
):
"""
Create address table lookup.
Parameters:
- account_key: str, lookup table account address
- writable_indexes: List[int], indexes of writable addresses
- readonly_indexes: List[int], indexes of readonly addresses
"""
@property
def account_key(self) -> str:
"""Lookup table account address."""
@property
def writable_indexes(self) -> List[int]:
"""Indexes of writable addresses."""
@property
def readonly_indexes(self) -> List[int]:
"""Indexes of readonly addresses."""Usage Examples
Transaction Status Monitoring
from solders.transaction_status import TransactionStatus, TransactionConfirmationStatus
from solders.rpc.requests import GetSignatureStatuses
# Monitor transaction confirmation
def monitor_transaction_status(signatures: List[Signature]) -> dict:
"""Monitor status of multiple transactions."""
status_request = GetSignatureStatuses(signatures)
# After RPC call, process responses
results = {}
# Example processing (would get actual response from RPC)
for i, signature in enumerate(signatures):
# Parse status response
status = TransactionStatus(
slot=100000000,
confirmations=31, # Number of confirmations
status=None, # None means success
confirmation_status=TransactionConfirmationStatus.Finalized
)
results[str(signature)] = {
'successful': status.is_successful(),
'confirmed': status.is_confirmed(),
'slot': status.slot,
'confirmations': status.confirmations,
'confirmation_level': str(status.confirmation_status)
}
return results
# Wait for transaction confirmation
def wait_for_confirmation(signature: Signature, target_confirmations: int = 31):
"""Wait for transaction to reach target confirmation level."""
import time
while True:
status_request = GetSignatureStatuses([signature])
# Get response from RPC...
# Example status check (would parse actual RPC response)
if status and status.confirmations and status.confirmations >= target_confirmations:
print(f"Transaction confirmed with {status.confirmations} confirmations")
return True
time.sleep(1) # Wait 1 second before checking againTransaction Metadata Analysis
from solders.transaction_status import UiTransactionStatusMeta
def analyze_transaction_execution(meta: UiTransactionStatusMeta):
"""Analyze transaction execution results."""
# Check success/failure
if meta.is_successful():
print("Transaction executed successfully")
else:
print(f"Transaction failed with error: {meta.err}")
# Analyze fees and balance changes
print(f"Transaction fee: {meta.fee} lamports ({meta.fee / 1e9:.9f} SOL)")
balance_changes = meta.get_balance_changes()
for i, change in enumerate(balance_changes):
if change != 0:
print(f"Account {i} balance change: {change:+,} lamports")
# Fee payer analysis
fee_payer_change = meta.get_fee_payer_balance_change()
print(f"Fee payer total change: {fee_payer_change:+,} lamports (including fees)")
# Compute unit consumption
if meta.compute_units_consumed:
print(f"Compute units used: {meta.compute_units_consumed:,}")
efficiency = (meta.compute_units_consumed / 200000) * 100 # Assuming 200k limit
print(f"Compute efficiency: {efficiency:.1f}% of limit")
# Log analysis
if meta.log_messages:
print(f"Program logs ({len(meta.log_messages)} messages):")
for log in meta.log_messages[:5]: # Show first 5 logs
print(f" {log}")
if len(meta.log_messages) > 5:
print(f" ... and {len(meta.log_messages) - 5} more")
def analyze_token_transfers(meta: UiTransactionStatusMeta):
"""Analyze token transfers in transaction."""
if not meta.pre_token_balances or not meta.post_token_balances:
print("No token balance information available")
return
# Calculate token changes
token_changes = calculate_token_balance_changes(
meta.pre_token_balances,
meta.post_token_balances
)
print(f"Token transfers ({len(token_changes)} changes):")
for mint, owner, change in token_changes:
if change != 0:
print(f" {owner}: {change:+.6f} tokens (mint: {mint})")Inner Instruction Analysis
def analyze_cross_program_invocations(meta: UiTransactionStatusMeta):
"""Analyze cross-program invocations and inner instructions."""
if not meta.inner_instructions:
print("No cross-program invocations")
return
print(f"Cross-program invocations: {len(meta.inner_instructions)} instruction groups")
for inner_group in meta.inner_instructions:
print(f"\nInstruction {inner_group.index} invoked {len(inner_group.instructions)} inner instructions:")
for inner_ix in inner_group.instructions:
if isinstance(inner_ix, ParsedInstruction):
print(f" {inner_ix.program}: {inner_ix.parsed.get('type', 'unknown')}")
elif isinstance(inner_ix, UiPartiallyDecodedInstruction):
print(f" Program {inner_ix.program_id}: {len(inner_ix.accounts)} accounts")
if inner_ix.stack_height:
print(f" Stack height: {inner_ix.stack_height}")
def trace_program_execution(meta: UiTransactionStatusMeta):
"""Trace program execution through logs and inner instructions."""
execution_trace = []
# Add main instructions (would need transaction data)
# execution_trace.append("Main instruction: ...")
# Add inner instructions
if meta.inner_instructions:
for inner_group in meta.inner_instructions:
for inner_ix in inner_group.instructions:
if isinstance(inner_ix, ParsedInstruction):
execution_trace.append(f"CPI: {inner_ix.program}")
# Correlate with logs
if meta.log_messages:
for log in meta.log_messages:
if "invoke" in log.lower():
execution_trace.append(f"Log: {log}")
print("Execution trace:")
for i, event in enumerate(execution_trace):
print(f"{i+1:2d}. {event}")Reward Distribution Analysis
def analyze_epoch_rewards(rewards: List[Reward]):
"""Analyze epoch reward distribution."""
if not rewards:
print("No rewards in this transaction")
return
total_rewards = sum(reward.lamports for reward in rewards)
print(f"Total rewards distributed: {total_rewards:,} lamports ({total_rewards / 1e9:.6f} SOL)")
# Group by reward type
by_type = {}
for reward in rewards:
reward_type = str(reward.reward_type) if reward.reward_type else "Unknown"
if reward_type not in by_type:
by_type[reward_type] = []
by_type[reward_type].append(reward)
for reward_type, type_rewards in by_type.items():
type_total = sum(r.lamports for r in type_rewards)
print(f"{reward_type}: {len(type_rewards)} recipients, {type_total:,} lamports")
# Show largest rewards
sorted_rewards = sorted(rewards, key=lambda r: r.lamports, reverse=True)
print(f"\nTop rewards:")
for reward in sorted_rewards[:5]:
print(f" {reward.pubkey}: {reward.lamports:,} lamports")
if reward.commission:
print(f" Commission: {reward.commission}%")Transaction Return Data Processing
def process_return_data(return_data: TransactionReturnData):
"""Process program return data."""
print(f"Return data from program: {return_data.program_id}")
print(f"Data length: {len(return_data.data)} bytes")
# Try to decode as string
try:
text_data = return_data.decode_string()
print(f"String data: {text_data}")
except UnicodeDecodeError:
print("Data is not valid UTF-8 text")
# Try to decode as JSON
try:
json_data = return_data.decode_json()
print(f"JSON data: {json_data}")
except:
print("Data is not valid JSON")
# Show raw bytes (first 50 bytes)
hex_data = return_data.data[:50].hex()
print(f"Raw data (hex): {hex_data}{'...' if len(return_data.data) > 50 else ''}")Address Lookup Table Analysis
def analyze_loaded_addresses(loaded_addresses: UiLoadedAddresses):
"""Analyze addresses loaded from lookup tables."""
total = loaded_addresses.total_loaded()
print(f"Loaded addresses: {total} total")
print(f" Writable: {len(loaded_addresses.writable)}")
print(f" Readonly: {len(loaded_addresses.readonly)}")
# Calculate compression savings
# Each address is 32 bytes, lookup table reference is much smaller
bytes_saved = total * 32 - (total * 1) # Simplified calculation
print(f"Approximate bytes saved: {bytes_saved}")
# Show some addresses
if loaded_addresses.writable:
print("Writable addresses:")
for addr in loaded_addresses.writable[:3]:
print(f" {addr}")
if len(loaded_addresses.writable) > 3:
print(f" ... and {len(loaded_addresses.writable) - 3} more")
def calculate_transaction_efficiency(meta: UiTransactionStatusMeta):
"""Calculate transaction efficiency metrics."""
metrics = {
'fee_per_account': meta.fee / len(meta.pre_balances) if meta.pre_balances else 0,
'accounts_modified': sum(1 for change in meta.get_balance_changes() if change != 0),
'inner_instruction_count': 0,
'log_message_count': len(meta.log_messages) if meta.log_messages else 0
}
if meta.inner_instructions:
metrics['inner_instruction_count'] = sum(
len(group.instructions) for group in meta.inner_instructions
)
if meta.compute_units_consumed:
metrics['compute_efficiency'] = meta.compute_units_consumed / 200000 # Assuming 200k limit
# Fee efficiency
if meta.compute_units_consumed:
metrics['fee_per_cu'] = meta.fee / meta.compute_units_consumed
return metricsError Analysis
Transaction Error Types
# Union types for transaction errors
TransactionErrorType = Union[
TransactionErrorFieldless,
TransactionErrorInstructionError,
TransactionErrorDuplicateInstruction,
TransactionErrorInsufficientFundsForRent,
TransactionErrorProgramExecutionTemporarilyRestricted
]
InstructionErrorType = Union[
InstructionErrorFieldless,
InstructionErrorCustom,
InstructionErrorBorshIO
]Error Processing
def analyze_transaction_error(error: TransactionErrorType):
"""Analyze and explain transaction error."""
if isinstance(error, TransactionErrorInstructionError):
instruction_index = error.instruction_index
instruction_error = error.error
print(f"Instruction {instruction_index} failed:")
if isinstance(instruction_error, InstructionErrorCustom):
print(f" Custom error code: {instruction_error.code}")
elif isinstance(instruction_error, InstructionErrorFieldless):
print(f" Error type: {instruction_error}")
else:
print(f" Error: {instruction_error}")
elif isinstance(error, TransactionErrorInsufficientFundsForRent):
account_index = error.account_index
print(f"Account {account_index} has insufficient funds for rent")
else:
print(f"Transaction error: {error}")
def suggest_error_fixes(error: TransactionErrorType) -> List[str]:
"""Suggest fixes for common transaction errors."""
suggestions = []
if "InsufficientFunds" in str(error):
suggestions.append("Add more SOL to the fee payer account")
suggestions.append("Reduce transaction complexity to lower fees")
elif "AccountNotFound" in str(error):
suggestions.append("Ensure all referenced accounts exist")
suggestions.append("Create missing accounts before transaction")
elif "InvalidAccountData" in str(error):
suggestions.append("Check account data format and size")
suggestions.append("Ensure account is owned by correct program")
elif "CustomError" in str(error):
suggestions.append("Check program documentation for error codes")
suggestions.append("Verify instruction parameters")
return suggestionsPerformance Monitoring
Transaction Performance Metrics
def calculate_performance_metrics(meta: UiTransactionStatusMeta) -> dict:
"""Calculate transaction performance metrics."""
metrics = {}
# Cost efficiency
metrics['cost_per_account'] = meta.fee / len(meta.pre_balances)
# Compute efficiency
if meta.compute_units_consumed:
metrics['compute_units_used'] = meta.compute_units_consumed
metrics['compute_utilization'] = meta.compute_units_consumed / 1400000 # Max CU
metrics['cost_per_compute_unit'] = meta.fee / meta.compute_units_consumed
# Complexity metrics
metrics['accounts_involved'] = len(meta.pre_balances)
metrics['accounts_modified'] = len([c for c in meta.get_balance_changes() if c != 0])
if meta.inner_instructions:
metrics['cpi_count'] = sum(len(group.instructions) for group in meta.inner_instructions)
metrics['max_stack_depth'] = max(
ix.stack_height or 1
for group in meta.inner_instructions
for ix in group.instructions
if hasattr(ix, 'stack_height') and ix.stack_height
)
return metrics