Files
xrpl-dev-portal/_code-samples/claim-payment-channel/py/claim_pay_channel.py
2025-07-17 15:18:59 +01:00

86 lines
2.9 KiB
Python

"""
Create, claim and verify a Payment Channel.
Reference: https://xrpl.org/docs/references/protocol/ledger-data/ledger-entry-types/paychannel
"""
from xrpl.clients import JsonRpcClient
from xrpl.models import (
AccountObjects,
PaymentChannelCreate,
PaymentChannelClaim,
)
from xrpl.wallet import generate_faucet_wallet
from xrpl.transaction import submit, submit_and_wait
from xrpl.account import get_balance
"""The snippet walks us through creating and claiming a Payment Channel."""
client = JsonRpcClient("https://s.altnet.rippletest.net:51234")
# Creating wallets as prerequisite
print("Setting up wallets...")
wallet1 = generate_faucet_wallet(client)
wallet2 = generate_faucet_wallet(client)
print("Balances of wallets before Payment Channel is claimed:")
print(f"Balance of {wallet1.address} is {get_balance(wallet1.address, client)} XRP")
print(f"Balance of {wallet2.address} is {get_balance(wallet2.address, client)} XRP")
# Create a Payment Channel and submit and wait for tx to be validated
payment_channel_create = PaymentChannelCreate(
account=wallet1.address,
amount="100", # 10 XRP
destination=wallet2.address,
settle_delay=86400, # 1 day in seconds
public_key=wallet1.public_key,
)
print("\nSubmitting a PaymentChannelCreate transaction...")
payment_channel_response = submit_and_wait(
payment_channel_create,
client,
wallet1
)
print("PaymentChannelCreate transaction response:")
print(payment_channel_response.result)
# Check that the object was actually created
account_objects_request = AccountObjects(
account=wallet1.address,
)
account_objects_response = client.request(account_objects_request)
account_objects = account_objects_response.result["account_objects"]
# Find the PayChannel object to get the correct channel ID
channel_id = None
if 'meta' in payment_channel_response.result and 'AffectedNodes' in payment_channel_response.result['meta']:
for node in payment_channel_response.result["meta"]["AffectedNodes"]:
if 'CreatedNode' in node and node["CreatedNode"]["LedgerEntryType"] == 'PayChannel':
channel_id = node['CreatedNode']['LedgerIndex']
break
if not channel_id:
raise Exception("Payment Channel ID not found in the response.")
print(f"Payment Channel ID: {channel_id}")
# Destination claims the Payment Channel and we see the balances to verify.
payment_channel_claim = PaymentChannelClaim(
account=wallet2.address,
channel=channel_id,
amount="100",
)
print("\nSubmitting a PaymentChannelClaim transaction...")
channel_claim_response = submit_and_wait(
payment_channel_claim,
client,
wallet2,
)
print("PaymentChannelClaim transaction response:")
print(channel_claim_response.result)
print("Balances of wallets after Payment Channel is claimed:")
print(f"Balance of {wallet1.address} is {get_balance(wallet1.address, client)} XRP")
print(f"Balance of {wallet2.address} is {get_balance(wallet2.address, client)} XRP")