【Jaguar's Blog 6】SDK Python - Symbol Transactions
この記事は2023年12月27日にNEM/Symbolのコア開発者Jaguar氏によって投稿された記事「SDK Python - Symbol Transactions」をChatGPTを用いて翻訳して記事です。
Symbol Python SDKはSymbolトランザクションの送信をサポートしています。このガイドでは、SDKとaiohttpだけを使用して非同期にトランザクションを送信する方法について説明します。トランザクションディスクリプタが与えられた場合に、簡単にネットワークにトランザクションを送信するための関数を作成します。
すでにSDK Python - NEMトランザクションの記事を読んでいる場合、例は非常に馴染んで見えるはずです。これはSDKの明確なデザイン目標であり、意図的なものです。
Setup
このガイドを使用して新しいプロジェクトを作成することをお勧めします。セットアップが完了したら、依存関係を追加しましょう：
pip install aiohttp
pip install symbol-sdk-python新しいpythonファイルを作成し、先頭に以下を追加します：
import asyncio
import time
from aiohttp import ClientSession
from symbolchain.CryptoTypes import PrivateKey
from symbolchain.facade.SymbolFacade import SymbolFacade
from symbolchain.sc import Amount
from symbolchain.symbol.Network import NetworkTimestamp
SYMBOL_PRIVATE_KEY = '<your private key>'
SYMBOL_PRIVATE_KEY_2 = '<your other private key>'
SYMBOL_API_ENDPOINT = 'http://your-testnet-node:7890'トップレベルの定数を適切な値に設定することを確認してください。
SYMBOL_PRIVATE_KEYはXYMを送信するアカウントの秘密キーである必要があります。
SYMBOL_PRIVATE_KEY_2はXYMを受け取るアカウントの秘密キーである必要があります。
SYMBOL_API_ENDPOINTはSymbolネットワークのノードのエンドポイントで、プロトコルとポートを含める必要があります。
ℹ️ このガイドでは、実際には2つ目の秘密キーは必要ありません。代わりに、アドレスに置き換えることもできます。
SYMBOL_PRIVATE_KEYに資金を供給する必要がある場合は、Symbol faucet を使用できます。
Network Time
Symbolブロックチェーンにトランザクションを送信する際、それは現在のネットワーク時間を過ぎた締め切りを持っている必要があります。さらに、それは含まれるブロックから最大で6時間以内の締め切りを持っている必要があります。唯一の例外は、アグリゲートボンデッドトランザクションで、将来の最大で2日先まで締め切りを持つことができます。
締め切りを正しく設定するためには、現在のネットワーク時間を取得する必要があります。get_network_timeと呼ばれる関数を作成して、それを行うことにしましょう。
async def get_network_time():
async with ClientSession(raise_for_status=True) as session:
# initiate a HTTP GET request to a SYMBOL REST endpoint
async with session.get(f'{SYMBOL_API_ENDPOINT}/node/time') as response:
# wait for the (JSON) response
response_json = await response.json()
# extract the network time from the json
timestamp = NetworkTimestamp(int(response_json['communicationTimestamps']['sendTimestamp']))
return timestampまず、ClientSessionを作成し、raise_for_statusを設定して、サーバーエラーを例外に変換できるようにします。次に、ルートnode/timeにGETリクエストを送信します。このルートはSymbolノードの時刻同期ルーチンで使用されますが、ここでは現在の時刻を判断するために使用します。これはsendTimestampとreceiveTimestampを含むcommunicationTimestampsオブジェクトを持つJSONオブジェクトを返します。これらはいずれもSymbolネットワークのタイムスタンプ（ミリ秒単位）です。そのうちの一つ（sendTimestamp）を選び、Symbolのタイムスタンプを表すNetworkTimestampオブジェクトでラップします。これは追加の時間を追加するためのヘルパー関数を提供します。
Push Transaction to Network
また、ネットワークにトランザクションを送信できるようにする必要もあります。Symbolでは、トランザクションはシリアライズされたトランザクションペイロードとその署名を含む形でネットワークに送信されます。このデータで構成されたJSON文字列があると仮定します。これをネットワークに送信するためのpush_transactionと呼ばれる関数を作成しましょう。
async def push_transaction(json_payload):
async with ClientSession(raise_for_status=True) as session:
# initiate a HTTP PUT request to a SYMBOL REST endpoint
async with session.put(f'{SYMBOL_API_ENDPOINT}/transactions', data=json_payload, headers={
'Content-Type': 'application/json'
}) as response:
# wait for the (JSON) response
return await response.json()まず、ClientSessionを作成し、raise_for_statusを設定して、サーバーエラーを例外に変換できるようにします。次に、transactionsルートにPUTリクエストを送信します。PUTデータはJSON文字列であり、コンテントタイプをapplication/jsonに設定することを確認します。最後に、レスポンスを待ちます。Symbolでは、トランザクション処理が非同期で行われるため、トランザクションが送信されたことの確認しか得られません。トランザクションが受け入れられたかどうか（未確認トランザクションキャッシュに移動されたか）および最終的に確認されたかどうか（ブロックに含まれたか）を確認するために、ノードを監視する必要があります。
Wait for Status
トランザクションのステータスは、ネットワークをポーリングして監視できます。最良の結果を得るためには、トランザクションを送信したノードからトランザクションのステータスをクエリすることがお勧めです。
async def wait_for_transaction_status(transaction_hash, desired_status):
async with ClientSession(raise_for_status=False) as session:
for _ in range(6):
# query the status of the transaction
async with session.get(f'{SYMBOL_API_ENDPOINT}/transactionStatus/{transaction_hash}') as response:
# wait for the (JSON) response
response_json = await response.json()
# check if the transaction has transitioned
if 200 == response.status:
status = response_json['group']
print(f'transaction {transaction_hash} has status "{status}"')
if desired_status == status:
print(f'transaction {transaction_hash} has transitioned to {desired_status}')
return
if 'failed' == status:
print(f'transaction {transaction_hash} failed validation: {response_json["code"]}')
break
else:
print(f'transaction {transaction_hash} has unknown status')
# if not, wait 20s before trying again
time.sleep(20)
# fail if the transaction didn't transition to the desired status after 2m
raise RuntimeError(f'transaction {transaction_hash} did not transition to {desired_status} in alloted time period')まず、ClientSessionを作成し、raise_for_statusを設定しないでおきます。これにより、サーバーエラーを手動で処理できます。次に、transactionStatusルートにトランザクションのハッシュを含めたGETリクエストを送信します。リクエストが成功した場合、レスポンスを検査してトランザクションのステータスを判断します。ステータスが目的のステータスまたは失敗した場合、ポーリングを停止します。それ以外の場合は、5秒間スリープしてから再試行します。
Currency Mosaic ID
Symbolでは、ネットワークには異なるベースモザイクが存在します。一般的には、汎用のコードを記述する際には、ネットワークからその通貨モザイクIDをクエリすることが最善の慣習です。これにより、コードが異なるネットワーク（たとえば、メインネットとテストネット）とシームレスに動作するようになります。
Symbol RESTノードはconfig-network.propertiesファイル全体を公開しています。その構成をクエリして通貨モザイクIDを抽出しましょう：
async def get_currency_mosaic_id():
async with ClientSession(raise_for_status=True) as session:
# initiate a HTTP GET request to a SYMBOL REST endpoint
async with session.get(f'{SYMBOL_API_ENDPOINT}/network/properties') as response:
# wait for the (JSON) response
response_json = await response.json()
# extract the currency mosaic id from the json
formatted_currency_mosaic_id = response_json['chain']['currencyMosaicId']
return int(formatted_currency_mosaic_id.replace('\'', ''), 16)まず、ClientSessionを作成し、raise_for_statusを設定して、サーバーエラーを例外に変換できるようにします。次に、network/propertiesルートにGETリクエストを送信します。これにより、完全なネットワーク構成を含むJSONオブジェクトが返されます。そこからcurrencyMosaicId設定を抽出します。最後に、千の区切り文字（''）を削除し、16進文字列から整数に変換して返します。
Transaction Descriptor
Symbol SDKでは、トランザクションのプロパティのディレクショナリはトランザクションディスクリプタと呼ばれます。SDKはこのディスクリプタを受け入れ、トランザクションオブジェクトを返します。
3つの引数を受け入れる関数を書いてみましょう：
facade: (Symbol)ネットワークとの対話のためのSymbol SDKファサード
signer_key_pair: トランザクションの送信者のキーペアで、トランザクションに署名するために使用されます
transaction_descriptor: 望ましいトランザクションプロパティで構成されたトランザクションディスクリプタ
async def prepare_and_send_transaction(facade, signer_key_pair, transaction_descriptor):まず、前に書いたget_network_time関数を使用してネットワーク時間をクエリする必要があります。
async with ClientSession(raise_for_status=True) as session:
# get the current network time from the network
network_time = await get_network_time(session)ℹ️ 注意してください、get_network_timeを作成するのではなく、ClientSessionを受け入れるように変更しました。可能であれば、ClientSessionオブジェクトを再利用することがお勧めされます。各セッションは接続プールをカプセル化しており、この接続プールを活用することでアプリケーションのパフォーマンスが向上する可能性があります。
次に、transaction_descriptor引数からトランザクションオブジェクトを作成する必要があります。これは、ファサードのトランザクションファクトリのcreateメソッドを使用して行うことができます：
さらに、すべてのトランザクションに共通するいくつかのプロパティを設定します：
signer_public_key: トランザクションの署名者の公開鍵；signer_key_pairから派生します
deadline: タイムスタンプの1時間後；ここではNetworkTimestampが提供するadd_hoursを使用してこれを計算しています
fee: Symbolトランザクション手数料はバイトごとのコスト手数料です；ここでは手数料の乗数が100の場合を想定して手数料を設定しています
3番目に、トランザクションに署名し、適切なトランザクションペイロードを構築する必要があります：
# sign the transaction and attach its signature
signature = facade.sign_transaction(signer_key_pair, transaction)
json_payload = facade.transaction_factory.attach_signature(transaction, signaturesign_transactionはsigner_key_pairでトランザクションに署名し、その結果の署名を返します。attach_signatureは、直接ネットワークに送信できる、シリアライズされたトランザクションデータと署名の両方で構成されたトランザクションペイロードを構築します。
4番目に、オプションとして、hash_transactionを呼び出してトランザクションのハッシュを計算します：
# hash the transaction (this is dependent on the signature)
transaction_hash = facade.hash_transaction(transaction)
print(f'transaction hash {transaction_hash}')5番目に、先に作成したpush_transaction関数を使用してトランザクションをネットワークに送信します：
# send the transaction to the network
push_result = await push_transaction(session, json_payload)
print('transaction was sent')ここでは、トランザクションが送信されたことしかわかりませんが、それが受け入れられたか、および/または確認されたかどうかはわかりません。
最後に、トランザクションのステータスをクエリし、確認されるのを待ちましょう：
# wait for the transaction to be confirmed
await wait_for_transaction_status(transaction_hash, 'confirmed')単に先に書いたヘルパー関数を呼び出し、トランザクションが確認されるのを待つだけです。
Usage
今書いた関数を使って、ネットワークにトランザクションを送信する方法を見てみましょう！
まず、テストネットワークのためのSymbolFacadeを作成し、2つのアカウントをロードしてください：
facade = SymbolFacade('testnet')
signer_key_pair = facade.KeyPair(PrivateKey(SYMBOL_PRIVATE_KEY))
signer_address = facade.network.public_key_to_address(signer_key_pair.public_key)
print(f'signer address {signer_address}')
recipient_key_pair = facade.KeyPair(PrivateKey(SYMBOL_PRIVATE_KEY_2))
recipient_address = facade.network.public_key_to_address(recipient_key_pair.public_key)
print(f'recipient address {recipient_address}')次に、ネットワークの通貨モザイクIDをクエリしてください：
currency_mosaic_id = await get_currency_mosaic_id()
print(f'currency mosaic id {currency_mosaic_id:16X}')最後に、適切なディスクリプタを使用して関数を呼び出します：
await prepare_and_send_transaction(facade, signer_key_pair, {
'type': 'transfer_transaction_v1',
'recipient_address': recipient_address,
'mosaics': [
{'mosaic_id': currency_mosaic_id, 'amount': 2_000000}
]
})この場合、1つのアカウント（SYMBOL_PRIVATE_KEY）から別のアカウント（SYMBOL_PRIVATE_KEY_2）に2 XEMを送信するトランザクションを準備して送信しています。これは転送トランザクションなので、typeをtransfer_transaction_v1と指定しています。recipient_addressはSYMBOL_PRIVATE_KEY_2から派生しています。mosaics配列を使用すると複数のモザイクを一度に送信できますが、ここでは通貨モザイクID（XYM）の単位を2つだけ送信しています。
すべてのステップに従った場合、prepare_and_send_transactionを呼び出すことで、簡単にSymbolトランザクションをネットワークに送信できるようになりましたね！このコードは非常にDRY（Don't Repeat Yourself）です。異なるトランザクションを送信するには、prepare_and_send_transactionに渡すトランザクションディスクリプタを変更するだけです。さらに、共通のフィールドはすべて1か所で正しく設定されており、コード（またはディスクリプタ）を混乱させる必要はありません。
Bonus - Symbol Connector
自分でSymbol RESTエンドポイントをどのように見つけるか疑問に思っている場合は、symbol-lightapiという別のライブラリが助けになります。このライブラリはまだ包括的ではありませんが、常にプルリクエストを歓迎していますし、現在も進行中の作業です。それにもかかわらず、これは上記で書いた多くのヘルパー関数を置き換えるために使用できます。どのような機能が提供されているかを簡単に見てみましょう。
まず最初に、pipを使用してsymbol-lightapiパッケージをインストールする必要があります：
pip install symbol-lightapiこのパッケージを使用して、Symbolノードを中心にコネクタを作成することができます。このコネクタを使用すると、Symbolネットワークのプロパティをクエリできます。実際、get_network_timeとget_currency_mosaic_idは、このコネクタ上での関数呼び出しによって完全に置き換えることができます：
from symbollightapi.connector.SymbolConnector import SymbolConnector
async def query_network_with_connector():
connector = SymbolConnector(SYMBOL_API_ENDPOINT)
currency_mosaic_id = await connector.currency_mosaic_id()
print(f'currency_mosaic_id: {currency_mosaic_id}')
network_timestamp = await connector.network_time()
print(f' network_timestamp: {network_timestamp}')さらに、このコネクタを使用してネットワークにトランザクションを送信することもできます：
try:
await connector.announce_transaction(transaction)
except NodeException:
pass # ignore 202ℹ️ 現時点では、HTTP 202のレスポンスがエラーとして扱われるバグがあります。これが修正されると、try/exceptブロックは削除できるし、削除するべきです。
この記事が気に入ったらサポートをしてみませんか？