zeth.core.contracts Namespace Reference

Classes
class	InstanceDescription

Functions
int	get_block_number (Any web3)
None	install_sol ()
Any	compile_files (List[str] files, **Any kwargs)
bytes	send_contract_call (Any web3, Any call, str sender_eth_addr, Optional[bytes] sender_eth_private_key=None, Optional[EtherValue] value=None, Optional[int] gas=None)
Any	local_contract_call (Any call, str sender_eth_addr, Optional[EtherValue] value=None, Optional[int] gas=None)
Iterator[Any]	get_event_logs (Any web3, Any instance, str event_name, int start_block, int end_block, Optional[int] batch_size)
Iterator[Any]	get_event_logs_from_tx_receipt (Any instance, str event_name, Any tx_receipt)

Variables
int	SYNC_BLOCKS_PER_BATCH = 1000
	Interface = Dict[str, Any]

Function Documentation

compile_files()

Any zeth.core.contracts.compile_files	(	List[str]	files,
		**Any	kwargs
	)

Wrapper around solcx which ensures the required version of the compiler is
used.

Definition at line 122 of file contracts.py.

def compile_files(files: List[str], **kwargs: Any) -> Any:
    """
    Wrapper around solcx which ensures the required version of the compiler is
    used.
    """
    solcx.set_solc_version(SOL_COMPILER_VERSION)
    return solcx.compile_files(files, optimize=True, **kwargs)
 
 

Here is the caller graph for this function:

get_block_number()

int zeth.core.contracts.get_block_number

(

Any

web3

)

Definition at line 114 of file contracts.py.

def get_block_number(web3: Any) -> int:
    return web3.eth.blockNumber
 
 

Here is the caller graph for this function:

get_event_logs()

Iterator[Any] zeth.core.contracts.get_event_logs	(	Any	web3,
		Any	instance,
		str	event_name,
		int	start_block,
		int	end_block,
		Optional[int]	batch_size
	)

Query the attached node for all events emitted by the given contract
instance, with the given name. Yields an iterator of event-specific objects
to be decoded by the caller.

Definition at line 179 of file contracts.py.

def get_event_logs(
        web3: Any,
        instance: Any,
        event_name: str,
        start_block: int,
        end_block: int,
        batch_size: Optional[int]) -> Iterator[Any]:
    """
    Query the attached node for all events emitted by the given contract
    instance, with the given name. Yields an iterator of event-specific objects
    to be decoded by the caller.
    """
 
    # It is possible to achieve this via the contract interface, with code of
    # the form:
    #
    #   event = instance.events[event_name]
    #   filter = event.createFilter(fromBlock=start_block, toBlock=to_block)
    #   logs = web3.eth.getFilterLogs(filter)
    #
    # However, this creates filters on the host node, which may not be
    # permitted in all configurations. Hence, the code here iterates manually,
    # skpping events with other topics, from the same contract.
 
    contract_address = instance.address
    contract_event = instance.events[event_name]()
 
    event_abi = find_matching_event_abi(instance.abi, event_name=event_name)
    log_topic = event_abi_to_log_topic(cast(Dict[str, Any], event_abi))
 
    batch_size = batch_size or SYNC_BLOCKS_PER_BATCH
    while start_block <= end_block:
        # Filters are *inclusive* wrt "toBlock", hence the -1 here, and +1 to
        # set start_block before iterating.
        to_block = min(start_block + batch_size - 1, end_block)
        filter_params = {
            'fromBlock': start_block,
            'toBlock': to_block,
            'address': contract_address,
        }
        logs = web3.eth.getLogs(filter_params)
        for log in logs:
            if log_topic == log['topics'][0]:
                yield contract_event.processLog(log)
        start_block = to_block + 1
 
 

get_event_logs_from_tx_receipt()

Iterator[Any] zeth.core.contracts.get_event_logs_from_tx_receipt	(	Any	instance,
		str	event_name,
		Any	tx_receipt
	)

Query a transaction receipt for all events emitted by the given contract
instance with a given event name. Yields an iterator of event-specific
objects to be decoded by the caller. This function intentionally avoids
connecting to a node, or creating host-side filters.

Definition at line 226 of file contracts.py.

def get_event_logs_from_tx_receipt(
        instance: Any,
        event_name: str,
        tx_receipt: Any) -> Iterator[Any]:
    """
    Query a transaction receipt for all events emitted by the given contract
    instance with a given event name. Yields an iterator of event-specific
    objects to be decoded by the caller. This function intentionally avoids
    connecting to a node, or creating host-side filters.
    """
    contract_address = instance.address
    contract_event = instance.events[event_name]()
 
    event_abi = find_matching_event_abi(instance.abi, event_name=event_name)
    log_topic = event_abi_to_log_topic(cast(Dict[str, Any], event_abi))
    for log in tx_receipt.logs:
        if log.address == contract_address and log_topic == log['topics'][0]:
            yield contract_event.processLog(log)

Here is the caller graph for this function:

install_sol()

None zeth.core.contracts.install_sol

(

)

Definition at line 118 of file contracts.py.

def install_sol() -> None:
    solcx.install_solc(SOL_COMPILER_VERSION)
 
 

local_contract_call()

Any zeth.core.contracts.local_contract_call	(	Any	call,
		str	sender_eth_addr,
		Optional[EtherValue]	value = None,
		Optional[int]	gas = None
	)

Make a contract call locally on the RPC host and return the result. Does
not create a transaction.

Definition at line 162 of file contracts.py.

def local_contract_call(
        call: Any,
        sender_eth_addr: str,
        value: Optional[EtherValue] = None,
        gas: Optional[int] = None) -> Any:
    """
    Make a contract call locally on the RPC host and return the result. Does
    not create a transaction.
    """
    tx_desc: Dict[str, Union[str, int]] = {'from': sender_eth_addr}
    if value:
        tx_desc["value"] = value.wei
    if gas:
        tx_desc["gas"] = gas
    return call.call(tx_desc)
 
 

send_contract_call()

bytes zeth.core.contracts.send_contract_call	(	Any	web3,
		Any	call,
		str	sender_eth_addr,
		Optional[bytes]	sender_eth_private_key = None,
		Optional[EtherValue]	value = None,
		Optional[int]	gas = None
	)

Broadcast a transaction for a contract call, handling the difference
between hosted keys (sender_eth_private_key is None) and local keys
(sender_eth_private_key is not None). Returns the hash of the broadcast
transaction.

Definition at line 131 of file contracts.py.

def send_contract_call(
        web3: Any,
        call: Any,
        sender_eth_addr: str,
        sender_eth_private_key: Optional[bytes] = None,
        value: Optional[EtherValue] = None,
        gas: Optional[int] = None) -> bytes:
    """
    Broadcast a transaction for a contract call, handling the difference
    between hosted keys (sender_eth_private_key is None) and local keys
    (sender_eth_private_key is not None). Returns the hash of the broadcast
    transaction.
    """
    tx_desc: Dict[str, Union[str, int]] = {'from': sender_eth_addr}
    if value:
        tx_desc["value"] = value.wei
    if gas:
        tx_desc["gas"] = gas
    if sender_eth_private_key:
        tx_desc["gasPrice"] = web3.eth.gasPrice
        tx_desc["nonce"] = web3.eth.getTransactionCount(sender_eth_addr)
        transaction = call.buildTransaction(tx_desc)
        signed_tx = web3.eth.account.signTransaction(
            transaction, sender_eth_private_key)
        print(f"send_contract_call: size={len(signed_tx.rawTransaction)}")
        return web3.eth.sendRawTransaction(signed_tx.rawTransaction)
 
    # Hosted path
    return call.transact(tx_desc)
 
 

Here is the caller graph for this function:

Variable Documentation

Interface

zeth.core.contracts.Interface = Dict[str, Any]

Definition at line 18 of file contracts.py.

SYNC_BLOCKS_PER_BATCH

int zeth.core.contracts.SYNC_BLOCKS_PER_BATCH = 1000

Definition at line 16 of file contracts.py.