tessl/pypi-xrpl-py
A complete Python library for interacting with the XRP Ledger blockchain, providing transaction creation, account management, and comprehensive XRPL protocol support
ledger.mddocs/
Ledger Operations
Query ledger state information including current fees, ledger sequences, and network status. These functions provide essential ledger-level information for building XRPL applications.
Capabilities
Fee Information
Query current network fees and reserve requirements.
from xrpl import ledger
def get_fee(client, max_fee_xrp: str = None) -> str:
"""
Get current base fee and reserve amounts from the network.
Args:
client: XRPL client for network communication
max_fee_xrp: Optional maximum fee limit in XRP
Returns:
Current base fee in drops as string
Raises:
XRPLException: If fee query fails
"""Ledger Sequence Information
Query ledger sequence numbers for timing and validation purposes.
def get_latest_open_ledger_sequence(client) -> int:
"""
Get the latest open (not yet validated) ledger sequence number.
Args:
client: XRPL client for network communication
Returns:
Latest open ledger sequence number
Raises:
XRPLException: If ledger query fails
"""
def get_latest_validated_ledger_sequence(client) -> int:
"""
Get the latest validated ledger sequence number.
Args:
client: XRPL client for network communication
Returns:
Latest validated ledger sequence number
Raises:
XRPLException: If ledger query fails
"""Usage Examples
Checking Current Network Fees
from xrpl.clients import JsonRpcClient
from xrpl import ledger
# Connect to XRPL network
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
# Get current base fee
try:
base_fee = ledger.get_fee(client)
print(f"Current base fee: {base_fee} drops")
# Convert to XRP for display
from xrpl.utils import drops_to_xrp
fee_xrp = drops_to_xrp(base_fee)
print(f"Base fee in XRP: {fee_xrp}")
except Exception as e:
print(f"Failed to get fee: {e}")
# Get fee with maximum limit
try:
limited_fee = ledger.get_fee(client, max_fee_xrp="0.01")
print(f"Fee with limit: {limited_fee} drops")
except Exception as e:
print(f"Failed to get limited fee: {e}")Monitoring Ledger Progress
from xrpl.clients import JsonRpcClient
from xrpl import ledger
import time
def monitor_ledger_progress(client, duration: int = 60):
"""Monitor ledger sequence progression."""
print("Monitoring ledger progress...")
start_time = time.time()
while time.time() - start_time < duration:
try:
# Get current ledger info
open_seq = ledger.get_latest_open_ledger_sequence(client)
validated_seq = ledger.get_latest_validated_ledger_sequence(client)
print(f"Open: {open_seq}, Validated: {validated_seq}, Gap: {open_seq - validated_seq}")
time.sleep(5) # Check every 5 seconds
except Exception as e:
print(f"Error monitoring ledger: {e}")
time.sleep(5)
# Usage
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
monitor_ledger_progress(client, duration=30)Transaction Timing Utilities
from xrpl.clients import JsonRpcClient
from xrpl import ledger
def get_transaction_timing_info(client) -> dict:
"""Get timing information for transaction submission."""
try:
# Get current ledger sequences
open_seq = ledger.get_latest_open_ledger_sequence(client)
validated_seq = ledger.get_latest_validated_ledger_sequence(client)
# Calculate recommended LastLedgerSequence
# Typically set 3-10 ledgers ahead for safety
recommended_last_ledger = open_seq + 5
return {
"current_open": open_seq,
"current_validated": validated_seq,
"ledger_gap": open_seq - validated_seq,
"recommended_last_ledger": recommended_last_ledger,
"time_window_seconds": 5 * 4 # ~4 seconds per ledger
}
except Exception as e:
return {"error": str(e)}
# Usage
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
timing_info = get_transaction_timing_info(client)
if "error" not in timing_info:
print("Transaction Timing Info:")
print(f" Current open ledger: {timing_info['current_open']}")
print(f" Current validated ledger: {timing_info['current_validated']}")
print(f" Recommended LastLedgerSequence: {timing_info['recommended_last_ledger']}")
print(f" Estimated time window: {timing_info['time_window_seconds']} seconds")
else:
print(f"Error getting timing info: {timing_info['error']}")Fee Estimation for Transactions
from xrpl.clients import JsonRpcClient
from xrpl import ledger
from xrpl.utils import xrp_to_drops
def calculate_transaction_cost(client, transaction_size_factor: float = 1.0) -> dict:
"""Calculate estimated transaction cost."""
try:
# Get base fee
base_fee_drops = int(ledger.get_fee(client))
# Account for transaction complexity
# Simple transactions use base fee, complex ones may need more
estimated_fee_drops = int(base_fee_drops * transaction_size_factor)
# Add some buffer for network congestion
recommended_fee_drops = int(estimated_fee_drops * 1.2)
return {
"base_fee_drops": base_fee_drops,
"estimated_fee_drops": estimated_fee_drops,
"recommended_fee_drops": recommended_fee_drops,
"recommended_fee_xrp": float(recommended_fee_drops / 1_000_000)
}
except Exception as e:
return {"error": str(e)}
# Usage examples
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
# Simple payment
simple_cost = calculate_transaction_cost(client, 1.0)
print("Simple transaction cost:", simple_cost)
# Complex transaction (e.g., with paths, multiple currencies)
complex_cost = calculate_transaction_cost(client, 2.0)
print("Complex transaction cost:", complex_cost)
# Multi-sign transaction
multisign_cost = calculate_transaction_cost(client, 1.5)
print("Multi-sign transaction cost:", multisign_cost)Asynchronous Ledger Operations
import asyncio
from xrpl.asyncio.clients import AsyncJsonRpcClient
from xrpl.asyncio import ledger
async def get_ledger_info_async(client):
"""Get comprehensive ledger information asynchronously."""
# Run multiple queries concurrently
fee_task = ledger.get_fee(client)
open_seq_task = ledger.get_latest_open_ledger_sequence(client)
validated_seq_task = ledger.get_latest_validated_ledger_sequence(client)
# Wait for all to complete
fee, open_seq, validated_seq = await asyncio.gather(
fee_task, open_seq_task, validated_seq_task
)
return {
"base_fee_drops": fee,
"open_ledger_sequence": open_seq,
"validated_ledger_sequence": validated_seq,
"ledger_gap": open_seq - validated_seq
}
async def main():
client = AsyncJsonRpcClient("https://s.altnet.rippletest.net:51234")
try:
ledger_info = await get_ledger_info_async(client)
print("Ledger Information:")
for key, value in ledger_info.items():
print(f" {key}: {value}")
finally:
await client.close()
# Run async example
asyncio.run(main())Integration with Transaction Management
The ledger functions are commonly used with transaction operations:
from xrpl.clients import JsonRpcClient
from xrpl import ledger, transaction
from xrpl.models.transactions import Payment
from xrpl.wallet import Wallet
def create_payment_with_timing(client, wallet, destination: str, amount: str):
"""Create payment with proper timing information."""
# Get current ledger state
open_seq = ledger.get_latest_open_ledger_sequence(client)
base_fee = ledger.get_fee(client)
# Create payment with timing info
payment = Payment(
account=wallet.address,
destination=destination,
amount=amount,
fee=str(int(int(base_fee) * 1.2)), # 20% buffer on fee
last_ledger_sequence=open_seq + 5 # 5 ledger window
)
print(f"Created payment with fee: {payment.fee} drops")
print(f"LastLedgerSequence: {payment.last_ledger_sequence}")
return payment
# Usage
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
wallet = Wallet.create() # In practice, use funded wallet
payment = create_payment_with_timing(
client,
wallet,
"rReceiver...",
"1000000"
)Install with Tessl CLI
npx tessl i tessl/pypi-xrpl-py