Service Components

Components that perform logic services for the Block Crawler

MemoryBlockTimeCache

A simple, memory-based cache to store block times. It may be pre-loaded as well as iterated upon. It’s primary reason for being is to provide a cache for the BlockTimeService.

Example:

    block_time_cache = MemoryBlockTimeCache()
    await block_time_cache.set(1, 12345)
    block_1_time = await block_time_cache.get(1)
    print("Block 1 Time:", block_1_time)
    block_2_time = await block_time_cache.get(2)
    print("Block 2 Time:", block_2_time)

    print("\nCache Data:")
    for block_id, time in block_time_cache:
        print(f"  Block/Time: {block_id}/{time}")

BlockTimeService

A service to get the timestamp for a block. The service will utilize the provided RPC client to get the timestamp from block data on the blockchain if the timestamp for the block does not exist in the provided cache. Once retrieved it will be stored in the cache.

Example:

    block_time_cache = MemoryBlockTimeCache()
    async with EvmRpcClient(rpc_uri, stats_service) as rpc_client:
        block_time_service = BlockTimeService(block_time_cache, rpc_client)
        block_1_time = await block_time_service.get_block_timestamp(HexInt(1))

Complete Simple Example

import asyncio

import click

from blockcrawler.core.stats import StatsService
from blockcrawler.core.types import HexInt
from blockcrawler.evm.rpc import EvmRpcClient
from blockcrawler.evm.services import MemoryBlockTimeCache, BlockTimeService


async def run(rpc_uri):
    block_time_cache = MemoryBlockTimeCache()
    await block_time_cache.set(1, 12345)
    block_1_time = await block_time_cache.get(1)
    print("Block 1 Time:", block_1_time)
    block_2_time = await block_time_cache.get(2)
    print("Block 2 Time:", block_2_time)

    stats_service = StatsService()
    async with EvmRpcClient(rpc_uri, stats_service) as rpc_client:
        block_time_service = BlockTimeService(block_time_cache, rpc_client)
        block_1_time = await block_time_service.get_block_timestamp(HexInt(1))
        print("Block 1 Time:", block_1_time.int_value)
        block_2_time = await block_time_service.get_block_timestamp(HexInt(2))
        print("Block 2 Time:", block_2_time.int_value)
        print("Get Block Calls:", stats_service.get_count(EvmRpcClient.STAT_REQUEST_SENT))
        print("Block 2 Time:", block_2_time.int_value)
        print("Get Block Calls:", stats_service.get_count(EvmRpcClient.STAT_REQUEST_SENT))

    print("\nCache Data:")
    for block_id, time in block_time_cache:
        print(f"  Block/Time: {block_id}/{time}")


@click.command()
@click.argument("RPC_URI")
def command(rpc_uri: str):
    """
    Provide the RPC URI for Websockets to process the example
    """
    asyncio.run(run(rpc_uri))


if __name__ == "__main__":
    command()

Executing the above example providing a RPC API Websocket URI will result in something similar to the following being displayed in the console:

Block 1 Time: 12345

Block 2 Time: None

Block 1 Time: 12345

Block 2 Time: 1438270017

Get Block Calls: 1

Block 2 Time: 1438270017

Get Block Calls: 1

Cache Data:

Block/Time: 1/12345

Block/Time: 2/1438270017

Download the example and try it yourself.