0

I'm working on a script to update a DB when a transaction with a specific memo is sent to a specific address. I am able to get confirmed transaction hashes for the address, but I am struggling to find a way to collect the SOL amount, and Memo data for each transaction. Any guidance or help here would be much appreciated

import asyncio
import json
from solana.rpc.async_api import AsyncClient
from solders.signature import Signature
from solders.pubkey import Pubkey as PublicKey
SOLANA_ADDRESS = ""
async def fetch_and_process_transactions():
    async with AsyncClient("", commitment="confirmed") as client:
        pubkey = PublicKey.from_string(SOLANA_ADDRESS)
        print(f"Checking transactions for address: {SOLANA_ADDRESS}")
        response = await client.get_signatures_for_address(pubkey, limit=5)
        transactions = response.value
        if transactions:
            print(f"Found {len(transactions)} transactions.")
            for tx in transactions:
                signature_str = str(tx.signature)
                signature = Signature.from_string(signature_str)
                tx_detail_response = await client.get_transaction(signature, encoding="jsonParsed")
                tx_detail = tx_detail_response.value
                print(f"Details for transaction {signature_str}:")
                try:
                    # Simplified detail extraction for demonstration
                    slot = tx_detail.get('slot', 'N/A')
                    for instruction in tx_detail['transaction']['message']['instructions']:
                        program_id = instruction.get('programId', 'N/A')
                        # Convert parsed instruction to string if necessary
                        instruction_data = json.dumps(instruction.get('parsed', instruction.get('data', 'N/A')))
                        print(f"\tSlot: {slot}, Program ID: {program_id}, Instruction Data: {instruction_data}")
                except Exception as e:
                    print(f"Error parsing transaction details: {e}")
asyncio.run(fetch_and_process_transactions())

1 Answer 1

0

To get SOL amount and memo data for each transaction, you need to parse the transaction instructions and metadata

import asyncio
import base64
from solana.rpc.async_api import AsyncClient
from solders.signature import Signature
from solders.pubkey import Pubkey as PublicKey

SOLANA_ADDRESS = ""
MEMO_PROGRAM_ID = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr"

async def fetch_and_process_transactions():
    async with AsyncClient("", commitment="confirmed") as client:
        pubkey = PublicKey.from_string(SOLANA_ADDRESS)
        print(f"Checking transactions for address: {SOLANA_ADDRESS}")
        response = await client.get_signatures_for_address(pubkey, limit=5)
        transactions = response.value
        if transactions:
            print(f"Found {len(transactions)} transactions.")
            for tx in transactions:
                signature_str = str(tx.signature)
                signature = Signature.from_string(signature_str)
                tx_detail_response = await client.get_transaction(signature, encoding="jsonParsed")
                tx_detail = tx_detail_response.value
                print(f"Details for transaction {signature_str}:")
                try:
                    slot = tx_detail.get('slot', 'N/A')
                    for instruction in tx_detail['transaction']['message']['instructions']:
                        program_id = instruction.get('programId', 'N/A')
                        if program_id == MEMO_PROGRAM_ID:
                            memo_data = base64.b64decode(instruction['data'][0]).decode('utf-8')
                            print(f"\tSlot: {slot}, Memo: {memo_data}")
                        elif program_id == "11111111111111111111111111111111":  # System Program
                            for account_metadata in instruction['accounts']:
                                if account_metadata['isSigner'] and account_metadata['isWritable']:
                                    sol_amount = account_metadata['lamports'] / 1e9
                                    print(f"\tSlot: {slot}, SOL Amount: {sol_amount}")
                except Exception as e:
                    print(f"Error parsing transaction details: {e}")

asyncio.run(fetch_and_process_transactions())

What changes I added here is: defined a constant MEMO_PROGRAM_ID with the Memo program ID then Inside the loop that iterates over instructions, it checks if the programId matches the Memo program ID. If it does, it decodes the data field (which contains the memo) from base64 and prints the memo. It also checks if the programId is the System Program ID (11111111111111111111111111111111). If it is, it loops over the accounts array and looks for an account that is both isSigner and isWritable. This account represents the source account from which SOL was transferred.

For source account, it extracts the lamports value from the and converts it to SOL by dividing by 1e9.

keep in mind that I am only assumed that there is only one memo instruction and one SOL transfer instruction per transaction. If a transaction includes multiple instructions of the same type, you need to modify the code according to your need.

If this works fine mark this question as accepted.

2
  • I can't seem to get past the fact that the response(transaction detail) from the Solana Client is in a Rust format. Is there a simply way to extract the data from it? by Rust format I mean that the JSON is enclosed in EncodedTransactionWithStatusMeta(EncodedTransactionWithStatusMeta Commented Apr 2 at 2:29
  • Would you like to share response screenshot because it already parsing what you wanted, use imgbb and share the link in comment. Commented Apr 2 at 8:22

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.