Use FortiBlox Nexus RPC with Solana Python SDK for backend services and data analysis

Solana Python SDK

The Solana Python SDK provides both synchronous and asynchronous interfaces for interacting with the Solana blockchain. This guide shows you how to use it with FortiBlox Nexus RPC.

Installation

Install the required Python packages:

pip install solana solders

Optional dependencies for enhanced functionality:

pip install python-dotenv aiohttp

Package Overview

solana: Main SDK with sync and async clients
solders: High-performance Rust-based types (Keypair, PublicKey, Transaction)
python-dotenv: Load environment variables
aiohttp: Enhanced async HTTP support

Authentication Setup

Environment Variables

Create a .env file:

FORTIBLOX_API_KEY=your-api-key-here
FORTIBLOX_RPC_URL=https://nexus.fortiblox.com/rpc

Load in your Python application:

import os
from dotenv import load_dotenv

load_dotenv()

FORTIBLOX_API_KEY = os.getenv('FORTIBLOX_API_KEY')
FORTIBLOX_RPC_URL = os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc')

Client Configuration

Configure the client with FortiBlox endpoint and API key:

from solana.rpc.api import Client
from solana.rpc.commitment import Confirmed

# Synchronous client
client = Client(
    endpoint=FORTIBLOX_RPC_URL,
    extra_headers={'X-API-Key': FORTIBLOX_API_KEY},
    commitment=Confirmed
)

For async operations:

from solana.rpc.async_api import AsyncClient

# Asynchronous client
async_client = AsyncClient(
    endpoint=FORTIBLOX_RPC_URL,
    extra_headers={'X-API-Key': FORTIBLOX_API_KEY},
    commitment=Confirmed
)

Code Examples

1. Get Account Balance

Fetch SOL balance of any account:

from solana.rpc.api import Client
from solders.pubkey import Pubkey
import os

def get_balance(address: str) -> int:
    """Get SOL balance for an address."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        pubkey = Pubkey.from_string(address)
        response = client.get_balance(pubkey)

        if response.value is not None:
            lamports = response.value
            sol = lamports / 1_000_000_000

            print(f"Address: {address}")
            print(f"Balance: {sol:.9f} SOL")
            print(f"Lamports: {lamports}")

            return lamports
        else:
            print("Failed to fetch balance")
            return 0

    except Exception as e:
        print(f"Error fetching balance: {e}")
        raise

# Example usage
if __name__ == "__main__":
    get_balance("7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh")

2. Get Account Info

Retrieve detailed account information:

from solana.rpc.api import Client
from solders.pubkey import Pubkey
import os

def get_account_info(address: str):
    """Get detailed account information."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        pubkey = Pubkey.from_string(address)
        response = client.get_account_info(pubkey)

        if response.value is None:
            print("Account not found")
            return None

        account = response.value

        print("Account Info:")
        print(f"- Lamports: {account.lamports}")
        print(f"- Owner: {account.owner}")
        print(f"- Executable: {account.executable}")
        print(f"- Rent Epoch: {account.rent_epoch}")
        print(f"- Data Length: {len(account.data)}")

        return account

    except Exception as e:
        print(f"Error fetching account info: {e}")
        raise

# Example usage - Token Program
if __name__ == "__main__":
    get_account_info("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")

3. Send Transaction

Send a SOL transfer transaction:

from solana.rpc.api import Client
from solana.rpc.commitment import Confirmed
from solana.transaction import Transaction
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
import os

def send_sol(from_keypair: Keypair, to_address: str, amount_sol: float) -> str:
    """Send SOL from one account to another."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')},
            commitment=Confirmed
        )

        to_pubkey = Pubkey.from_string(to_address)
        lamports = int(amount_sol * 1_000_000_000)

        # Create transfer instruction
        transfer_ix = transfer(
            TransferParams(
                from_pubkey=from_keypair.pubkey(),
                to_pubkey=to_pubkey,
                lamports=lamports
            )
        )

        # Get recent blockhash
        recent_blockhash = client.get_latest_blockhash().value.blockhash

        # Create and sign transaction
        transaction = Transaction(
            recent_blockhash=recent_blockhash,
            fee_payer=from_keypair.pubkey()
        )
        transaction.add(transfer_ix)
        transaction.sign(from_keypair)

        # Send transaction
        print("Sending transaction...")
        response = client.send_transaction(
            transaction,
            from_keypair,
            opts={'preflight_commitment': Confirmed}
        )

        signature = response.value
        print(f"Transaction successful!")
        print(f"Signature: {signature}")
        print(f"Explorer: https://explorer.solana.com/tx/{signature}")

        return str(signature)

    except Exception as e:
        print(f"Error sending transaction: {e}")
        raise

# Example usage
# keypair = Keypair.from_base58_string("YOUR_PRIVATE_KEY")
# send_sol(keypair, "7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh", 0.1)

4. Get Recent Transactions

Fetch transaction history for an account:

from solana.rpc.api import Client
from solders.pubkey import Pubkey
from datetime import datetime
import os

def get_recent_transactions(address: str, limit: int = 10):
    """Get recent transactions for an address."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        pubkey = Pubkey.from_string(address)

        # Get signatures
        response = client.get_signatures_for_address(pubkey, limit=limit)

        signatures = response.value
        print(f"Found {len(signatures)} recent transactions:\n")

        for sig_info in signatures:
            print("---")
            print(f"Signature: {sig_info.signature}")
            print(f"Slot: {sig_info.slot}")

            if sig_info.block_time:
                timestamp = datetime.fromtimestamp(sig_info.block_time)
                print(f"Block Time: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")

            print(f"Status: {'Failed' if sig_info.err else 'Success'}")

            if sig_info.memo:
                print(f"Memo: {sig_info.memo}")

            print()

        return signatures

    except Exception as e:
        print(f"Error fetching transactions: {e}")
        raise

# Example usage
if __name__ == "__main__":
    get_recent_transactions("7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh", 5)

5. Get Program Accounts

Query all accounts owned by a specific program:

from solana.rpc.api import Client
from solders.pubkey import Pubkey
import os

def get_program_accounts(program_id: str):
    """Get all accounts owned by a program."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        program_pubkey = Pubkey.from_string(program_id)

        print(f"Fetching accounts for program: {program_id}")

        response = client.get_program_accounts(program_pubkey)
        accounts = response.value

        print(f"Found {len(accounts)} accounts\n")

        # Show first 5 accounts
        for i, account_data in enumerate(accounts[:5]):
            print(f"Account {i + 1}:")
            print(f"- Address: {account_data.pubkey}")
            print(f"- Lamports: {account_data.account.lamports}")
            print(f"- Data Length: {len(account_data.account.data)}")
            print(f"- Owner: {account_data.account.owner}")
            print()

        return accounts

    except Exception as e:
        print(f"Error fetching program accounts: {e}")
        raise

# Example usage - Token Program
if __name__ == "__main__":
    get_program_accounts("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")

6. Get Block Information

Fetch details about a specific block:

from solana.rpc.api import Client
from datetime import datetime
import os

def get_block_info(slot: int = None):
    """Get information about a specific block."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        # Get latest slot if not provided
        if slot is None:
            slot = client.get_slot().value
            print(f"Using latest slot: {slot}")

        # Get block
        response = client.get_block(slot)

        if response.value is None:
            print("Block not found or not finalized yet")
            return None

        block = response.value

        print("\nBlock Information:")
        print(f"- Blockhash: {block.blockhash}")
        print(f"- Parent Slot: {block.parent_slot}")

        if block.block_time:
            timestamp = datetime.fromtimestamp(block.block_time)
            print(f"- Block Time: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")

        print(f"- Transactions: {len(block.transactions)}")
        print(f"- Block Height: {block.block_height}")

        return block

    except Exception as e:
        print(f"Error fetching block: {e}")
        raise

# Example usage
if __name__ == "__main__":
    get_block_info()

7. Monitor Multiple Accounts

Check balances for multiple accounts efficiently:

from solana.rpc.api import Client
from solders.pubkey import Pubkey
from typing import List, Dict
import os

def get_multiple_balances(addresses: List[str]) -> Dict[str, int]:
    """Get balances for multiple accounts efficiently."""
    try:
        client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

        pubkeys = [Pubkey.from_string(addr) for addr in addresses]

        # Get multiple accounts in one request
        response = client.get_multiple_accounts(pubkeys)

        balances = {}

        print("Account Balances:\n")
        for i, (address, account_info) in enumerate(zip(addresses, response.value)):
            if account_info:
                lamports = account_info.lamports
                sol = lamports / 1_000_000_000
                balances[address] = lamports

                print(f"{i + 1}. {address}")
                print(f"   Balance: {sol:.9f} SOL")
                print()
            else:
                print(f"{i + 1}. {address}")
                print(f"   Account not found")
                print()
                balances[address] = 0

        return balances

    except Exception as e:
        print(f"Error fetching balances: {e}")
        raise

# Example usage
if __name__ == "__main__":
    addresses = [
        "7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh",
        "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
    ]
    get_multiple_balances(addresses)

8. Async Operations

Use async/await for concurrent operations:

import asyncio
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey
import os

async def async_get_balance(address: str) -> int:
    """Async function to get balance."""
    async with AsyncClient(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    ) as client:
        pubkey = Pubkey.from_string(address)
        response = await client.get_balance(pubkey)
        return response.value

async def async_get_multiple_balances(addresses: list[str]):
    """Fetch multiple balances concurrently."""
    print("Fetching balances concurrently...\n")

    # Create tasks for concurrent execution
    tasks = [async_get_balance(addr) for addr in addresses]

    # Wait for all tasks to complete
    balances = await asyncio.gather(*tasks)

    # Display results
    for address, lamports in zip(addresses, balances):
        sol = lamports / 1_000_000_000
        print(f"{address}")
        print(f"Balance: {sol:.9f} SOL\n")

    return dict(zip(addresses, balances))

# Example usage
if __name__ == "__main__":
    addresses = [
        "7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh",
        "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA",
    ]

    asyncio.run(async_get_multiple_balances(addresses))

Async Patterns

Context Manager Pattern

Use context managers for automatic cleanup:

import asyncio
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey
import os

async def fetch_data_with_context():
    """Use async context manager for automatic cleanup."""
    async with AsyncClient(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    ) as client:
        # Client is automatically closed when exiting context
        slot = await client.get_slot()
        print(f"Current slot: {slot.value}")

        health = await client.get_health()
        print(f"Node health: {health}")

asyncio.run(fetch_data_with_context())

Concurrent Data Fetching

Fetch multiple pieces of data concurrently:

import asyncio
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey
import os

async def fetch_account_data(address: str):
    """Fetch comprehensive account data concurrently."""
    async with AsyncClient(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    ) as client:
        pubkey = Pubkey.from_string(address)

        # Fetch multiple pieces of data concurrently
        balance_task = client.get_balance(pubkey)
        account_task = client.get_account_info(pubkey)
        signatures_task = client.get_signatures_for_address(pubkey, limit=5)

        balance, account, signatures = await asyncio.gather(
            balance_task,
            account_task,
            signatures_task
        )

        print(f"Address: {address}")
        print(f"Balance: {balance.value / 1_000_000_000:.9f} SOL")
        print(f"Data Length: {len(account.value.data) if account.value else 0}")
        print(f"Recent Transactions: {len(signatures.value)}")

        return {
            'balance': balance.value,
            'account': account.value,
            'signatures': signatures.value
        }

# Example usage
asyncio.run(fetch_account_data("7EqQdEUXxogiCVXpZUyS6b8Z3Z1dkrKR1A5Q8cPcGvQh"))

Async Transaction Sending

Send transactions asynchronously:

import asyncio
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
from solana.transaction import Transaction
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
import os

async def async_send_sol(from_keypair: Keypair, to_address: str, amount_sol: float):
    """Send SOL asynchronously."""
    async with AsyncClient(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')},
        commitment=Confirmed
    ) as client:
        to_pubkey = Pubkey.from_string(to_address)
        lamports = int(amount_sol * 1_000_000_000)

        # Create transfer instruction
        transfer_ix = transfer(
            TransferParams(
                from_pubkey=from_keypair.pubkey(),
                to_pubkey=to_pubkey,
                lamports=lamports
            )
        )

        # Get recent blockhash
        recent_blockhash = (await client.get_latest_blockhash()).value.blockhash

        # Create and sign transaction
        transaction = Transaction(
            recent_blockhash=recent_blockhash,
            fee_payer=from_keypair.pubkey()
        )
        transaction.add(transfer_ix)
        transaction.sign(from_keypair)

        # Send transaction
        print("Sending transaction...")
        response = await client.send_transaction(
            transaction,
            from_keypair,
            opts={'preflight_commitment': Confirmed}
        )

        signature = response.value
        print(f"Transaction sent: {signature}")

        # Wait for confirmation
        await client.confirm_transaction(signature, Confirmed)
        print(f"Transaction confirmed!")

        return str(signature)

Error Handling

Implement robust error handling:

from solana.rpc.api import Client
from solana.rpc.core import RPCException
from solders.pubkey import Pubkey
import os
import time

def robust_get_balance(address: str, max_retries: int = 3) -> int:
    """Get balance with retry logic and error handling."""
    client = Client(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    )

    for attempt in range(max_retries):
        try:
            pubkey = Pubkey.from_string(address)
            response = client.get_balance(pubkey)

            if response.value is not None:
                return response.value

        except ValueError as e:
            print(f"Invalid address: {e}")
            return 0

        except RPCException as e:
            if '429' in str(e):
                print("Rate limit exceeded. Waiting before retry...")
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            elif '401' in str(e) or '403' in str(e):
                print("Authentication failed. Check your API key.")
                return 0
            else:
                print(f"RPC error: {e}")
                if attempt == max_retries - 1:
                    raise

        except Exception as e:
            print(f"Unexpected error: {e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(1)

    return 0

Custom Exception Handler

Create a reusable exception handler:

from functools import wraps
from solana.rpc.core import RPCException
import time

def handle_rpc_errors(max_retries=3, backoff_factor=2):
    """Decorator for handling RPC errors with retry logic."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)

                except RPCException as e:
                    error_msg = str(e)

                    if '429' in error_msg and attempt < max_retries - 1:
                        wait_time = backoff_factor ** attempt
                        print(f"Rate limited. Retrying in {wait_time}s...")
                        time.sleep(wait_time)
                        continue

                    elif '401' in error_msg or '403' in error_msg:
                        print("Authentication failed")
                        raise

                    else:
                        print(f"RPC error: {e}")
                        raise

                except Exception as e:
                    print(f"Error: {e}")
                    raise

        return wrapper
    return decorator

# Usage
@handle_rpc_errors(max_retries=3)
def get_balance(address: str):
    client = Client(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    )
    pubkey = Pubkey.from_string(address)
    return client.get_balance(pubkey).value

Best Practices

1. Client Reuse

Create a client instance once and reuse it:

# Good: Create once, reuse
class SolanaService:
    def __init__(self):
        self.client = Client(
            endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
        )

    def get_balance(self, address: str) -> int:
        pubkey = Pubkey.from_string(address)
        return self.client.get_balance(pubkey).value

# Bad: Creating new clients repeatedly
def get_balance(address: str):
    client = Client(...)  # Don't do this repeatedly

2. Use Async for Concurrent Operations

Leverage async/await for better performance:

import asyncio
from solana.rpc.async_api import AsyncClient

async def process_accounts(addresses: list[str]):
    async with AsyncClient(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    ) as client:
        tasks = [
            client.get_balance(Pubkey.from_string(addr))
            for addr in addresses
        ]
        results = await asyncio.gather(*tasks)
        return results

3. Batch Operations

Use batch methods for multiple accounts:

def get_all_account_data(addresses: list[str]):
    client = Client(
        endpoint=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
        extra_headers={'X-API-Key': os.getenv('FORTIBLOX_API_KEY', '')}
    )

    pubkeys = [Pubkey.from_string(addr) for addr in addresses]

    # Single request for multiple accounts
    accounts = client.get_multiple_accounts(pubkeys).value

    return accounts

4. Type Hints

Use type hints for better code quality:

from typing import Optional, List, Dict
from solana.rpc.api import Client
from solders.pubkey import Pubkey
from solders.keypair import Keypair

def get_balance_with_types(
    client: Client,
    address: str
) -> Optional[int]:
    """Get balance with proper type hints."""
    try:
        pubkey: Pubkey = Pubkey.from_string(address)
        response = client.get_balance(pubkey)
        return response.value
    except Exception:
        return None

5. Configuration Management

Centralize configuration:

from dataclasses import dataclass
from solana.rpc.api import Client
from solana.rpc.commitment import Commitment, Confirmed

@dataclass
class FortiBloxConfig:
    rpc_url: str
    api_key: str
    commitment: Commitment = Confirmed

    @classmethod
    def from_env(cls) -> 'FortiBloxConfig':
        return cls(
            rpc_url=os.getenv('FORTIBLOX_RPC_URL', 'https://nexus.fortiblox.com/rpc'),
            api_key=os.getenv('FORTIBLOX_API_KEY', ''),
            commitment=Confirmed
        )

    def create_client(self) -> Client:
        return Client(
            endpoint=self.rpc_url,
            extra_headers={'X-API-Key': self.api_key},
            commitment=self.commitment
        )

# Usage
config = FortiBloxConfig.from_env()
client = config.create_client()

Performance Tips

Use Async for I/O: Async operations prevent blocking
Batch Requests: Combine multiple account queries
Connection Pooling: Reuse client instances
Caching: Cache static data like program IDs
Commitment Levels: Use appropriate commitment for your use case

from solana.rpc.commitment import Processed, Confirmed, Finalized

# Fast but less reliable
client.get_balance(pubkey, commitment=Processed)

# Balanced (default)
client.get_balance(pubkey, commitment=Confirmed)

# Slowest but most reliable
client.get_balance(pubkey, commitment=Finalized)

Next Steps

Explore Solana Web3.js for frontend development
Check out Go SDK for high-performance applications
Review API Reference for all available methods
Join our Discord for community support

Solana Python SDK

Solana Python SDK

Installation

Package Overview

Authentication Setup

Environment Variables

Client Configuration

Code Examples

1. Get Account Balance

2. Get Account Info

3. Send Transaction

4. Get Recent Transactions

5. Get Program Accounts

6. Get Block Information

7. Monitor Multiple Accounts

8. Async Operations

Async Patterns

Context Manager Pattern

Concurrent Data Fetching

Async Transaction Sending

Error Handling

Custom Exception Handler

Best Practices

1. Client Reuse

2. Use Async for Concurrent Operations

3. Batch Operations

4. Type Hints

5. Configuration Management

Performance Tips

Next Steps

Additional Resources

On this page