Can you access fields of a note (encoded as MsgPack) from past transactions like a dictionary?

AlgoSDK
1

I’m wondering if there’s an easier way to access information encoded in the note fields of transactions without simply copying/pasting from block explorer?

I’ve tried decoding with base64, but I get bytes that I can’t seem to figure out how to decode…
Here’s what I have so far:

def unpack_note(txid, client):
    pack = client.pending_transaction_info(txid)['txn']['txn']['note']
    undone = base64.b64decode(pack)
    print(pack, undone)
    return undone

Neither “pack” nor “undone” will print like a dictionary :thinking:. Extracting the info from “undone” is possible but I think I have to convert to a string, then back to bytes so I can decode each part of the note field properly.

Any ideas?

2

After doing more research, I’ve found a snippet of code that converts the “undone” bytes into my desired dictionary! :partying_face: :partying_face: Credit to this stack overflow post!

The finished code looks this:

import msgpack
import base64
from algosdk.v2client import algod 

def unpack_note(txid, client):
    pack = client.pending_transaction_info(txid)['txn']['txn']['note']
    undone = base64.b64decode(pack)
    dictionary = msgpack.loads(undone, raw=False)
    return dictionary
3

I don’t know if you might find it useful but, this is a Python snippet I used to write, try to retrieve and validate data as dictionaries using a 0 ALGO transaction note field.

Note: I didn’t checked if the copy&paste kept the correct indentation… :sweat_smile:

Example’s dependencies:

import time
import json
import base64
import msgpack
from schema import Schema, And, Optional
from datetime import datetime
from algosdk import mnemonic
from algosdk.error import *
from algosdk.future.transaction import PaymentTxn

Write data on-chain:

data_publisher_account = {
  'pk': mnemonic.to_public_key(passphrase),
  'sk': mnemonic.to_private_key(passphrase)
}

data = {
    'counter': 4,
    'answer': 42,
    'coefficient': 0.1,
    'timestamp': str(datetime.now())
}

bytes_note = msgpack.packb(data)

unsigned_txn = PaymentTxn(
  sender=data_publisher_account['pk'],
  sp=params,
  receiver=data_publisher_account['pk'],
  amt=0,
  note=bytes_note
)

signed_txn = unsigned_txn.sign(data_publisher_account['sk'])
txid = algod_client.send_transaction(signed_txn)

Retrive transactions note form an address, from start_block up to end_block:

def get_address_txns_note(
  indexer_client,
  address,
  start_block=None,
  end_block=None
):

  nexttoken = ""
  numtx = 1
  address_txns = []
  while numtx > 0:
    result = indexer_client.search_transactions_by_address(
      address=address,
      limit=1000,
      next_page=nexttoken,
      min_round=start_block,
      max_round=end_block
    )
    txns = result['transactions']
    address_txns += txns
    numtx = len(txns)
    if numtx > 0:
      # pointer to the next chunk of requests
      nexttoken = result['next-token']
  txns_note = [txn.get('note') for txn in address_txns]
  return txns_note


CONNECTION_ATTEMPT_DELAY_SEC = 3
MAX_CONNECTION_ATTEMPTS = 10

attempts = 1
txns_note = None
while attempts <= MAX_CONNECTION_ATTEMPTS:
  try:
    txns_note = get_address_txns_note(
      indexer_client, data_publisher_account['pk'], start_block, end_block
    )
    break
  except IndexerHTTPError:
    print(f'Indexer Client connection attempt {attempts}/{MAX_CONNECTION_ATTEMPTS}')
    print('Trying to connect to Indexer Client again...')
    time.sleep(CONNECTION_ATTEMPT_DELAY_SEC)
  finally:
    attempts += 1
if not txns_note:
  quit("Unable to connect to Indexer Client.")

Validate data using schema:

data_schema = Schema({
  'counter': int,
  'answer': And(int, lambda n: 0 <= n),
  Optional('coefficient'): And(float, lambda n: 0 <= n <= 1),
  'timestamp': str
})

data = []
for txn_note in txns_note:
  try:
    data = data_schema.validate(
      msgpack.unpackb(base64.b64decode(txn_note))
    )
    if data['answer'] == 42:
      data += [data]
  except:
    pass

if not data:
  raise Exception(
    f'Impossible to find valid data published by {data_publisher_account['pk']} '
    f'starting from block {start_block}.'
  )
1 Like