dojo.network

Network manages everything related to blockchain forking and connection.

This includes the retrieval of contract objects from the blockchain network. And anything related to brownie

class BaseBackend(abc.ABC):

The Backend class handles comms with the local fork network.

BaseBackend(*, chain: dojo.common.constants.Chain, port: Optional[int] = None)

Initialize the class.

Parameters

port: Port to use for the forked chain. If None, a free port will be found.

def mine_block(self, force: bool = True) -> None:

Force the backend to mine a block.

def contract_call( self, function: web3.contract.contract.ContractFunction, function_params: Union[list[Any], tuple[()], tuple[Any, ...]], call_params: Optional[dict[str, Any]] = None) -> Any:

Submit a contract read-only call.

Parameters

function: Web3.py contract function to call.
function_params: Parameters to pass to the contract function.
call_params: Call parameters.

def contract_transact_with_postprocess( self, function: web3.contract.contract.ContractFunction, function_params: Union[list[Any], tuple[()], tuple[Any, ...]], transact_params: Optional[dict[str, Any]], process_reciept: Callable[..., Any]) -> web3.types.PendingTx:

Submit a contract write transaction.

Parameters

function: Web3.py contract function to call.
function_params: Parameters to pass to the contract function.
transact_params: Transaction parameters.

Raises

Exception: if called from the live backend, due to the polling frequency being very high

def contract_transact( self, function: web3.contract.contract.ContractFunction, function_params: Union[list[Any], tuple[()], tuple[Any, ...]], transact_params: Optional[dict[str, Any]] = None) -> web3.types.PendingTx:

Submit a contract write transaction.

Parameters

function: Web3.py contract function to call.
function_params: Parameters to pass to the contract function.
transact_params: Transaction parameters.

@abstractmethod

def connect( self, date_range: tuple[datetime.datetime, datetime.datetime], backend: str = 'anvil') -> None:

Instantiate and connect to the local network.

Parameters

date_range: simulation date range.
backend: Type of backend to use. Can be one of ['hardhat', 'anvil'].

def save_state(self) -> None:

Save the current state of the local forked network.

def load_state(self) -> None:

Load the state of the local forked network.

Requires save_state() to have been called first.

@lru_cache(maxsize=None)

def web3_contract( self, address: eth_typing.evm.Address, abi: str) -> web3.contract.contract.Contract:

Get a web3 contract object directly via web3.

Parameters

address: deployed contract address.
abi: contract abi ID.

def register_contract(self, name: str, contract: web3.contract.contract.Contract) -> None:

Parameters

name: dojo contract name.
contract: the deployed contract.

Raises

ValueError: If contract is already registered.

def get_contract(self, name: str) -> web3.contract.contract.Contract:

Get a web3 contract object.

Parameters

name: dojo contract name.

Raises

ValueError: If contract is not registered.

def set_code( self, contract_address: eth_typing.evm.HexAddress, code: hexbytes.main.HexBytes) -> None:

Set the code at the specified address immediately, without mining.

def set_storage( self, contract_address: eth_typing.evm.HexAddress, storage_address: eth_typing.evm.HexAddress, storage_value: eth_typing.encoding.HexStr) -> None:

Set the storage at the specified addresses immediately without mining.

def deploy_live_contract( self, protocol: str, name: str, args: list[typing.Any] = [], bytecode: Optional[str] = None) -> web3.contract.contract.Contract:

Deploy an existing live contract to the dojo network.

Parameters

protocol: protocol name (e.g. UniswapV3, Tokens etc.).
name: dojo contract name.
args: contract constructor arguments.

Returns

the deployed contract

def lookup(self, name_or_address: Union[eth_typing.evm.ChecksumAddress, str]) -> str:

Lookup a contract name by address or vice versa.

Parameters

name_or_address: contract name or address.

@abstractmethod

def mint_token(self, token: str, quantity: Union[int, decimal.Decimal]) -> None:

Mint tokens for local backend.

def rpc_url(self) -> str:

Get the rpc url for this backend.

def block_to_timestamp(self, block_number: int) -> web3.types.Timestamp:

Get the timestamp of a block.

This method is very slow, consider batching requests. batch_block_to_datetime is currently under development.

def block_to_datetime(self, block_number: int) -> datetime.datetime:

Get the datetime of a block.

This method is very slow, consider batching requests. batch_block_to_datetime is currently under development.

def closest_block_to(self, datetime_: datetime.datetime) -> int:

Get the closest block to the date given.

Raises: BlockBeforeDeploymentOrInFuture: Block not found for date.

def closest_block_lt(self, datetime_: datetime.datetime) -> int:

Get the closest block less than the date given.

class ForkedBackend(dojo.network.BaseBackend):

The ForkedBackend class handles comms with a forked network.

ForkedBackend(chain: dojo.common.constants.Chain, port: Optional[int] = None)

Initialize the class.

Parameters

port: Port to use for the forked chain. If None, a free port will be found.

def connect( self, date_range: tuple[datetime.datetime, datetime.datetime], backend: str = 'anvil') -> None:

Instantiate and connect to the dojo network.

Parameters

date_range: simulation date range.
backend: Type of backend to use. Can be one of 'hardhat' or 'anvil'.

Raises

ValueError: if requested backend is not supported.

def disconnect(self) -> None:

Disconnect and close the connection to the local network.

def mint_token(self, token: str, quantity: Union[int, decimal.Decimal]) -> None:

Empty function for minting tokens on forked backend.

class LocalBackend(dojo.network.BaseBackend):

The LocalBackend class handles comms with a local network.

LocalBackend(*, chain: dojo.common.constants.Chain, port: Optional[int] = None)

Initialize the class.

Parameters

port: Port to use for the forked chain. If None, a free port will be found.

def connect( self, date_range: tuple[datetime.datetime, datetime.datetime], backend: str = 'anvil') -> None:

Instantiate and connect to the dojo network.

Parameters

date_range: simulation date range.
backend: Type of backend to use. Can be one of 'hardhat'.

Raises

ValueError: if requested backend is not supported.

def mint_token(self, token: str, quantity: Union[int, decimal.Decimal]) -> None:

Mint tokens to the account that deployed the contract.

Parameters

token: Name of the token to mint.
quantity: quantity to mint in human or machine readable format.

class LiveBackend(dojo.network.BaseBackend):

The live backend connects to a live network.

LiveBackend(chain: dojo.common.constants.Chain)

Instantiate the class.

def connect( self, date_range: tuple[datetime.datetime, datetime.datetime], backend: str = 'anvil'):

Instantiate and connect to a live network.

Parameters

date_range: simulation date range.
backend: Type of backend to use. Can be one of 'hardhat'.

def contract_call( self, function: web3.contract.contract.ContractFunction, function_params: Union[list, tuple], call_params: Optional[dict] = None):

Submit a contract read-only call.

Parameters

function: Web3.py contract function to call.
function_params: Parameters to pass to the contract function.
call_params: Call parameters.

def contract_transact( self, function: web3.contract.contract.ContractFunction, function_params: Union[list, tuple], transact_params: Optional[dict] = None) -> web3.types.PendingTx:

Submit a contract write transaction.

Parameters

function: Web3.py contract function to call.
function_params: Parameters to pass to the contract function.
transact_params: Transaction parameters.

def disconnect(self):

Disconnect and close the connection to the local network.

def save_state(self):

This backend doesn't support saving state.

def load_state(self):

Calls self.connect() instead.

def mint_token(self, token: str, quantity: Union[int, decimal.Decimal]) -> None:

Empty function for minting tokens on live backend.