Skip to content

Instantly share code, notes, and snippets.

@GraphicalDot
Created January 8, 2022 13:01
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save GraphicalDot/d28b708f23a52a8cf0dd4da74d88e762 to your computer and use it in GitHub Desktop.
Save GraphicalDot/d28b708f23a52a8cf0dd4da74d88e762 to your computer and use it in GitHub Desktop.
"""
Moralis provides a great funstionality where they provide you a mongoDb instance to save events data from blockchain
blocks. The interface to listen to events is pretty easy as compared to other solutions like Graph or writing your own
sockets code The other great functionality which moralis provides is that you can run your own local EVM blockchain and
sync it with the remore moralis mongodb instance and save all the events data in mongodb
The small issue is: Whenever you redeploy your code or make changes to events,
you will have to change the sync settings and update the new contract addresses or events ABI.
This script will help you to get all your events abi from the .json artifact file of your contract and update the
EventSync table on remote moralis mongoInstance with events ABI and create/update existing tables with new contract
addresses if required.
Make sure, the ip from which you will be running this script is whitelisted on the Moralis mongodb instance.
This script has to be run for each contract.
"""
from pathlib import Path
import json
import pymongo
from web3 import Web3
import datetime
w3 = Web3()
ARTIFACT_FILE_PATH = "/home/graphicaldot/Programs/ama.fans/py-server-backend/artifacts/AmaFansCore.json"
MONGO_HOST_IP = ""
MONGO_PORT = 56728
CHAIN_ID = '0x539' # This is for local ganache blockchain, see docs for other chain ids
CONTRACT_ADDRESS = "0x175D36b48Ea313b7BA4e55A7Ae153216897eB87C" # Change it
SYNC_HISTORICAL = False # change it if you want to sync historical events also
ORIGIN = "admin"
def events_data():
assert Path(ARTIFACT_FILE_PATH).suffix == '.json'
abi = ""
with open(ARTIFACT_FILE_PATH) as f:
data = f.read()
json_data = json.loads(data)
abi = json_data["abi"]
return [contract_object for contract_object in abi if contract_object["type"] == "event"]
def create_topic_hash(abi):
inputs = ",".join([e["type"] for e in abi["inputs"]])
event_name = abi["name"]
combined = event_name + "(" + inputs + ")"
return w3.keccak(text=combined).hex()
def make_document(abi):
return {
'chainId': CHAIN_ID,
"address": CONTRACT_ADDRESS,
"topic": create_topic_hash(abi),
"tableName": abi["name"],
"sync_historical": SYNC_HISTORICAL,
"abi": abi,
"origin": ORIGIN,
"last_sync_error": "",
"createdAt": int(datetime.datetime.utcnow().timestamp()*1000),
"updateAt": int(datetime.datetime.utcnow().timestamp()*1000),
"is_syncing": False
}
connection = pymongo.MongoClient(MONGO_HOST_IP, MONGO_PORT)
events_collection = connection["parse"]["EventSync"]
bulk = events_collection.initialize_unordered_bulk_op()
abi_data = [make_document(abi) for abi in events_data()]
print (abi_data)
for document in abi_data:
bulk.insert(document)
bulk.execute()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment