9

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);
1
  • The transaction exists on my private chain, hence it does not exist on the public ethereum chain that's why.
    – alper
    Nov 29, 2018 at 21:23

4 Answers 4

15

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.

2
5

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.

3
  • 2
    In my version web3.py 6.1.0 using python 3.10, this line coded = self.contract.web3.codec.decode(types, cast(HexBytes, params)) produces the error: AttributeError: 'Contract' object has no attribute 'web3' Jun 3, 2023 at 19:04
  • 1
    seems like contract.web3 is the underlying web3. I just added it to the constructor of the EventLogDecoder. @BuzzMoschetti
    – Michuelnik
    Jun 5, 2023 at 16:11
  • I have adjusted the code according to new version web3 library. It seems that attribute is no longer called web3 but just w3. After this slight change everything works (tested on web3==6.4.0) Jun 6, 2023 at 20:44
2

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.

1

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))
  1. 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.

2
  • thank you. I have used names = [i['name'] or f"{i['type']}{i}" for i in abi['inputs']] to handle anonymous arguments, but this was exactly what I needed.
    – hornobster
    Jan 15 at 19:24
  • Errata Corrige: names = [i['name'] or f"{i['type']}{idx}" for idx, i in enumerate(abi['inputs'])]
    – hornobster
    Jan 15 at 19:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.