How do I decode the transaction's log with Web3.py?

Question

I can filter my events (please see), but instead of that if I know only the transaction hash and if its already deployed, is it possible to obtain and parse the transaction's log data using Web3.py? Please see the solution for web3.js.

=> Is there any web3.eth.abi.decodeLog function under Web3.py?

---

For example, from receipt we can obtain the logs.data.

> tx = '0x5c7e74a21419a6ff825aca9b54df1a86599b4b1ee82e60e6410e8d54cbb58b2c'    
> web3.eth.getTransactionReceipt(tx).logs
    [{
        address: "0x611dc53934550684825f3477ecb68029b1b908f3",
        blockHash: "0x356829b068046b6d44147663f8bd6c187e9507c578d9ec9d69e65d7a248ff368",
        blockNumber: 1180103,
        data: "0x0000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000b000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e0000000000000000000000000000000000000000000000000000000000000002e516d52736142454763717851634a6242784369314c4e39697a3562444147445752364878375a76577167716d64520000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000007536369656e636500000000000000000000000000000000000000000000000000",
        logIndex: 0,
        removed: false,
        topics: ["0x711007a4be830d68a6d2a7b6546227b676bf9e7d7a0e3c248552ed72cfec2441", "0x0000000000000000000000004e4a0750350796164d8defc442a712b7557bf282"],
        transactionHash: "0x5c7e74a21419a6ff825aca9b54df1a86599b4b1ee82e60e6410e8d54cbb58b2c",
        transactionIndex: 0
    }]

Example log I have is, where string has a dynamic size. If possible, I want to decode web3.eth.getTransactionReceipt(tx).logs[<index>]['data] and web3.eth.getTransactionReceipt(tx).logs[<index>]['topics].

event LogJob(address indexed clusterAddress,
             string indexed key,
             uint index,
             uint8 fileID,
             string desc,
             string hash);

Answer 1

From Web3.py documentation:

myContract = web3.eth.contract(address=contract_address, abi=contract_abi)
tx_hash = myContract.functions.myFunction().transact()
receipt = web3.eth.getTransactionReceipt(tx_hash)
logs = myContract.events.myEvent().processReceipt(receipt)

ContractEvent provides methods to interact with contract events. Positional and keyword arguments supplied to the contract event subclass will be used to find the contract event by signature.

Answer 2

I was struggling with same issue. When you are subscribed to contract's topic log, aditional searching for data using transaction hash is adding latency to processing.

I have written this class for log decoding using Web3.py:

from typing import Any, Dict, cast, Union

from eth_typing import HexStr
from eth_utils import event_abi_to_log_topic
from hexbytes import HexBytes
from web3._utils.abi import get_abi_input_names, get_abi_input_types, map_abi_data
from web3._utils.normalizers import BASE_RETURN_NORMALIZERS
from web3.contract import Contract


class EventLogDecoder:
    def __init__(self, contract: Contract):
        self.contract = contract
        self.event_abis = [abi for abi in self.contract.abi if abi['type'] == 'event']
        self._sign_abis = {event_abi_to_log_topic(abi): abi for abi in self.event_abis}
        self._name_abis = {abi['name']: abi for abi in self.event_abis}

    def decode_log(self, result: Dict[str, Any]):
        data = [t[2:] for t in result['topics']]
        data += [result['data'][2:]]
        data = "0x" + "".join(data)
        return self.decode_event_input(data)

    def decode_event_input(self, data: Union[HexStr, str], name: str = None) -> Dict[str, Any]:
        # type ignored b/c expects data arg to be HexBytes
        data = HexBytes(data)  # type: ignore
        selector, params = data[:32], data[32:]

        if name:
            func_abi = self._get_event_abi_by_name(event_name=name)
        else:
            func_abi = self._get_event_abi_by_selector(selector)

        names = get_abi_input_names(func_abi)
        types = get_abi_input_types(func_abi)

        decoded = self.contract.w3.codec.decode(types, cast(HexBytes, params))
        normalized = map_abi_data(BASE_RETURN_NORMALIZERS, types, decoded)

        return dict(zip(names, normalized))

    def _get_event_abi_by_selector(self, selector: HexBytes) -> Dict[str, Any]:
        try:
            return self._sign_abis[selector]
        except KeyError:
            raise ValueError("Event is not presented in contract ABI.")

    def _get_event_abi_by_name(self, event_name: str) -> Dict[str, Any]:
        try:
            return self._name_abis[event_name]
        except KeyError:
            raise KeyError(f"Event named '{event_name}' was not found in contract ABI.")

To decode log, you must create contract instance first:

w3 = Web3(provider=Web3.HTTPProvider("https://rpc.ankr.com/polygon"))
contract = w3.eth.contract(address=address, abi=contract_abi)
eld = EventLogDecoder(contract)

Now you get this data from topic's subscription. (I have used: Uniswap V3: USDC/ETH 0.05% Pool on polygon):

result = {
'address': '0x45dda9cb7c25131df268515131f647d726f50608', 
'topics': ['0xc42079f94a6350d7e6235f29174924f928cc2ac818eb64fed8004e115fbcca67', '0x0000000000000000000000004c60051384bd2d3c01bfc845cf5f4b44bcbe9de5', '0x000000000000000000000000a1bc0292191192b6f03fc0045a26592ce4719bba'], 
'data': '0xfffffffffffffffffffffffffffffffffffffffffffffffffffffffff4af79b90000000000000000000000000000000000000000000000000199430ac4f396b900000000000000000000000000000000000060344446a57b653f75b746ecd6e80000000000000000000000000000000000000000000000001db24a4d833c90130000000000000000000000000000000000000000000000000000000000031603', 
'blockNumber': '0x25235e9', 
'transactionHash': '0x03c147dd2c1e1e65b01628b1adc1e268db7188b7c4de2d38e1964a96f3395073',
'transactionIndex': '0x4e', 
'blockHash': '0x57ebe7b3dd2a3c0bf9cc7191bbaeb49f74043f2da81bb98220874909b040b6a0',
'logIndex': '0x142', 
'removed': False
}

When result is passed to eld.decode_log`:

out = eld.decode_log(result)
print(out)

you get the decoded log:

{
  'amount0': -189826631,
  'amount1': 115196979007690425,
  'liquidity': 2139854469729128467,
  'recipient': '0xa1bc0292191192b6F03FC0045a26592CE4719BBa',
  'sender': '0x4C60051384bd2d3C01bfc845Cf5F4b44bcbE9de5',
  'sqrtPriceX96': 1951252316788244045616425051412200,
  'tick': 202243,
}

EventLogDecoder is generic and works for any SmartContract on ETH. In example I have used contract Polygon network but on other ETH compatible network this should work as well.

Answer 3

The above solutions are excellent, but only work on legacy transactions. To decode any type of transaction (legacy, EIP1559, EIP2930, etc) please rely on the py-evm package.

Here's a working example, taken from Marc Garreau's excellent blog:

from eth.vm.forks.arrow_glacier.transactions import ArrowGlacierTransactionBuilder as TransactionBuilder
from eth_utils import (
  encode_hex,
  to_bytes,
)

# 1) the signed transaction to decode:
original_hexstr = '0xf8a910850684ee180082e48694a0b86991c6218b36c1d19d4a2e9eb0ce3606eb4880b844a9059cbb000000000000000000000000b8b59a7bc828e6074a4dd00fa422ee6b92703f9200000000000000000000000000000000000000000000000000000000010366401ba0e2a4093875682ac6a1da94cdcc0a783fe61a7273d98e1ebfe77ace9cab91a120a00f553e48f3496b7329a7c0008b3531dd29490c517ad28b0e6c1fba03b79a1dee'

# 2) convert the hex string to bytes:
signed_tx_as_bytes = to_bytes(hexstr=original_hexstr)

# 3) deserialize the transaction using the latest transaction builder:
decoded_tx = TransactionBuilder().decode(signed_tx_as_bytes)
print(decoded_tx.__dict__)
# {'type_id': 2, '_inner': DynamicFeeTransaction(chain_id=1, nonce=4, max_priority_fee_per_gas=2500000000, max_fee_per_gas=118977454018, gas=45000, to=b'\xe9\xcb\...', value=0, data=b'', access_list=(), y_parity=1, r=23532..., s=28205...)}

# 4) the (human-readable) sender's address:
sender = encode_hex(decoded_tx.sender)
print(sender)
# 0x240ab...

EIP-1559 transaction

The transaction in the above example is a legacy transaction. To test the code with an EIP-1559 transaction, please use the following line:

original_hexstr = '0x02f86a0a75843b9aca0084412e386682520894240abf8acb28205b92d39181e2dab0b0d8ea6e5d6480c080a0a942241378fafc80670c6dde3c39ba5b1c4f992cd26fa263e7bbc253696c9035a008cf3399808cb511f0171be2ba766ea5c9d424a97dae3fb24685c440eeefc4fa'

Web3.py

As of Sep 2023, there's no equivalent function in the web3.py library. The matter of implementing it is being discussed in a Github issue I have created recently.

Answer 4

Along the lines of the previous answer (and because the API must have changed slightly), below is a simplified answer in two versions given that you have an ABI in the form of:

abi = {
    "type": "event",
    "name": "MyEvent",
        "inputs": [
            {
                "type": "uint256",
                "name": "my_arg_1",
                "indexed": False
            },
            {
                "type": "bytes32",
                "name": "my_arg_2",
                "indexed": False
            }
        ],
        "anonymous": False                                                     
    }

which can be constructed by hand or fetched from a contract object. Different frameworks present the log structure in slightly different ways but somewhere in the result there will be a field data which is a hex string e.g.:

"data":"0x00000000000000000000000000000000000000000000000000000000000000403c2eb04e43c3c52eab2f704a8b0cc7078c355cf8eb623986f..."

1. If you are already using a full web3 object created similar to

w3 = Web3(Web3.HTTPProvider("http://127.0.0.1:8545")

then use:

from web3 import Web3

def decode_log_data(abi, data):
    types = [i['type'] for i in abi['inputs']]
    names = [i['name'] for i in abi['inputs']]

    bb = bytes.fromhex(data[2:]) # Remember to jump over 0x...                                             
    values = w3.codec.decode(types, bb)

    return dict(zip(names, values))

2. If you are building a utility or otherwise do not have access to a full blown web3 object, use the eth_abi version:

import eth_abi

def decode_log_data(abi, data):
    types = [i['type'] for i in abi['inputs']]
    names = [i['name'] for i in abi['inputs']]

    bb = bytes.fromhex(data[2:]) # Remember to jump over 0x...                                             
    values = eth_abi.decode(types, bb)                                                                    
    
    return dict(zip(names, values))

The logic to match an incoming log topic to the proper ABI entry is the same as answer given above. Looking up by name is straightforward; lookup by function signature is facilitated by event_abi_to_log_topic which will conveniently take the event name and argument types, make a string signature from them, then keccak the result.