Merge pull request #134 from blockchain-etl/feat/workaround-non-batch…

…-geth-requests

Workaround non-batch `geth` requests

airflow/requirements_airflow.txt

@@ -5,4 +5,4 @@
 discord-webhook==0.14.0
 eth-rlp==0.2.1             # Fixes install conflicts issue in Composer
 eth-utils==1.8.4           # Fixes install conflicts issue in Composer
-polygon-etl==0.3.1
+polygon-etl==0.3.2

cli/polygonetl/cli/__init__.py

@@ -43,7 +43,7 @@
 
 
 @click.group()
-@click.version_option(version='0.3.1')
+@click.version_option(version='0.3.2')
 @click.pass_context
 def cli(ctx):
     pass

cli/polygonetl/jobs/export_geth_traces_job.py

@@ -44,6 +44,7 @@ def __init__(
         self.start_block = start_block
         self.end_block = end_block
 
+        self.batch_size = batch_size
         self.batch_web3_provider = batch_web3_provider
 
         self.batch_work_executor = BatchWorkExecutor(batch_size, max_workers)
@@ -71,7 +72,14 @@ def _export(self):
 
     def _export_batch(self, block_number_batch):
         trace_block_rpc = list(generate_trace_block_by_number_json_rpc(block_number_batch))
-        response = self.batch_web3_provider.make_batch_request(json.dumps(trace_block_rpc))
+
+        # For nodes that don't support batch debug_traceBlockByNumber
+        if self.batch_size == 1:
+            request = json.dumps(trace_block_rpc[0])
+            response = [self.batch_web3_provider.make_batch_request(request)]
+        else:
+            request = json.dumps(trace_block_rpc)
+            response = self.batch_web3_provider.make_batch_request(request)
 
         for response_item in response:
             block_number = response_item.get('id')

cli/setup.py

@@ -11,7 +11,7 @@ def read(fname):
 
 setup(
     name="polygon-etl",
-    version="0.3.1",
+    version="0.3.2",
     author="Evgeny Medvedev",
     author_email="[email protected]",
     description="Tools for exporting Polygon blockchain data to CSV or JSON",

cli/tests/polygonetl/job/mock_batch_web3_provider.py

@@ -34,10 +34,17 @@ def __init__(self, read_resource):
     def make_batch_request(self, text):
         batch = json.loads(text)
         web3_response = []
-        for req in batch:
-            method = req["method"]
-            params = req["params"]
+        if isinstance(batch, dict):
+            method = batch["method"]
+            params = batch["params"]
             file_name = build_file_name(method, params)
             file_content = self.read_resource(file_name)
-            web3_response.append(json.loads(file_content))
+            return json.loads(file_content)
+        else:
+            for req in batch:
+                method = req["method"]
+                params = req["params"]
+                file_name = build_file_name(method, params)
+                file_content = self.read_resource(file_name)
+                web3_response.append(json.loads(file_content))
         return web3_response

streaming/charts/polygon-etl-streaming/Chart.yaml

@@ -2,4 +2,4 @@ apiVersion: v1
 appVersion: "1.0"
 description: A Helm chart to deploy polygon ETL streaming apps
 name: polygon-etl-streaming
-version: 0.3.1
+version: 0.3.2

streaming/charts/polygon-etl-streaming/values.yaml

@@ -33,7 +33,7 @@ init:
 stream:
   image:
     repository: blockchainetl/polygon-etl
-    tag: 0.3.1
+    tag: 0.3.2
     pullPolicy: IfNotPresent
   resources:
     requests:

streaming/example_values.yaml

@@ -1,7 +1,7 @@
 stream:
   image:
     repository: blockchainetl/polygon-etl
-    tag: 0.3.1
+    tag: 0.3.2
 config:
   PROVIDER_URI: "grpcs://api.mainnet.polygon.one:443"
   STREAM_OUTPUT: "projects/<your_project>/topics/crypto_polygon"