Made preprocess method that will add missing fields to NftEvent from web3

datasets/nfts/nfts/cli.py

@@ -1,9 +1,16 @@
-from moonstreamdb.db import yield_db_session_ctx
+import os
 
-from .materialize import get_rows, EventType
+from moonstreamdb.db import yield_db_session_ctx
+from web3 import Web3
+
+from .materialize import get_rows, EventType, preproccess
 
 if __name__ == "__main__":
+    web3_path = os.environ.get("MOONSTREAM_IPC_PATH")
+    web3_client = Web3(Web3.HTTPProvider(web3_path))
+
     with yield_db_session_ctx() as db_session:
         rows = get_rows(db_session, EventType.TRANSFER)
+        rows = preproccess(db_session, web3_client, rows)
         for row in rows:
             print(row)

datasets/nfts/nfts/materialize.py

@@ -9,6 +9,7 @@ from moonstreamdb.models import (
     EthereumBlock,
 )
 from sqlalchemy.orm import Session
+from web3 import Web3
 
 
 @dataclass
@@ -35,6 +36,22 @@ class NFTEvent:
     timestamp: Optional[int] = None
 
 
+def preproccess(
+    db_session: Session, web3_client: Web3, nft_events: List[NFTEvent]
+) -> List[NFTEvent]:
+    """
+    Adds block number, value, timestamp from web3 if they are None (because that transaction is missing in db)
+    """
+    for nft_event in nft_events:
+        if nft_event.block_number is None:
+            transaction = web3_client.eth.get_transaction(nft_event.transaction_hash)
+            nft_event.value = transaction["value"]
+            nft_event.block_number = transaction["blockNumber"]
+            block = web3_client.eth.get_block(transaction["blockNumber"])
+            nft_event.timestamp = block["timestamp"]
+    return nft_events
+
+
 def get_rows(
     db_session: Session, event_type: EventType, bounds: Optional[BlockBounds] = None
 ) -> List[NFTEvent]:
@@ -44,12 +61,19 @@ def get_rows(
             EthereumAddress.address,
             EthereumLabel.label_data,
             EthereumLabel.transaction_hash,
+            EthereumTransaction.value,
+            EthereumTransaction.block_number,
+            EthereumBlock.timestamp,
         )
         .join(EthereumAddress, EthereumLabel.address_id == EthereumAddress.id)
         .outerjoin(
             EthereumTransaction,
             EthereumLabel.transaction_hash == EthereumTransaction.hash,
         )
+        .outerjoin(
+            EthereumBlock,
+            EthereumTransaction.block_number == EthereumBlock.block_number,
+        )
         .filter(EthereumLabel.label == event_type.value)
         .limit(10)
     )
@@ -61,6 +85,9 @@ def get_rows(
             from_address=label_data["from"],
             to_address=label_data["to"],
             transaction_hash=transaction_hash,
+            value=value,
+            block_number=block_number,
+            timestamp=timestamp,
         )
-        for label, address, label_data, transaction_hash in query
+        for label, address, label_data, transaction_hash, value, block_number, timestamp in query
     ]

datasets/nfts/sample.env

@@ -0,0 +1,2 @@
+export MOONSTREAM_DB_URI=""
+export MOONSTREAM_IPC_PATH=null