michaelhly
Best way to send solana or an spl wih latest version?
I am trying to send solana or tokens (depending on the contract_address parameter). The problem is, it doesnt take the priority fee into account.
I am using 0.27.0 because I was initially having problems sending transactions with the new api (Invalid base58 key when trying to send transactions).
I put it into chatgpt to fix it, but it made it worse.
Here it is:
`from spl.token.instructions import transfer_checked, create_associated_token_account, get_associated_token_address, TransferCheckedParams from spl.token.constants import TOKEN_PROGRAM_ID from solana.rpc.api import Client from solana.rpc.types import TxOpts from solana.keypair import Keypair from solana.publickey import PublicKey from solana.transaction import Transaction from solana.system_program import transfer, TransferParams import base58 import struct import time import random
List of multiple RPC endpoints for load balancing
RPC_ENDPOINTS = [ "https://magical-small-sun.solana-mainnet.quiknode.pro/1e7f2b948b444d55509c80ba0338c640e879a74a" ]
Rate limiter parameters
MAX_REQUESTS_PER_SECOND = 10 # Adjust based on RPC rate limits request_count = 0 start_time = time.time()
def get_random_rpc_endpoint(): """Select a random RPC endpoint for load balancing.""" return random.choice(RPC_ENDPOINTS)
def rate_limit(): """Implements rate limiting to prevent exceeding RPC rate limits.""" global request_count, start_time request_count += 1 elapsed_time = time.time() - start_time if elapsed_time < 1.0: # Check if within the same second if request_count > MAX_REQUESTS_PER_SECOND: time.sleep(1.0 - elapsed_time) # Wait until 1 second has passed else: start_time = time.time() request_count = 0
def send_transaction_with_retry(client, transaction, keypair, opts, max_retries=10): """ Sends a transaction with retry mechanism to handle rate limiting (429 errors) and unconfirmed transactions.
Parameters:
client (Client): Solana client instance.
transaction (Transaction): The transaction object to be sent.
keypair (Keypair): The sender's keypair for signing.
opts (TxOpts): Transaction options.
max_retries (int): Maximum number of retry attempts.
Returns:
str: Transaction signature.
"""
retries = 0
while retries < max_retries:
try:
tx_signature = client.send_transaction(transaction, keypair, opts=opts)
print(f"Transaction sent: {tx_signature.value}")
return tx_signature.value # Access the value attribute for the transaction signature
except Exception as e:
if "429" in str(e) or "Too Many Requests" in str(e):
print(f"Rate limit hit. Retrying... ({retries + 1}/{max_retries})")
retries += 1
time.sleep(2 ** retries) # Exponential backoff
elif "UnconfirmedTxError" in str(e):
print(f"Transaction unconfirmed. Retrying... ({retries + 1}/{max_retries})")
retries += 1
time.sleep(2 ** retries)
else:
print(f"An error occurred: {e}")
print("Max retries exceeded. Could not send transaction.")
return None
def confirm_transaction(client, tx_signature, commitment="finalized", max_retries=10): """ Manually checks the confirmation status of a transaction using get_signature_statuses.
Parameters:
client (Client): Solana client instance.
tx_signature (str): Transaction signature to check.
commitment (str): Desired commitment level.
max_retries (int): Maximum number of retries.
Returns:
bool: True if the transaction is confirmed, False otherwise.
"""
retries = 0
while retries < max_retries:
try:
response = client.get_signature_statuses([tx_signature])
status = response.value[0]
if status is not None:
print(f"Transaction {tx_signature} confirmed.")
return True
else:
print(f"Waiting for transaction {tx_signature} to be confirmed... ({retries + 1}/{max_retries})")
retries += 1
time.sleep(2 ** retries) # Exponential backoff
except Exception as e:
print(f"An error occurred while checking confirmation: {e}")
retries += 1
time.sleep(2 ** retries)
print(f"Unable to confirm transaction {tx_signature}")
return False
def send_sol_transaction(senderPrivateKey, recipientAddress, contractAddress, amount, priorityfee): """ Sends Solana or a custom SPL token from the sender to the recipient.
Parameters:
senderPrivateKey (str): Base58-encoded private key of the sender.
recipientAddress (str): Solana address of the recipient.
contractAddress (str): SPL token mint address for custom tokens, or 'SOL' for native SOL.
amount (float): Amount of SOL or SPL tokens to send.
priorityfee (float): Priority fee in SOL (will be converted to lamports).
Returns:
str: Transaction signature.
"""
# Convert priority fee from SOL to lamports
priority_fee_lamports = int(priorityfee * 1e9) # 1 SOL = 1,000,000,000 lamports
# Select a random RPC endpoint and initialize Solana client
solana_client = Client(get_random_rpc_endpoint())
# Decode sender's private key and generate Keypair
sender_keypair = Keypair.from_secret_key(base58.b58decode(senderPrivateKey))
sender_pubkey = sender_keypair.public_key
# Convert recipient address to PublicKey
recipient_pubkey = PublicKey(recipientAddress)
# Create a transaction object
transaction = Transaction()
if contractAddress == 'SOL':
# Native SOL transfer using SystemProgram
lamports = int(amount * 1e9) # Convert SOL to lamports
transaction.add(
transfer(
TransferParams(
from_pubkey=sender_pubkey,
to_pubkey=recipient_pubkey,
lamports=lamports
)
)
)
else:
# SPL Token transfer
mint_address = PublicKey(contractAddress)
# Get or create associated token accounts
sender_token_address = get_associated_token_address(sender_pubkey, mint_address)
recipient_token_address = get_associated_token_address(recipient_pubkey, mint_address)
# Debug: Print sender's token account address
print(f"Sender token address: {sender_token_address}")
# Debug: Check and print account info
account_info = solana_client.get_account_info(sender_token_address).value
print(f"Account info: {account_info}")
# Ensure the recipient has an associated token account
recipient_account_info = solana_client.get_account_info(recipient_token_address).value
if recipient_account_info is None:
transaction.add(
create_associated_token_account(
payer=sender_pubkey,
owner=recipient_pubkey,
mint=mint_address
)
)
# Check sender's token balance before proceeding
balance_response = solana_client.get_token_account_balance(sender_token_address)
balance = balance_response.value.ui_amount
print(f"Sender token balance: {balance}")
if balance < amount:
print(f"Insufficient funds: Sender only has {balance} tokens, but {amount} tokens are required.")
return None
# Query mint information to get raw data
mint_info = solana_client.get_account_info(mint_address).value
mint_data = mint_info.data
# Decode the number of decimals (1 byte at offset 44)
decimals = struct.unpack_from("B", mint_data, offset=44)[0]
print(f"Token decimals: {decimals}")
# Add SPL token transfer instruction
transaction.add(
transfer_checked(
TransferCheckedParams(
program_id=TOKEN_PROGRAM_ID,
source=sender_token_address,
mint=mint_address,
dest=recipient_token_address,
owner=sender_pubkey,
amount=int(amount * (10 ** decimals)),
decimals=decimals
)
)
)
recent_blockhash_resp = solana_client.get_latest_blockhash()
recent_blockhash = recent_blockhash_resp.value.blockhash
transaction.recent_blockhash = str(recent_blockhash)
transaction.fee_payer = sender_pubkey
# Set transaction options without unsupported parameters
tx_options = TxOpts(
skip_confirmation=False,
preflight_commitment="processed" # Use lower-level commitment for faster inclusion
)
# Apply rate limiting before sending the transaction
rate_limit()
# Send transaction with retry mechanism
tx_signature = send_transaction_with_retry(solana_client, transaction, sender_keypair, tx_options)
# Manually check transaction confirmation
if tx_signature and not confirm_transaction(solana_client, tx_signature, commitment="finalized"):
print(f"Unable to confirm transaction {tx_signature}")
# Print the transaction signature for confirmation
return tx_signature
Example usage
if name == "main": sender_private_key = "PRIV" # Replace with actual sender private key recipient_address = "RECPUB" contract_address = "SOL" # Example SPL token mint address amount = 0.0001 # Amount to send (in SOL) priority_fee = 0.01 # Priority fee in SOL (will be converted to lamports)
result = send_sol_transaction(sender_private_key, recipient_address, contract_address, amount, priority_fee)
print(f"Transaction signature: {result}")
`
i upgraded back to 0.32.2. Is there an example of sending sol or SPL tokens with it?
Example to transfer SPL tokens: https://github.com/michaelhly/solana-py/blob/37cdade3a26d592baa21dd1b76ad0720e73cf7f2/tests/integration/test_async_token_client.py#L145
Example to transfer SOL: https://github.com/michaelhly/solana-py/blob/37cdade3a26d592baa21dd1b76ad0720e73cf7f2/tests/integration/test_http_client.py#L63
Thanks i got it working perfectly sending solana, but with SPL tokens it keeps saying I do not have balance, despite i do. I checked in debugs, it shows that the correct token account is being called, it shows the right balance, but when it sends it says it has 100x less:
`import asyncio from solders.keypair import Keypair from solders.pubkey import Pubkey from solana.rpc.async_api import AsyncClient from spl.token.async_client import AsyncToken from spl.token.constants import TOKEN_PROGRAM_ID from spl.token.instructions import get_associated_token_address import solders.system_program as sp from solders.transaction import Transaction from solders.message import Message from solana.rpc.types import TxOpts import base58
OPTS = TxOpts(skip_confirmation=False, preflight_commitment="processed") RCP_CLI = "RPC_CLIENT"
async def get_or_create_associated_token_account(client, payer_keypair, owner_pubkey, mint_pubkey): """ Get or create the associated token account for a given mint and owner.
Args:
client (AsyncClient): Solana RPC client.
payer_keypair (Keypair): Keypair of the payer (who pays for account creation).
owner_pubkey (Pubkey): Public key of the token account owner.
mint_pubkey (Pubkey): Public key of the mint (SPL token).
Returns:
Pubkey: Associated token account address.
"""
associated_account = get_associated_token_address(owner_pubkey, mint_pubkey)
# Check if the associated token account already exists
resp = await client.get_account_info(associated_account)
if resp.value is not None:
print(f"Associated token account already exists: {associated_account}")
return associated_account
# Create the associated token account if it doesn't exist
print(f"Creating associated token account: {associated_account}")
transaction = Transaction()
transaction.add(
AsyncToken.create_associated_token_account_instruction(
payer=payer_keypair.pubkey(),
owner=owner_pubkey,
mint=mint_pubkey,
associated_account=associated_account,
)
)
# Send the transaction
tx_resp = await client.send_transaction(transaction, payer_keypair)
await client.confirm_transaction(tx_resp.value, commitment="confirmed")
print(f"Created associated token account with signature: {tx_resp.value}")
return associated_account
async def send_sol_transaction( sender_private_key: str, recipient_address: str, contract_address: str, amount: float, priority_fee: float = 0.0, ): """ Sends SOL or SPL tokens using AsyncToken and AsyncClient.
Args:
sender_private_key (str): Base58-encoded private key of the sender.
recipient_address (str): Base58-encoded public key of the recipient.
contract_address (str): Base58-encoded mint address for SPL tokens or None for SOL transfer.
amount (float): Amount to transfer in SOL (if contract_address is None) or token units (if SPL tokens).
priority_fee (float): Priority fee in SOL.
Returns:
str: Transaction signature.
"""
# Decode the Base58 private key
decoded_key = base58.b58decode(sender_private_key)
# Ensure the key is 64 bytes (private key + public key)
if len(decoded_key) != 64:
raise ValueError("Invalid private key length. Expected 64 bytes.")
sender_keypair = Keypair.from_bytes(decoded_key)
sender_pubkey = sender_keypair.pubkey()
recipient_pubkey = Pubkey.from_string(recipient_address)
async with AsyncClient(RCP_CLI) as client:
if contract_address:
print("SPL token transfer logic triggered")
# Handle SPL token transfer
mint_address = Pubkey.from_string(contract_address)
token_client = AsyncToken(
conn=client,
pubkey=mint_address,
program_id=TOKEN_PROGRAM_ID,
payer=sender_keypair,
)
# Get or create associated token accounts for sender and recipient
sender_token_account = await get_or_create_associated_token_account(
client, sender_keypair, sender_pubkey, mint_address
)
recipient_token_account = await get_or_create_associated_token_account(
client, sender_keypair, recipient_pubkey, mint_address
)
# Debug: Output token accounts and balances
sender_balance_resp = await token_client.get_balance(sender_token_account)
print(f"Sender token account: {sender_token_account}")
print(f"Sender token balance: {sender_balance_resp.value.ui_amount} tokens")
recipient_balance_resp = await token_client.get_balance(recipient_token_account)
print(f"Recipient token account: {recipient_token_account}")
print(f"Recipient token balance: {recipient_balance_resp.value.ui_amount} tokens")
# Ensure sufficient balance before proceeding
sender_balance = sender_balance_resp.value.ui_amount # Use human-readable balance directly
if sender_balance < amount:
raise Exception(
f"Insufficient funds: Sender only has {sender_balance} tokens, but {amount} tokens are required."
)
# Convert amount to smallest token unit (assuming 9 decimals)
amount_in_smallest_unit = int(amount * (10 ** 9))
# Transfer SPL tokens
print(f"Transferring {amount} SPL tokens...")
transfer_resp = await token_client.transfer(
source=sender_token_account,
dest=recipient_token_account,
owner=sender_keypair,
amount=amount_in_smallest_unit,
opts=OPTS,
)
# Confirm the transaction
await client.confirm_transaction(transfer_resp.value, commitment="confirmed")
print(f"SPL Token Transaction successful with signature: {transfer_resp.value}")
return transfer_resp.value
else:
print("SOL transfer logic triggered")
# Handle SOL transfer
blockhash_resp = await client.get_latest_blockhash()
blockhash = blockhash_resp.value.blockhash
# Convert amount and priority fee to lamports
amount_in_lamports = int(amount * 1_000_000_000)
priority_fee_in_lamports = int(priority_fee * 1_000_000_000)
# Add priority fee to the total transfer amount
total_lamports = amount_in_lamports + priority_fee_in_lamports
# Create transfer instruction
transfer_ix = sp.transfer(
sp.TransferParams(from_pubkey=sender_pubkey, to_pubkey=recipient_pubkey, lamports=total_lamports)
)
# Create message and transaction
msg = Message.new_with_blockhash([transfer_ix], sender_pubkey, blockhash)
transaction = Transaction([sender_keypair], msg, blockhash)
# Simulate the transaction
sim_resp = await client.simulate_transaction(transaction)
print(f"Simulation response: {sim_resp}")
# Check for simulation errors
if sim_resp.value.err:
raise Exception(f"Transaction simulation failed: {sim_resp.value.err}")
# Send the transaction
tx_resp = await client.send_transaction(transaction)
print(f"SOL Transaction successful with signature: {tx_resp.value}")
# Confirm the transaction
await client.confirm_transaction(tx_resp.value, commitment="confirmed")
return tx_resp.value
Example usage
if name == "main": async def main(): # Example parameters sender_private_key = "PK" recipient_address = "PUB" # Replace with actual recipient pubkey contract_address = "CEhFvMotKm3zucKUBEHVvTaxQ4e9QVPaAjSfkzFLpump" # Replace with mint address for SPL tokens or keep None for SOL transfer amount = 10 # Amount in SOL or token units priority_fee = 0.005 # Priority fee in SOL
try:
signature = await send_sol_transaction(
sender_private_key, recipient_address, contract_address, amount, priority_fee
)
print(f"Transaction successful with signature: {signature}")
except Exception as e:
print(f"Error occurred: {e}")
asyncio.run(main())
`
@michaelhly I got both functions working, my main issue now is that i cannot get the SPL tokens to take a SOL fee.