Add BLOCKS table schema

blockchainetl/jobs/exporters/converters/int_to_decimal_field_converter.py

@@ -35,6 +35,8 @@
 from decimal import Decimal
 
 
+# Large ints are not handled correctly by pg8000 so we use Decimal instead:
+# https://github.com/mfenniak/pg8000/blob/412eace074514ada824e7a102765e37e2cda8eaa/pg8000/core.py#L1703
 class IntToDecimalFieldConverter:
 
     def __init__(self):

blockchainetl/streaming/postgres_utils.py

@@ -4,11 +4,13 @@ from sqlalchemy.dialects.postgresql import insert
 def create_insert_statement_for_table(table):
     insert_stmt = insert(table)
 
-    insert_stmt = insert_stmt.on_conflict_do_update(
-        index_elements=[column.name for column in table.columns if column.primary_key],
-        set_={
-            column.name: insert_stmt.excluded[column.name] for column in table.columns if not column.primary_key
-        }
-    )
+    primary_key_fields = [column.name for column in table.columns if column.primary_key]
+    if primary_key_fields:
+        insert_stmt = insert_stmt.on_conflict_do_update(
+            index_elements=primary_key_fields,
+            set_={
+                column.name: insert_stmt.excluded[column.name] for column in table.columns if not column.primary_key
+            }
+        )
 
     return insert_stmt

ethereumetl/streaming/item_exporter_creator.py

@@ -40,11 +40,12 @@ def create_item_exporter(output):
         from blockchainetl.streaming.postgres_utils import create_insert_statement_for_table
         from blockchainetl.jobs.exporters.converters.unix_timestamp_field_converter import UnixTimestampFieldConverter
         from blockchainetl.jobs.exporters.converters.int_to_decimal_field_converter import IntToDecimalFieldConverter
-        from ethereumetl.streaming.postgres_tables import TOKEN_TRANSFERS
+        from ethereumetl.streaming.postgres_tables import BLOCKS, TOKEN_TRANSFERS
 
         item_exporter = PostgresItemExporter(
             output, item_type_to_insert_stmt_mapping={
-                'token_transfer': create_insert_statement_for_table(TOKEN_TRANSFERS)
+                'block': create_insert_statement_for_table(BLOCKS),
+                'token_transfer': create_insert_statement_for_table(TOKEN_TRANSFERS),
             },
             converters=[UnixTimestampFieldConverter(), IntToDecimalFieldConverter()])
     else:

ethereumetl/streaming/postgres_tables.py

@@ -24,6 +24,30 @@ from sqlalchemy import Table, Column, BigInteger, String, Numeric, MetaData, TIM
 
 metadata = MetaData()
 
+# SQL schema is here https://github.com/blockchain-etl/ethereum-etl-postgres/tree/master/schema
+
+BLOCKS = Table(
+    'blocks', metadata,
+    Column('timestamp', TIMESTAMP),
+    Column('number', BigInteger),
+    Column('hash', String, primary_key=True),
+    Column('parent_hash', String),
+    Column('nonce', String),
+    Column('sha3_uncles', String),
+    Column('logs_bloom', String),
+    Column('transactions_root', String),
+    Column('state_root', String),
+    Column('receipts_root', String),
+    Column('miner', String),
+    Column('difficulty', Numeric(38)),
+    Column('total_difficulty', Numeric(38)),
+    Column('size', BigInteger),
+    Column('extra_data', String),
+    Column('gas_limit', BigInteger),
+    Column('gas_used', BigInteger),
+    Column('transaction_count', BigInteger),
+)
+
 TOKEN_TRANSFERS = Table(
     'token_transfers', metadata,
     Column('token_address', String),