Implement DAG builder for parsing Ethereum logs

[medvedev1088]

The DAG builder is simply a Python function that generates an Airflow DAG given a list of parameters. An example of a DAG builder can be found here.

The input for the DAG builder for parsing Ethereum logs is a list of json files, each of which is a table definition file. An example table definition file is given below:

{
    "table": {
        "project_name": "blockchain-etl",
        "dataset_name": "zeroex",
        "table_name": "Exchange_event_LogFill",
        "table_description": "Lorem ipsum.",
        "schema": [
            {
                "name": "block_timestamp",
                "description": "Lorem ipsum.",
                "type": "TIMESTAMP"
            },
            {
                "name": "maker",
                "type": "STRING"
            },
            {
                "name": "taker",
                "type": "STRING"
            },
            {
                "name": "feeRecipient",
                "type": "STRING"
            },
            {
                "name": "makerToken",
                "type": "STRING"
            },
            {
                "name": "takerToken",
                "type": "STRING"
            },
            {
                "name": "filledMakerTokenAmount",
                "type": "STRING"
            },
            {
                "name": "filledTakerTokenAmount",
                "type": "STRING"
            },
            {
                "name": "paidMakerFee",
                "type": "STRING"
            },
            {
                "name": "paidTakerFee",
                "type": "STRING"
            },
            {
                "name": "tokens",
                "type": "STRING"
            },
            {
                "name": "orderHash",
                "type": "STRING"
            }
        ]
    },
    "parser": {
        "type": "log",
        "contract_address": "0x12459c951127e0c374ff9105dda097662a027093",
        "abi": {
            "anonymous": false,
            "inputs": [
                {
                    "indexed": true,
                    "name": "maker",
                    "type": "address"
                },
                {
                    "indexed": false,
                    "name": "taker",
                    "type": "address"
                },
                {
                    "indexed": true,
                    "name": "feeRecipient",
                    "type": "address"
                },
                {
                    "indexed": false,
                    "name": "makerToken",
                    "type": "address"
                },
                {
                    "indexed": false,
                    "name": "takerToken",
                    "type": "address"
                },
                {
                    "indexed": false,
                    "name": "filledMakerTokenAmount",
                    "type": "uint256"
                },
                {
                    "indexed": false,
                    "name": "filledTakerTokenAmount",
                    "type": "uint256"
                },
                {
                    "indexed": false,
                    "name": "paidMakerFee",
                    "type": "uint256"
                },
                {
                    "indexed": false,
                    "name": "paidTakerFee",
                    "type": "uint256"
                },
                {
                    "indexed": true,
                    "name": "tokens",
                    "type": "bytes32"
                },
                {
                    "indexed": false,
                    "name": "orderHash",
                    "type": "bytes32"
                }
            ],
            "name": "LogFill",
            "type": "event"
        },
        "field_mapping": {
            "TODO": "if necessary define rules for mapping abi fields to BigQuery table columns"
        }
    }
}

The output is an Airflow DAG:

• For each table definition file a PythonOperator task should be created which executes a BigQuery query job with destination table (an example can be found here).
• The SQL for the BigQuery job should be generated from table and parser definitions, using Jinja template (an example Jinja template can be found here. An example log parsing query can be found here).

[gitcoinbot]

Issue Status: 1. Open 2. Started 3. Submitted 4. Done

---

This issue now has a funding of 500.0 DAI (500.0 USD @ $1.0/DAI) attached to it.

• If you would like to work on this issue you can 'start work' on the Gitcoin Issue Details page.
• Want to chip in? Add your own contribution here.
• Questions? Checkout Gitcoin Help or the Gitcoin Slack
• $150,218.76 more funded OSS Work available on the Gitcoin Issue Explorer

[gitcoinbot]

Issue Status: 1. Open 2. Started 3. Submitted 4. Done

---

Work has been started.

These users each claimed they can complete the work by 2 weeks from now.
Please review their action plans below:

1) igetgames has applied to start work (Funders only: approve worker | reject worker).

I will implement the DAG builder according to ethereum-etl-airflow/issues#17 and the README.
2) askeluv has started work.

Create an Airflow DAG builder which can later be used to parse Ethereum logs given an ABI. For example 0x transactions and ENS events.

Learn more on the Gitcoin Issue Details page.

[askeluv]

Should we also add a field event_topic inside the parser object, so we can filter out the right log rows? Or is there perhaps some easy way to convert from the event name to the topic / signature?

E.g. Fill events for 0x should be topic 0x0bcc4c97732e47d9946f229edb95f5b6323f601300e4690de719993f3c371129

[medvedev1088]

@askeluv There is an example here for converting event signature e.g. Transfer(address,address,uint256) to it's hash: https://github.com/blockchain-etl/ethereum-etl/blob/develop/ethereumetl/cli/get_keccak_hash.py. It uses the keccak function from eth_utils lib.

I've also found these 2 functions in eth_utils/abi.py:

def event_signature_to_log_topic(event_signature: str) -> bytes:
    return keccak(text=event_signature.replace(" ", ""))


def event_abi_to_log_topic(event_abi: Dict[str, Any]) -> bytes:
    event_signature = _abi_to_signature(event_abi)
    return event_signature_to_log_topic(event_signature)

[gitcoinbot]

Issue Status: 1. Open 2. Started 3. Submitted 4. Done

---

Work for 500.0 DAI (500.0 USD @ $1.0/DAI) has been submitted by:

1. @askeluv

@ceresstation please take a look at the submitted work:

• PR by @askeluv

---

• Learn more on the Gitcoin Issue Details page
• Want to chip in? Add your own contribution here.
• Questions? Checkout Gitcoin Help or the Gitcoin Slack
• $152,094.30 more funded OSS Work available on the Gitcoin Issue Explorer

[gitcoinbot]

Issue Status: 1. Open 2. Started 3. Submitted 4. Done

---

The funding of 500.0 DAI (500.0 USD @ $1.0/DAI) attached to this issue has been approved & issued to @askeluv.

• Learn more on the Gitcoin Issue Details page
• Questions? Checkout Gitcoin Help or the Gitcoin Slack
• $151,033.53 more funded OSS Work available on the Gitcoin Issue Explorer