blockchain-etl · andrewmccall · Dec 6, 2020 · Jan 9, 2021 · Jun 22, 2021 · Aug 4, 2021
diff --git a/.gitignore b/.gitignore
@@ -47,3 +47,6 @@ coverage.xml
 .venv
 venv/
 ENV/
+
+# etl
+/last_synced_block.txt
diff --git a/.travis.yml b/.travis.yml
@@ -2,15 +2,15 @@ language: python
 dist: xenial
 matrix:
   include:
-    - python: "3.5"
-      env: TOX_POSARGS="-e py35"
     - python: "3.6"
       env: TOX_POSARGS="-e py36"
     - python: "3.7"
       env: TOX_POSARGS="-e py37"
     - python: "3.8"
       env: TOX_POSARGS="-e py38"
+    - python: "3.9"
+      env: TOX_POSARGS="-e py39"
 install:
   - travis_retry pip install tox
 script:
-  - tox $TOX_POSARGS
+  - travis_wait tox $TOX_POSARGS
diff --git a/README.md b/README.md
@@ -1,6 +1,6 @@
 # Ethereum ETL
 
-[![Build Status](https://travis-ci.com/blockchain-etl/ethereum-etl.png)](https://travis-ci.com/blockchain-etl/ethereum-etl)
+[![Build Status](https://app.travis-ci.com/blockchain-etl/ethereum-etl.svg?branch=develop)](https://travis-ci.com/github/blockchain-etl/ethereum-etl)
 [![Join the chat at https://gitter.im/ethereum-eth](https://badges.gitter.im/ethereum-etl.svg)](https://gitter.im/ethereum-etl/Lobby?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 [![Telegram](https://img.shields.io/badge/telegram-join%20chat-blue.svg)](https://t.me/joinchat/GsMpbA3mv1OJ6YMp3T5ORQ)
 [![Discord](https://img.shields.io/badge/discord-join%20chat-blue.svg)](https://discord.gg/wukrezR)

diff --git a/blockchainetl/jobs/exporters/gcs_item_exporter.py b/blockchainetl/jobs/exporters/gcs_item_exporter.py
@@ -0,0 +1,111 @@
+# MIT License
+#
+# Copyright (c) 2020 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import json
+import logging
+from collections import defaultdict
+
+from google.cloud import storage
+
+
+def build_block_bundles(items):
+    blocks = defaultdict(list)
+    transactions = defaultdict(list)
+    logs = defaultdict(list)
+    token_transfers = defaultdict(list)
+    traces = defaultdict(list)
+    for item in items:
+        item_type = item.get('type')
+        if item_type == 'block':
+            blocks[item.get('number')].append(item)
+        elif item_type == 'transaction':
+            transactions[item.get('block_number')].append(item)
+        elif item_type == 'log':
+            logs[item.get('block_number')].append(item)
+        elif item_type == 'token_transfer':
+            token_transfers[item.get('block_number')].append(item)
+        elif item_type == 'trace':
+            traces[item.get('block_number')].append(item)
+        else:
+            logging.info(f'Skipping item with type {item_type}')
+
+    block_bundles = []
+    for block_number in sorted(blocks.keys()):
+        if len(blocks[block_number]) != 1:
+            raise ValueError(f'There must be a single block for a given block number, was {len(blocks[block_number])} for block number {block_number}')
+        block_bundles.append({
+            'block': blocks[block_number][0],
+            'transactions': transactions[block_number],
+            'logs': logs[block_number],
+            'token_transfers': token_transfers[block_number],
+            'traces': traces[block_number],
+        })
+
+    return block_bundles
+
+
+class GcsItemExporter:
+
+    def __init__(
+            self,
+            bucket,
+            path='blocks',
+            build_block_bundles_func=build_block_bundles):
+        self.bucket = bucket
+        self.path = normalize_path(path)
+        self.build_block_bundles_func = build_block_bundles_func
+        self.storage_client = storage.Client()
+
+    def open(self):
+        pass
+
+    def export_items(self, items):
+        block_bundles = self.build_block_bundles_func(items)
+
+        for block_bundle in block_bundles:
+            block = block_bundle.get('block')
+            if block is None:
+                raise ValueError('block_bundle must include the block field')
+            block_number = block.get('number')
+            if block_number is None:
+                raise ValueError('block_bundle must include the block.number field')
+
+            destination_blob_name = f'{self.path}/{block_number}.json'
+
+            bucket = self.storage_client.bucket(self.bucket)
+            blob = bucket.blob(destination_blob_name)
+            blob.upload_from_string(json.dumps(block_bundle))
+            logging.info(f'Uploaded file gs://{self.bucket}/{destination_blob_name}')
+
+    def close(self):
+        pass
+
+
+def normalize_path(p):
+    if p is None:
+        p = ''
+    if p.startswith('/'):
+        p = p[1:]
+    if p.endswith('/'):
+        p = p[:len(p) - 1]
+
+    return p
diff --git a/blockchainetl/jobs/exporters/google_pubsub_item_exporter.py b/blockchainetl/jobs/exporters/google_pubsub_item_exporter.py
@@ -29,9 +29,19 @@
 
 class GooglePubSubItemExporter:
 
-    def __init__(self, item_type_to_topic_mapping, message_attributes=('item_id', 'item_timestamp')):
+    def __init__(self, item_type_to_topic_mapping, message_attributes=(),
+            batch_max_bytes=1024 * 5, batch_max_latency=1, batch_max_messages=1000,
+            enable_message_ordering=False):
         self.item_type_to_topic_mapping = item_type_to_topic_mapping
-        self.publisher = create_publisher()
+
+        self.batch_max_bytes = batch_max_bytes
+        self.batch_max_latency = batch_max_latency
+        self.batch_max_messages = batch_max_messages
+
+        self.enable_message_ordering = enable_message_ordering
+
+        self.publisher = self.create_publisher()
+
         self.message_attributes = message_attributes
 
     def open(self):
@@ -46,7 +56,7 @@ def export_items(self, items):
             # details = "channel is in state TRANSIENT_FAILURE"
             # https://stackoverflow.com/questions/55552606/how-can-one-catch-exceptions-in-python-pubsub-subscriber-that-are-happening-in-i?noredirect=1#comment97849067_55552606
             logging.info('Recreating Pub/Sub publisher.')
-            self.publisher = create_publisher()
+            self.publisher = self.create_publisher()
             raise e
 
     @timeout_decorator.timeout(300)
@@ -66,7 +76,8 @@ def export_item(self, item):
             topic_path = self.item_type_to_topic_mapping.get(item_type)
             data = json.dumps(item).encode('utf-8')
 
-            message_future = self.publisher.publish(topic_path, data=data, **self.get_message_attributes(item))
+            ordering_key = 'all' if self.enable_message_ordering else ''
+            message_future = self.publisher.publish(topic_path, data=data, ordering_key=ordering_key, **self.get_message_attributes(item))
             return message_future
         else:
             logging.warning('Topic for item type "{}" is not configured.'.format(item_type))
@@ -80,15 +91,15 @@ def get_message_attributes(self, item):
 
         return attributes
 
-    def close(self):
-        pass
-
+    def create_publisher(self):
+        batch_settings = pubsub_v1.types.BatchSettings(
+            max_bytes=self.batch_max_bytes,
+            max_latency=self.batch_max_latency,
+            max_messages=self.batch_max_messages,
+        )
 
-def create_publisher():
-    batch_settings = pubsub_v1.types.BatchSettings(
-        max_bytes=1024 * 5,  # 5 kilobytes
-        max_latency=1,  # 1 second
-        max_messages=1000,
-    )
+        publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=self.enable_message_ordering)
+        return pubsub_v1.PublisherClient(batch_settings=batch_settings, publisher_options=publisher_options)
 
-    return pubsub_v1.PublisherClient(batch_settings)
+    def close(self):
+        pass
diff --git a/blockchainetl/jobs/exporters/kafka_exporter.py b/blockchainetl/jobs/exporters/kafka_exporter.py
@@ -0,0 +1,54 @@
+import collections
+import json
+import logging
+
+from kafka import KafkaProducer
+
+from blockchainetl.jobs.exporters.converters.composite_item_converter import CompositeItemConverter
+
+
+class KafkaItemExporter:
+
+    def __init__(self, output, item_type_to_topic_mapping, converters=()):
+        self.item_type_to_topic_mapping = item_type_to_topic_mapping
+        self.converter = CompositeItemConverter(converters)
+        self.connection_url = self.get_connection_url(output)
+        print(self.connection_url)
+        self.producer = KafkaProducer(bootstrap_servers=self.connection_url)
+
+    def get_connection_url(self, output):
+        try:
+            return output.split('/')[1]
+        except KeyError:
+            raise Exception('Invalid kafka output param, It should be in format of "kafka/127.0.0.1:9092"')
+
+    def open(self):
+        pass
+
+    def export_items(self, items):
+        for item in items:
+            self.export_item(item)
+
+    def export_item(self, item):
+        item_type = item.get('type')
+        if item_type is not None and item_type in self.item_type_to_topic_mapping:
+            data = json.dumps(item).encode('utf-8')
+            print(data)
+            return self.producer.send(self.item_type_to_topic_mapping[item_type], value=data)
+        else:
+            logging.warning('Topic for item type "{}" is not configured.'.format(item_type))
+
+    def convert_items(self, items):
+        for item in items:
+            yield self.converter.convert_item(item)
+
+    def close(self):
+        pass
+
+
+def group_by_item_type(items):
+    result = collections.defaultdict(list)
+    for item in items:
+        result[item.get('type')].append(item)
+
+    return result
diff --git a/blockchainetl/jobs/exporters/multi_item_exporter.py b/blockchainetl/jobs/exporters/multi_item_exporter.py
@@ -0,0 +1,42 @@
+# MIT License
+#
+# Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+
+class MultiItemExporter:
+    def __init__(self, item_exporters):
+        self.item_exporters = item_exporters
+
+    def open(self):
+        for exporter in self.item_exporters:
+            exporter.open()
+
+    def export_items(self, items):
+        for exporter in self.item_exporters:
+            exporter.export_items(items)
+
+    def export_item(self, item):
+        for exporter in self.item_exporters:
+            exporter.export_item(item)
+
+    def close(self):
+        for exporter in self.item_exporters:
+            exporter.close()
diff --git a/blockchainetl/logging_utils.py b/blockchainetl/logging_utils.py
@@ -7,3 +7,5 @@ def logging_basic_config(filename=None):
         logging.basicConfig(level=logging.INFO, format=format, filename=filename)
     else:
         logging.basicConfig(level=logging.INFO, format=format)
+
+    logging.getLogger('ethereum_dasm.evmdasm').setLevel(logging.ERROR)
diff --git a/docs/commands.md b/docs/commands.md
@@ -207,11 +207,15 @@ You can tune `--batch-size`, `--max-workers` for performance.
 - This command outputs blocks, transactions, logs, token_transfers to the console by default.
 - Entity types can be specified with the `-e` option, 
 e.g. `-e block,transaction,log,token_transfer,trace,contract,token`.
-- Use `--output` option to specify the Google Pub/Sub topic or Postgres database where to publish blockchain data, 
+- Use `--output` option to specify the Google Pub/Sub topic, Postgres database or GCS bucket where to publish blockchain data, 
     - For Google PubSub: `--output=projects/<your-project>/topics/crypto_ethereum`. 
     Data will be pushed to `projects/<your-project>/topics/crypto_ethereum.blocks`, `projects/<your-project>/topics/crypto_ethereum.transactions` etc. topics.
     - For Postgres: `--output=postgresql+pg8000://<user>:<password>@<host>:<port>/<database_name>`, 
-    e.g. `--output=postgresql+pg8000://postgres:admin@127.0.0.1:5432/ethereum`. 
+    e.g. `--output=postgresql+pg8000://postgres:admin@127.0.0.1:5432/ethereum`.
+    - For GCS:  `--output=gs://<bucket_name>`. Make sure to install and initialize `gcloud` cli.
+    - For Kafka:  `--output=kafka/<host>:<port>`, e.g. `--output=kafka/127.0.0.1:9092`
+    - Those output types can be combined with a comma e.g. `--output=gs://<bucket_name>,projects/<your-project>/topics/crypto_ethereum`
+
     The [schema](https://github.com/blockchain-etl/ethereum-etl-postgres/tree/master/schema) 
     and [indexes](https://github.com/blockchain-etl/ethereum-etl-postgres/tree/master/indexes) can be found in this 
     repo [ethereum-etl-postgres](https://github.com/blockchain-etl/ethereum-etl-postgres). 

diff --git a/docs/dockerhub.md b/docs/dockerhub.md
@@ -1,7 +1,7 @@
 # Uploading to Docker Hub
 
 ```bash
-ETHEREUMETL_VERSION=1.7.2
+ETHEREUMETL_VERSION=1.10.1
 docker build -t ethereum-etl:${ETHEREUMETL_VERSION} -f Dockerfile .
 docker tag ethereum-etl:${ETHEREUMETL_VERSION} blockchainetl/ethereum-etl:${ETHEREUMETL_VERSION}
 docker push blockchainetl/ethereum-etl:${ETHEREUMETL_VERSION}

diff --git a/docs/exporting-the-blockchain.md b/docs/exporting-the-blockchain.md
@@ -14,8 +14,10 @@ export the data ~40 times faster, you will need to set up a local Ethereum node:
 Make sure it downloaded the blocks that you need by executing `eth.syncing` in the JS console.
 You can export blocks below `currentBlock`,
 there is no need to wait until the full sync as the state is not needed (unless you also need contracts bytecode
-and token details; for those you need to wait until the full sync).
-
+and token details; for those you need to wait until the full sync). Note that you may need to wait for another day or 
+   two for the node to download the states. See this issue https://github.com/blockchain-etl/ethereum-etl/issues/265#issuecomment-970451522. 
+Make sure to set `--txlookuplimit 0` if you use geth.
+
 1. Install Ethereum ETL: `> pip3 install ethereum-etl`
 
 1. Export all:

diff --git a/docs/limitations.md b/docs/limitations.md
@@ -1,8 +1,5 @@
 # Limitation
 
-- In case the contract is a proxy, which forwards all calls to a delegate, interface detection doesn’t work,
-which means `is_erc20` and `is_erc721` will always be false for proxy contracts and they will be missing in the `tokens`
-table.
 - The metadata methods (`symbol`, `name`, `decimals`, `total_supply`) for ERC20 are optional, so around 10% of the
 contracts are missing this data. Also some contracts (EOS) implement these methods but with wrong return type,
 so the metadata columns are missing in this case as well.

diff --git a/ethereumetl/cli/__init__.py b/ethereumetl/cli/__init__.py
@@ -19,6 +19,10 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
+
+from blockchainetl.logging_utils import logging_basic_config
+logging_basic_config()
+
 import click
 
 from ethereumetl.cli.export_all import export_all
@@ -44,7 +48,7 @@
 
 
 @click.group()
-@click.version_option(version='1.7.2')
+@click.version_option(version='1.10.1')
 @click.pass_context
 def cli(ctx):
     pass