GraphQLReorganisationMonitor

Documentation for eth_defi.event_reader.reorganisation_monitor.GraphQLReorganisationMonitor Python class.

class GraphQLReorganisationMonitor

Bases: eth_defi.event_reader.reorganisation_monitor.ReorganisationMonitor

Watch blockchain for reorgs using GoEthereum /graphql API.

Parameters

  • graphql_url – Give this or existing HTTPProvider

  • provider – Give this or graphql_url

Attributes summary

check_depth

How many blocks we replay from the blockchain to detect any chain organisations

last_block_read

Last block served by update_chain() in the duty cycle

max_cycle_tries

How many times we try to re-read data from the blockchain in the case of reorganisation.

reorg_wait_seconds

How long we allow our node to catch up in the case there has been a change in the chain tip.

Methods summary

__init__([graphql_url, provider])

param graphql_url

add_block(record)

Add new block to header tracking.

check_block_reorg(block_number, block_hash)

Check that newly read block matches our record.

fetch_block_data(start_block, end_block)

Read the new block headers.

figure_reorganisation_and_new_blocks([max_range])

Compare the local block database against the live data from chain.

get_block_by_number(block_number)

Get block header data for a specific block number from our memory buffer.

get_block_timestamp(block_number)

Return UNIX UTC timestamp of a block.

get_block_timestamp_as_pandas(block_number)

Return UNIX UTC timestamp of a block.

get_last_block_live()

Get the chain tip using GraphQL.

get_last_block_read()

Get the number of the last block served by update_chain().

has_data()

Do we have any data available yet.

load_initial_block_headers([block_count, ...])

Get the initial block buffer filled up.

load_pandas(df)

Load block header data from Pandas data frame.

restore(block_map)

Restore the chain state from a saved data.

skip_to_block(block_number)

Skip scanning initial chain and directly start from a certain block.

to_pandas([partition_size])

Convert the data to Pandas DataFrame format for storing.

truncate(latest_good_block)

Delete data after a block number because chain reorg happened.

update_chain()

Update the internal memory buffer of block headers from the blockchain node.
__init__(graphql_url=None, provider=None, **kwargs)
Parameters

  • graphql_url (Optional[str]) – Give this or existing HTTPProvider

  • provider (Optional[web3.providers.rpc.rpc.HTTPProvider]) – Give this or graphql_url

get_last_block_live()

Get the chain tip using GraphQL.

Return type

int

fetch_block_data(start_block, end_block)

Read the new block headers.

Parameters

  • start_block – The first block where to read (inclusive)

  • end_block – The block where to read (inclusive)

Return type

Iterable[eth_defi.event_reader.block_header.BlockHeader]

add_block(record)

Add new block to header tracking.

Blocks must be added in order.

Parameters

record (eth_defi.event_reader.block_header.BlockHeader) –

check_block_reorg(block_number, block_hash)

Check that newly read block matches our record.

  • Called during the event reader

  • Event reader gets the block number and hash with the event

  • We have initial block_map in memory, previously buffered in

  • We check if any of the blocks in the block map have different values on our event produces -> in this case we know there has been a chain reorganisation

If we do not have records, ignore.

Raises

ChainReorganisationDetected – When any if the block data in our internal buffer does not match those provided by events.

Parameters

  • block_number (int) –

  • block_hash (str) –

Return type

Optional[int]

figure_reorganisation_and_new_blocks(max_range=1000000)

Compare the local block database against the live data from chain.

Spot the differences in (block number, block header) tuples and determine a chain reorg.

Parameters

max_range (Optional[int]) –

Abort if we need to scan more than this amount of blocks.

This is because giving too long block range to scan is likely to take forever on non-graphql nodes.

Set None to ignore.

Raises

ChainReorganisationDetected – When any if the block data in our internal buffer does not match those provided by events.

get_block_by_number(block_number)

Get block header data for a specific block number from our memory buffer.

Parameters

block_number (int) –

Return type

eth_defi.event_reader.block_header.BlockHeader

get_block_timestamp(block_number)

Return UNIX UTC timestamp of a block.

Parameters

block_number (int) –

Return type

int

get_block_timestamp_as_pandas(block_number)

Return UNIX UTC timestamp of a block.

Parameters

block_number (int) –

Return type

pandas.Timestamp

get_last_block_read()

Get the number of the last block served by update_chain().

Return type

int

has_data()

Do we have any data available yet.

Return type

bool

load_initial_block_headers(block_count=None, start_block=None, tqdm=None, save_callable=None)

Get the initial block buffer filled up.

You can call this during the application start up, or when you start the chain. This interface is designed to keep the application on hold until new blocks have been served.

Parameters

  • block_count (Optional[int]) –

    How many latest block to load

    Give start_block or block_count.

  • start_block (Optional[int]) –

    What is the first block to read.

    Give start_block or block_count.

  • tqdm (Optional[Type[tqdm.std.tqdm]]) – To display a progress bar

  • save_callable (Optional[Callable]) –

    Save after every block.

    Called after every block.

    TODO: Hack. Design a better interface.

Returns

The initial block range to start to work with

Return type

Tuple[int, int]

load_pandas(df)

Load block header data from Pandas data frame.

Parameters

df (pandas.DataFrame) – Pandas DataFrame exported with to_pandas().

restore(block_map)

Restore the chain state from a saved data.

Parameters

block_map (dict) – Block number -> Block header dictionary

skip_to_block(block_number)

Skip scanning initial chain and directly start from a certain block.

Parameters

block_number (int) –

to_pandas(partition_size=0)

Convert the data to Pandas DataFrame format for storing.

Parameters

partition_size (int) –

To partition the outgoing data.

Set 0 to ignore.

Return type

pandas.DataFrame

truncate(latest_good_block)

Delete data after a block number because chain reorg happened.

Parameters

latest_good_block (int) – Delete all data starting after this block (exclusive)

update_chain()

Update the internal memory buffer of block headers from the blockchain node.

  • Do several attempt to read data (as a fork can cause other forks can cause fork)

  • Give up after some time if we detect the chain to be in a doom loop

Returns

What block range the consumer application should read.

What we think about the chain state.

Return type

eth_defi.event_reader.reorganisation_monitor.ChainReorganisationResolution