txn_parser.md

Usage

txns = getBlockTxns(block=100)
txn = txns[0]

sender = TxnFields.sender(txn)
gid = TxnFields.group(txn)
appId = TxnFields.appId(txn)
innerTxns = TxnFields.innerTxns(txn)

# txn_parser.py
from datetime import datetime

# Handles both ALGO and ASA transfers.
def parseTransferTxn (txn):
    if 'tx-type' in txn: # indexer
        if txn['tx-type'] == 'pay':
            assetId = 0
            amount = txn['payment-transaction']['amount']
        else:
            xfer = txn['asset-transfer-transaction']
            assetId = xfer['asset-id']
            amount = xfer['amount']
    else: # algod
        if txn['type'] == 'pay':
            assetId = 0
            amount = txn.get('amt', 0)
        else:
            assetId = txn['xaid']
            amount = txn.get('aamt', 0)

    return assetId, amount

def getTxnTimestamp (txn):
    return datetime.utcfromtimestamp(txn['round-time'])

def getBlockTxns (block: int, indexerBlock=False):
    if indexerBlock:
        res = indexer.block_info(block)
    else:
        res = algod.block_info(block)
    return getTransactionsList(res)

def getTransactionsList (response):
    if 'transactions' in response: # indexer
        return response['transactions']
    elif 'block' in response: # algod
        block = response['block']
        return [normalizeAlgodTxn(d, block['rnd'], block['ts']) for d in block['txns']]
    else:
        return []

def normalizeAlgodTxn(data, round, timestamp):
    """Converts algod txn format to have a similar structure to indexer txns."""
    txn = data['txn']
    txn['confirmed-round'] = round
    txn['round-time'] = timestamp

    if 'dt' in data:
        txn['dt'] = data['dt']

    return txn

class TxnFields:
    """Abstracts away the differences between indexer, algod,
    and mempool txn formats.
    """
    @classmethod
    def group(cls, txn):
        return txn.get('grp', txn.get('group'))

    @classmethod
    def type(cls, txn):
        return txn.get('type', txn.get('tx-type'))

    @classmethod
    def sender(cls, txn):
        return txn.get('snd', txn.get('sender'))

    @classmethod
    def receiver(cls, txn):
        if 'rcv' in txn:
            return txn['rcv']
        elif 'arcv' in txn:
            return txn['arcv']
        elif 'payment-transaction' in txn:
            return txn['payment-transaction']['receiver']
        elif 'asset-transfer-transaction' in txn:
            return txn['asset-transfer-transaction']['receiver']

    @classmethod
    def accounts(cls, txn):
        return cls.appField(txn, 'apat', 'accounts') or []

    @classmethod
    def transferAmount(cls, txn):
        if 'amt' in txn:
            return txn['amt']
        elif 'aamt' in txn:
            return txn['aamt']
        elif 'payment-transaction' in txn:
            return txn['payment-transaction']['amount']
        elif 'asset-transfer-transaction' in txn:
            return txn['asset-transfer-transaction']['amount']

    @classmethod
    def transferAssetId(cls, txn):
        txnType = cls.type(txn)
        if txnType == 'pay': # ALGO
            return 0
        elif 'xaid' in txn:
            return txn['xaid']
        elif 'asset-transfer-transaction' in txn:
            return txn['asset-transfer-transaction']['asset-id']

    @classmethod
    def appId(cls, txn):
        return cls.appField(txn, 'apid', 'application-id')

    @classmethod
    def appArgs(cls, txn):
        return cls.appField(txn, 'apaa', 'application-args') or []

    @classmethod
    def foreignAssets(cls, txn):
        return cls.appField(txn, 'apas', 'foreign-assets') or []

    @classmethod
    def foreignApps(cls, txn):
        return cls.appField(txn, 'apfa', 'foreign-apps') or []

    @classmethod
    def appField(cls, txn, short, long):
        if short in txn:
            return txn[short]
        elif 'application-transaction' in txn:
            return txn['application-transaction'][long]

    @classmethod
    def innerTxns(cls, txn):
        if 'inner-txns' in txn:
            return txn['inner-txns']
        elif 'dt' in txn:
            itx = txn['dt']['itx']
            txns = [normalizeAlgodTxn(d, txn['confirmed-round'], txn['round-time'])
                for d in itx]
            return txns
        else:
            return []

    @classmethod
    def isAssetTransfer(cls, txn):
        return 'asset-transfer-transaction' in txn or 'xaid' in txn

    @classmethod
    def isPayment(cls, txn):
        return cls.type(txn) == 'pay'