fetcher.blocks

Module for fetching and caching blocks metadata from web3.

The main class of this module is BlocksService. It is used for retrieving block metadata from web3 and storing it in cache so that subsequent requests are returned from cache.

Example

from datetime import datetime, timezone
from web3cat.fetcher.blocks import BlocksService

service = BlocksService.create()
latest_block = service.latest_block

new_year = datetime(2022, 1, 1, tzinfo=timezone.utc)
new_year_block = service.get_latest_blocks_by_timestamps(int(new_year.timestamp()))[0]
# => block 13916166

new_year_block = service.get_latest_blocks_by_timestamps(int(new_year.timestamp()))[0]
# cached result

new_year_block = service.get_blocks(new_year_block.number)
# cached result

class web3cat.fetcher.blocks.BlocksService(blocks_repo: BlocksRepo, **kwargs)

Bases: Core

Service for fetching and caching Ethereum block data.

Request/Response flow

        +---------------+            +-------+ +-------------+
        | BlocksService |            | Web3  | | BlocksRepo  |
        +---------------+            +-------+ +-------------+
-----------------  |                        |            |
| Request blocks |-|                        |            |
|----------------| |                        |            |
                   |                        |            |
                   | Get blocks             |            |
                   |------------------------------------>|
                   |                        |            |
                   | Get missing blocks     |            |
                   |----------------------->|            |
                   |                        |            |
                   | Save missing blocks    |            |
                   |------------------------------------>|
      -----------  |                        |            |
      | Response |-|                        |            |
      |----------| |                        |            |
                   |                        |            |

Parameters:

• blocks_repo – An instance of fetcher.blocks.BlocksRepo

• kwargs – Args for the fetcher.core.Core

See also

fetcher.core.Core for defining the block grid and how timestamps are approximated.

static create(**kwargs) → BlocksService

Create an instance of BlocksService

Parameters:

kwargs – Args for the fetcher.core.Core

Returns:

An instance of BlocksService

property latest_block: Block

Latest block from Ethereum (this value is cached on the first call)

refresh_latest_block()

Refresh latest_block (the value is cached by default)

get_latest_block_at_timestamp(timestamp: int) → web3cat.fetcher.blocks.block.Block | None

Get the first block after a timestamp.

Parameters:

timestamp – UNIX timestamp

Returns:

First block after timestamp, None if the block doesn’t exist

get_latest_blocks_by_timestamps(block_timestamps: Union[int, List[int]]) → List[Block]

Get a list of latest blocks as of timestamp.

Parameters:

block_timestamps – Unix timestamps

Returns:

A list of blocks in the same order

get_blocks(numbers: Union[int, List[int]]) → List[Block]

Get blocks by numbers.

Parameters:

number – block numbers

Returns:

Blocks with these numbers. None if the block doesn’t exist.

clear_cache()

Delete all cached entries

class web3cat.fetcher.blocks.Block(chain_id: int, number: int, timestamp: int)

Bases: object

Ethereum block data

chain_id: int

Ethereum chain_id

number: int

Block number

timestamp: int

Block timestamp

static from_row(row: Tuple[int, int, int]) → Block

Deserialize from web3cat.database row

Parameters:

row – database row

to_row() → Tuple[int, int, int]

Serialize to database row

Returns:

database row

static from_dict(dct: Dict[str, Any])

Create Block from dict

to_dict() → Dict[str, Any]

Convert Block to dict

class web3cat.fetcher.blocks.BlocksRepo(rpc: Optional[str] = None, cache_path: Optional[str] = None, block_grid_step: int = 1000, w3: Optional[Web3] = None, conn: Optional[Connection] = None)

Bases: Core

Reading and writing Block to database.

get_block_after_timestamp(timestamp: int) → web3cat.fetcher.blocks.block.Block | None

Get the first block strictly after the timestamp in database.

Parameters:

• chain_id – Ethereum chain_id

• timestamp – UNIX timestamp

Returns:

First block after the timestamp, None if the block doesn’t exist

get_block_before_timestamp(timestamp: int) → web3cat.fetcher.blocks.block.Block | None

Get the first block at or before the timestamp in database.

Parameters:

• chain_id – Ethereum chain_id

• timestamp – UNIX timestamp, UTC+0

Returns:

First block before the timestamp, None if the block doesn’t exist

find(blocks: Union[int, List[int]]) → Iterator[Block]

Find blocks by number

Warning

The order of the blocks are not preserved. Duplicates are eliminated.

Parameters:

blocks – block number or a list of block numbers

Returns:

A list of found blocks

all() → Iterator[Block]

All blocks in the database

save(blocks: List[Block])

Save a set of blocks into the database.

Parameters:

blocks – List of blocks to save

purge()

Clear all database entries