53 lines
1.4 KiB
Python
53 lines
1.4 KiB
Python
import logging
|
|
import os
|
|
import signal
|
|
import time
|
|
|
|
from web3 import Web3
|
|
|
|
from mev_inspect.block import get_latest_block_number
|
|
from mev_inspect.provider import get_base_provider
|
|
|
|
|
|
logging.basicConfig(filename="app.log", level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GracefulKiller:
|
|
"""
|
|
handle sigint / sigterm gracefully
|
|
taken from https://stackoverflow.com/a/31464349
|
|
"""
|
|
|
|
signal_names = {signal.SIGINT: "SIGINT", signal.SIGTERM: "SIGTERM"}
|
|
|
|
def __init__(self):
|
|
self.kill_now = False
|
|
signal.signal(signal.SIGINT, self.exit_gracefully)
|
|
signal.signal(signal.SIGTERM, self.exit_gracefully)
|
|
|
|
def exit_gracefully(self, signum, frame): # pylint: disable=unused-argument
|
|
signal_name = self.signal_names[signum]
|
|
logger.info(f"Received {signal_name} signal")
|
|
logger.info("Cleaning up resources. End of process")
|
|
self.kill_now = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
rpc = os.getenv("RPC_URL")
|
|
if rpc is None:
|
|
raise RuntimeError("Missing environment variable RPC_URL")
|
|
|
|
logger.info("Starting...")
|
|
|
|
killer = GracefulKiller()
|
|
base_provider = get_base_provider(rpc)
|
|
w3 = Web3(base_provider)
|
|
|
|
while not killer.kill_now:
|
|
latest_block_number = get_latest_block_number(w3)
|
|
logger.info(f"Latest block: {latest_block_number}")
|
|
time.sleep(5)
|
|
|
|
logger.info("Stopping...")
|