Code Examples

Practical examples for the evrmore-rpc package: configuration, assets, real-time notifications, and advanced patterns. Use the sidebar or the section headings below to jump to a topic.

Configuration examples

# From evrmore.conf
client = EvrmoreClient()

# From env vars (EVR_RPC_*)
client = EvrmoreClient()

# Manual override
client = EvrmoreClient(url="http://localhost:8819", rpcuser="user", rpcpassword="pass")

# Testnet flag
client = EvrmoreClient(testnet=True)

See Getting Started for installation and cookie authentication.

Stress test benchmarks

From python3 -m evrmore_rpc.stress_test, 100 calls to getblockcount, 10 concurrency:

Mode	Time	RPS	Avg (ms)	Median	Min	Max
Local Async	0.01 s	10442.42	0.59	0.50	0.39	1.84
Local Sync	0.06 s	1861.26	1.52	1.42	0.43	3.40
Remote Async	1.75 s	57.31	167.77	155.93	111	324
Remote Sync	1.86 s	53.83	160.39	163.26	112	310

Run: python3 -m evrmore_rpc.stress_test --sync --remote

Example files

File	Purpose
features/contextuality.py	Auto detect sync/async usage context
features/connection_pooling.py	Demonstrate connection reuse efficiency
features/complete_rpc_coverage.py	Validate RPC method coverage
features/flexible_config.py	Load configuration from various sources
features/basic_type_safety.py	Showcase type-safe access and results
features/zmq_notifications.py	Decoded ZMQ block/tx notifications
asset_operations.py	Asset transfers and issuance
basic_queries.py	Query blockchain data (blocks, balance, etc)
cookie_auth.py	Authentication using .cookie file
async_example.py	Async API interaction
zmq_auto_decode_example.py	Deep decoding of tx/block via ZMQ

Basic Usage

Initialize Client

from evrmore_rpc import EvrmoreClient

# Basic initialization
client = EvrmoreClient()

# With custom configuration
client = EvrmoreClient(
    host="127.0.0.1",
    port=8332,
    username="rpcuser",
    password="rpcpass",
    ssl=True
)

Get Blockchain Info

# Get basic blockchain info
info = client.getblockchaininfo()
print(f"Chain: {info.chain}")
print(f"Blocks: {info.blocks}")
print(f"Headers: {info.headers}")
print(f"Best block hash: {info.bestblockhash}")
print(f"Difficulty: {info.difficulty}")
print(f"Verification progress: {info.verificationprogress}")

Get Block Information

# Get block by height
block = client.getblock(1000)
print(f"Block hash: {block.hash}")
print(f"Previous block: {block.previousblockhash}")
print(f"Time: {block.time}")
print(f"Transactions: {len(block.tx)}")

# Get block by hash
block = client.getblock("0000000000000000000000000000000000000000000000000000000000000000")

Get Transaction Information

# Get transaction by ID
tx = client.getrawtransaction("txid", True)
print(f"Transaction ID: {tx.txid}")
print(f"Version: {tx.version}")
print(f"Size: {tx.size}")
print(f"Inputs: {len(tx.vin)}")
print(f"Outputs: {len(tx.vout)}")

# Get transaction details
for vin in tx.vin:
    print(f"Input: {vin.txid}:{vin.vout}")
    print(f"Sequence: {vin.sequence}")

for vout in tx.vout:
    print(f"Output value: {vout.value}")
    print(f"Script: {vout.scriptPubKey.hex}")

Asset Operations

Get Asset Information

# Get asset info
asset = client.getassetinfo("ASSET_NAME")
print(f"Asset name: {asset.name}")
print(f"Amount: {asset.amount}")
print(f"Units: {asset.units}")
print(f"Reissuable: {asset.reissuable}")
print(f"Has IPFS: {asset.has_ipfs}")
if asset.has_ipfs:
    print(f"IPFS hash: {asset.ipfs_hash}")

Transfer Assets

# Create asset transfer
txid = client.transferasset(
    "ASSET_NAME",
    "DESTINATION_ADDRESS",
    1.0,  # amount
    "COMMENT",
    "COMMENT_TO"
)
print(f"Transfer transaction ID: {txid}")

Real-Time Notifications

ZMQ Notifications

from evrmore_rpc.zmq import EvrmoreZMQClient, ZMQTopic

# Initialize ZMQ client
zmq = EvrmoreZMQClient()

# Register handlers
@zmq.on(ZMQTopic.BLOCK)
def handle_block(notification):
    print(f"New block: {notification.height}")

@zmq.on(ZMQTopic.TX)
def handle_tx(notification):
    print(f"New transaction: {notification.txid}")

# Start the client
zmq.start()

Advanced Examples

Block Explorer

class BlockExplorer:
    def __init__(self):
        self.client = EvrmoreClient()
        self.blocks_processed = 0
        self.transactions_processed = 0

    def get_block_info(self, height):
        block = self.client.getblock(height)
        print(f"\nBlock #{height}")
        print(f"Hash: {block.hash}")
        print(f"Time: {block.time}")
        print(f"Size: {block.size} bytes")
        print(f"Transactions: {len(block.tx)}")

        # Get transaction details
        for txid in block.tx:
            tx = self.client.getrawtransaction(txid, True)
            print(f"\nTransaction: {txid}")
            print(f"Size: {tx.size} bytes")
            print(f"Inputs: {len(tx.vin)}")
            print(f"Outputs: {len(tx.vout)}")

            # Check for asset transfers
            for vout in tx.vout:
                if "asset" in vout.get("scriptPubKey", {}).get("asset", {}):
                    asset = vout["scriptPubKey"]["asset"]
                    print(f"Asset transfer: {asset['name']} ({asset['amount']})")

Asset Tracker

class AssetTracker:
    def __init__(self):
        self.client = EvrmoreClient()
        self.assets = {}

    def track_asset(self, asset_name):
        # Get asset info
        asset = self.client.getassetinfo(asset_name)
        self.assets[asset_name] = {
            "name": asset.name,
            "amount": asset.amount,
            "units": asset.units,
            "reissuable": asset.reissuable,
            "transfers": []
        }

    def get_asset_transfers(self, asset_name, start_block=0):
        if asset_name not in self.assets:
            self.track_asset(asset_name)

        current_height = self.client.getblockcount()

        for height in range(start_block, current_height + 1):
            block = self.client.getblock(height)

            for txid in block.tx:
                tx = self.client.getrawtransaction(txid, True)

                for vout in tx.vout:
                    if "asset" in vout.get("scriptPubKey", {}).get("asset", {}):
                        asset = vout["scriptPubKey"]["asset"]
                        if asset["name"] == asset_name:
                            self.assets[asset_name]["transfers"].append({
                                "txid": txid,
                                "amount": asset["amount"],
                                "block": height,
                                "time": block.time
                            })

Transaction Monitor

class TransactionMonitor:
    def __init__(self):
        self.client = EvrmoreClient()
        self.watched_addresses = set()
        self.transactions = {}

    def watch_address(self, address):
        self.watched_addresses.add(address)

    def get_address_transactions(self, address):
        if address not in self.watched_addresses:
            self.watch_address(address)

        # Get all transactions for address
        txs = self.client.getaddresstxids({"addresses": [address]})

        for txid in txs:
            if txid not in self.transactions:
                tx = self.client.getrawtransaction(txid, True)
                self.transactions[txid] = {
                    "txid": txid,
                    "time": tx.time,
                    "inputs": [],
                    "outputs": []
                }

                # Process inputs
                for vin in tx.vin:
                    if "coinbase" not in vin:
                        prev_tx = self.client.getrawtransaction(vin.txid, True)
                        prev_vout = prev_tx.vout[vin.vout]
                        self.transactions[txid]["inputs"].append({
                            "address": prev_vout.scriptPubKey.addresses[0],
                            "amount": prev_vout.value
                        })

                # Process outputs
                for vout in tx.vout:
                    if "addresses" in vout.scriptPubKey:
                        self.transactions[txid]["outputs"].append({
                            "address": vout.scriptPubKey.addresses[0],
                            "amount": vout.value
                        })

Error Handling

from evrmore_rpc import EvrmoreRPCError

try:
    # Make RPC call
    result = client.getblockchaininfo()
except EvrmoreRPCError as e:
    print(f"RPC error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Performance Optimization

from evrmore_rpc import EvrmoreClient

# Reuse a single client instance to take advantage of internal connection reuse
client = EvrmoreClient()

def process_blocks(start_height, end_height):
    for height in range(start_height, end_height + 1):
        block = client.getblock(height)
        # Process block data

See Also

• Getting Started for basic usage
• API Reference for detailed API docs
• ZMQ Guide for real-time notifications
• Advanced Usage for production patterns