fetcher.calls

Module for fetching and caching contract calls.

The main class of this module is CallsService. It is used for making static calls to Ethereum contracts and caching them for subsequent calls.

Example

from web3cat.fetcher.calls import CallsService
from web3cat.fetcher.erc20_metas import ERC20MetasService

erc20 = ERC20MetasService.create()
dai = erc20.get("Dai")
compound_address = "0x5d3a536E4D6DbD6114cc1Ead35777bAB948E3643"

service = CallsService.create()
response = service.get_call(
    dai.contract.functions.balanceOf(compound_address), block_number=15632000
)
# => going for web3 rpc

response = service.get_call(
    dai.contract.functions.balanceOf(compound_address), block_number=15632000
)
# => serving from cache

#

class web3cat.fetcher.calls.CallsService(calls_repo: CallsRepo, **kwargs)

Bases: Core

Service for making contract static calls.

The sole purpose of this service is to cache web 3 calls and serve them on subsequent calls.

Request/Response flow

        +---------------+                 +-------+ +-----------+
        | CallsService  |                 | Web3  | | CallsRepo |
        +---------------+                 +-------+ +-----------+
---------------  |                             |           |
| Request call |-|                             |           |
|--------------| |                             |           |
                 |                             |           |
                 | Find response               |           |
                 |---------------------------------------->|
                 |                             |           |
                 | If not found: call Web3     |           |
                 |---------------------------->|           |
                 |                             |           |
                 | Save response               |           |
                 |---------------------------------------->|
    -----------  |                             |           |
    | Response |-|                             |           |
    |----------| |                             |           |
                 |                             |           |

Parameters:

• calls_repo – CallsRepo instance

• kwargs – Args for the fetcher.core.Core

static create(**kwargs) → CallsService

Create an instance of CallsService

Parameters:

kwargs – Args for the fetcher.core.Core

Returns:

An instance of CallsService

get_call(call: ContractFunction, block_number: int) → Call

Make contract call specified by parameters.

Parameters:

• call – web3.contract.ContractFunction specifying contract, function and call arguments.

• block_number – get call at this block

Returns:

A fetched fetcher.calls.Call

get_calls(calls: List[ContractFunction], block_numbers: List[int]) → List[Call]

Make a list of contract calls specified by parameters.

Note

Each call is make for each block

Parameters:

• calls – a list of web3.contract.ContractFunction specifying contract, function and call arguments.

• block_number – A list of blocks

Returns:

A fetched fetcher.calls.Call

clear_cache()

Delete all cached entries

class web3cat.fetcher.calls.Call(chain_id: int, address: str, calldata: str, block_number: int, response: Dict[str, Any])

Bases: object

Call represents a static call to Ethereum contract function

chain_id: int

Ethereum chain_id

address: str

Contract address for this call

calldata: str

Calldata for this call

block_number: int

Number of the block for this call

response: Dict[str, Any]

Response received for the call

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

Deserialize from web3cat.database row

Parameters:

row – database row

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

Serialize to database row

Returns:

database row

to_dict() → Dict[str, Any]

Convert Call to dict

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

Create Call from dict

class web3cat.fetcher.calls.CallsRepo(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 Call to database.

find(address: str, calldata: str, from_block: int = 0, to_block: int = 4294967295) → Iterator[Call]

Find all calls in the database.

Parameters:

• chain_id – Ethereum chain_id

• address – Contract address

• calldata – Ethereum calldata

• from_block – starting from this block (inclusive)

• to_block – ending with this block (non-inclusive)

Returns:

List of found calls

find_many(addresses_and_calldatas: List[Tuple[str, str]], blocks: List[int]) → Iterator[Call]

Find all calls in the database.

Parameters:

• addresses_and_calldatas – A list of addresses and calldatas

• blocks – A list of blocks

Returns:

A list of calls

save(calls: List[Call])

Save a list of calls into the database.

Parameters:

calls – List of calls to save

purge()

Clear all database entries