Symbol(Python SDK)覚え書き

どうも、Python初心者です。
これは、Symbol SDK Python版の覚え書きです。情報は随時追記していく予定です。

SDKバージョン

symbol-sdk-python 3.1.0

プライベートキーからアカウントを復元

privaetKey = "***************"
aliceKeyPair = KeyPair(PrivateKey(privaetKey))
aliceAddress = facade.network.public_key_to_address(aliceKeyPair.public_key)
print(aliceAddress)

トランスファートランザクションの送信

from datetime import datetime, timezone
from symbolchain.facade.SymbolFacade import SymbolFacade
from symbolchain.symbol.Network import Network, NetworkTimestamp
from symbolchain.CryptoTypes import Hash256
from symbolchain.symbol.KeyPair import KeyPair
from symbolchain.CryptoTypes import PrivateKey
import requests
import json
from symbolchain.symbol.IdGenerator import generate_mosaic_alias_id

SYMBOL_API_ENDPOINT = 'https://6.dusanjp.com:3001'

def announce(url, payload):
    headers = {'Content-Type': 'application/json'}
    response = requests.put(url, json=json.loads(payload), headers=headers)
    print(response.json())

def get_network_time():
    headers = {'Content-Type': 'application/json'}
    response = requests.get(f'{SYMBOL_API_ENDPOINT}/node/time', headers=headers)
    response_json = response.json()
    # extract the network time from the json
    timestamp = NetworkTimestamp(int(response_json['communicationTimestamps']['receiveTimestamp']))
    print(f'network time: {timestamp} ms')
    return timestamp

def main():

    facade = SymbolFacade("testnet")

    network_time: NetworkTimestamp = get_network_time()
    private_key = "**********"
    privateKey = KeyPair(PrivateKey(private_key))
    transaction = facade.transaction_factory.create({
        'type': 'transfer_transaction_v1',
        'signer_public_key': '"**********"',
        'fee': 1000000,
        'deadline': network_time.add_hours(2).timestamp,
        'recipient_address': 'TDQXD2LYTP625VUGLPWST7PIGHOZPYFFA2XYLBI',
        'mosaics': [
            {'mosaic_id': generate_mosaic_alias_id("symbol.xym"), 'amount': 1000000}
        ]
    })
    signature = facade.sign_transaction(privateKey, transaction)
    print(f'hash: {facade.hash_transaction(transaction)}')

    json_payload = facade.transaction_factory.attach_signature(transaction, signature)
    announce(f'{SYMBOL_API_ENDPOINT}' + "/transactions", json_payload)


main()

アグリゲートボンデッドトランザクションの送信

from datetime import datetime, timezone
from symbolchain.facade.SymbolFacade import SymbolFacade
from symbolchain.symbol.Network import Network, NetworkTimestamp
from symbolchain.sc import PublicKey, Amount
from symbolchain.CryptoTypes import Hash256
from symbolchain.symbol.KeyPair import KeyPair
from symbolchain.CryptoTypes import PrivateKey
import requests
import json
from symbolchain.symbol.IdGenerator import generate_mosaic_alias_id
import time

SYMBOL_API_ENDPOINT = 'https://6.dusanjp.com:3001'

def announce(url, payload):
    headers = {'Content-Type': 'application/json'}
    response = requests.put(url, json=json.loads(payload), headers=headers)
    print(response.json())

def get_network_time():
    headers = {'Content-Type': 'application/json'}
    response = requests.get(f'{SYMBOL_API_ENDPOINT}/node/time', headers=headers)
    response_json = response.json()
    # extract the network time from the json
    timestamp = NetworkTimestamp(int(response_json['communicationTimestamps']['receiveTimestamp']))
    print(f'network time: {timestamp} ms')
    return timestamp

def main():

    facade = SymbolFacade("testnet")

    # bob Address
    public_key_bytes = bytes.fromhex("***********")
    bobPublicKey = PublicKey(public_key_bytes)
    bobAddress = facade.network.public_key_to_address(bobPublicKey)
   
    # alice account
    privaetKey = "**********"
    aliceKeyPair = KeyPair(PrivateKey(privaetKey))
    aliceAddress = facade.network.public_key_to_address(aliceKeyPair.public_key)

    # 各アドレス
    print(f'alicePublicKey: {aliceKeyPair.public_key}')
    print(f'aliceAddress: {aliceAddress}')
    print(f'bobPublicKey: {bobPublicKey}')
    print(f'bobAddress: {bobAddress}')

    aliceTransaction = facade.transaction_factory.create_embedded({
        'type': 'transfer_transaction_v1',
        'signer_public_key': aliceKeyPair.public_key,
        'recipient_address': bobAddress,
        'mosaics': [
            {'mosaic_id': generate_mosaic_alias_id("symbol.xym"), 'amount': 1000000}
        ],
        'message': bytes(1) + "from alick->bob".encode('utf8')
    })

    bobTransaction = facade.transaction_factory.create_embedded({
        'type': 'transfer_transaction_v1',
        'signer_public_key': str(bobPublicKey),
        'recipient_address': aliceAddress,
        'message': bytes(1) + "from bob->alice".encode('utf8')
    })

    embeddedTransactions = [aliceTransaction, bobTransaction]
    merkle_hash = facade.hash_embedded_transactions(embeddedTransactions)
    print(f'merkle_hash: {merkle_hash}')

    # deadline作成
    aggregate_deadline: NetworkTimestamp = get_network_time()

    # aggregate transaction
    aggregate_transaction = facade.transaction_factory.create({
      'type': 'aggregate_bonded_transaction_v2',
      'signer_public_key': aliceKeyPair.public_key,
      'deadline': aggregate_deadline.add_hours(2).timestamp,
      'transactions_hash': merkle_hash,
      'transactions': embeddedTransactions
    })
    # 手数料が低すぎるとトランザクションが流れてしまうので、100→1000とかでもいいかもしれない
    aggregate_transaction.fee = Amount(100 * aggregate_transaction.size)

    signature = facade.sign_transaction(aliceKeyPair, aggregate_transaction)
    aggregate_transaction_payload = facade.transaction_factory.attach_signature(aggregate_transaction, signature)
    aggregate_hash = facade.hash_transaction(aggregate_transaction)

    # ハッシュロックトランザクション
    network_time = get_network_time()

    hash_lock_transaction = facade.transaction_factory.create({
        'type': 'hash_lock_transaction_v1',
        'signer_public_key': aliceKeyPair.public_key,
        'deadline': network_time.add_hours(2).timestamp,
        'duration': 480,
        'mosaic': { 'mosaic_id': generate_mosaic_alias_id("symbol.xym"), 'amount': 10_000000 },
        'hash': aggregate_hash
    })
    hash_lock_transaction.fee = Amount(100 * hash_lock_transaction.size)

    signature = facade.sign_transaction(aliceKeyPair, hash_lock_transaction)
    hash_lock_transaction_payload = facade.transaction_factory.attach_signature(hash_lock_transaction, signature)

    announce(f'{SYMBOL_API_ENDPOINT}' + "/transactions", hash_lock_transaction_payload)

    # 45秒待つ
    time.sleep(45)

    # # アグリゲートをアナウンス
    announce(f'{SYMBOL_API_ENDPOINT}' + "/transactions/partial", aggregate_transaction_payload)

    print(f'hash_lock hash: {facade.hash_transaction(hash_lock_transaction)}')
    # print(f'hash_lock payload: {hash_lock_transaction_payload}')
    print(f'aggregate hash: {facade.hash_transaction(aggregate_transaction)}')
    # print(f'aggregate payload: {aggregate_transaction_payload}')

main()

参考

symbol docs (github)