y package 

classmethod from_explorer(address, as_proxy_for=None, owner=None, silent=False, persist=True)[source]

Create a new Contract instance by fetching the ABI from a block explorer.

This method is useful for interacting with contracts that are not part of the current project, as it automatically fetches the contract’s ABI from a block explorer.

Parameters:

address (str) – The address of the deployed contract.
as_proxy_for (str | None) – The address of the implementation contract if this is a proxy contract.
owner (AccountsType | None) – The account that owns this contract instance.
silent (bool) – Whether to suppress console output during the process.
persist (bool) – Whether to persist the contract data to brownie’s db for future use.

Returns:

A new Contract instance.

Return type:

async get_code(block=None)[source]

Get the bytecode of the contract at a specific block.

Parameters:: block (optional) – The block number at which to get the bytecode. Defaults to latest block.
Return type:: HexBytes

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> await contract.get_code()
HexBytes('0x...')

get_method(calldata)

Parameters:: calldata (str)
Return type:: str | None

get_method_object(calldata)

Given a calldata hex string, returns a ContractMethod object.

Parameters:: calldata (str)
Return type:: _ContractMethod | None

classmethod get_solc_version(compiler_str, address)[source]

Return the solc compiler version either from the passed compiler string or try to find the latest available patch semver compiler version.

Parameters:

compiler_str (str) – The compiler string passed from the contract metadata.
address (str) – The contract address to check for.

Return type:

Version

has_method(method, return_response=False)[source]

Check if the contract has a specific method.

Parameters:

method (str) – The name of the method to check for.
return_response (optional) – If True, return the response of the method call instead of a boolean. Default False.

Return type:

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> contract.has_method("name()")
True

async has_methods(methods, _func=<built-in function all>)[source]

Check if the contract has all the specified methods.

Parameters:

methods (List[str]) – A list of method names to check for.
_func (optional) – The function to use for combining the results (either all() or any()). Default all().

Return type:

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> await contract.has_methods(["name()", "symbol()"])
True

info()

Display NatSpec documentation for this contract.

Return type:: None

classmethod remove_deployment(address=None, alias=None)[source]

Removes this contract from the internal deployments db with the passed address or alias.

Parameters:

address (str | None) – An address to apply
alias (str | None) – An alias to apply

Return type:

Tuple[Dict | None, Dict | None]

set_alias(alias, persist=True)[source]

Apply a unique alias this object. The alias can be used to restore the object in future sessions.

Parameters:

alias (str | None) – An alias to apply. If None, any existing alias is removed.
persist (bool)

Return type:

None

property abi: List

property alias: str | None

events: ContractEvents

A container for various event types associated with this contract.

Provides a convenient way to query contract events with minimal code.

signatures: Dict[Method, Signature]: A dictionary mapping method names to their corresponding signatures.

topics: Dict[str, str]: A dictionary mapping event names to their corresponding topics.

verified = True: True if the contract is verified on this network’s block explorer. False otherwise.

class y.contracts.ContractEvents[source]

Bases: ContractEvents

__getattr__(name)[source]

Parameters:: name (str)
Return type:: Events

__getitem__(event_name)

Parameters:: event_name (str)
Return type:: Type[BaseContractEvent]

__hasattr__(event_name)

Parameters:: event_name (str)
Return type:: bool

__init__(contract)[source]

Parameters:: contract (_DeployedContractBase)

__iter__()

Iterate over supported

Returns:: Iterable of ContractEvent
Return type:: Iterable[Type[BaseContractEvent]]

get_sequence(from_block, to_block=None, event_type=None)[source]

Returns the logs of events of type ‘event_type’ that occurred between the blocks ‘from_block’ and ‘to_block’. If ‘event_type’ is not specified, it retrieves the occurrences of all events in the contract.

Parameters:

from_block (int) – The block from which to search for events that have occurred.
to_block (int, optional) – The block on which to stop searching for events.
specified (if not)
block (it is set to the most recently mined)
None. (between the specified blocks. Defaults to)
event_type (ContractEvent, str, optional) – Type or name of the event to be searched
None.

Returns:

[list]: List of events of type ‘event_type’ that occurred between: ’from_block’ and ‘to_block’.
else:: event_logbook [dict]: Dictionary of events of the contract that occurred between ‘from_block’ and ‘to_block’.

Return type:

if ‘event_type’ is specified

listen(event_name, timeout=0)[source]

Creates a listening Coroutine object ending whenever an event matching ‘event_name’ occurs. If timeout is superior to zero and no event matching ‘event_name’ has occured, the Coroutine ends when the timeout is reached.

The Coroutine return value is an AttributeDict filled with the following fields :

‘event_data’ (AttributeDict): The event log receipt that was caught.
‘timed_out’ (bool): False if the event did not timeout, else True

If the ‘timeout’ parameter is not passed or is inferior or equal to 0, the Coroutine listens indefinitely.

Parameters:

event_name (str) – Name of the event to be listened to.
timeout (float, optional) – Timeout value in seconds. Defaults to 0.

Returns:

Awaitable object listening for the event matching ‘event_name’.

Return type:

Coroutine

subscribe(event_name, callback, delay=2.0)[source]

Subscribe to event with a name matching ‘event_name’, calling the ‘callback’ function on new occurrence giving as parameter the event log receipt.

Parameters:

event_name (str) – Name of the event to subscribe to.
callback (Callable[[AttributeDict], None]) – Function called whenever an event occurs.
delay (float, optional) – Delay between each check for new events. Defaults to 2.0.

Return type:

None

y.contracts.Contract_erc20(address)[source]

Create a Contract instance for an ERC20 token.

This function uses the standard ERC20 ABI instead of fetching the contract ABI from the block explorer.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the ERC20 token.
Returns:: A Contract instance for the ERC20 token.
Return type:: Contract

Example

>>> contract = Contract_erc20("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> contract.name()
'Dai Stablecoin'

y.contracts.Contract_with_erc20_fallback(address)[source]

Create a Contract instance for an address, falling back to an ERC20 token if the contract is not verified.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the contract or ERC20 token.
Returns:: A Contract instance for the contract address.
Return type:: Contract

Example

>>> contract = Contract_with_erc20_fallback("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> contract.symbol()
'DAI'

Get the build name of a contract.
Args:
address: The address of the contract. return_None_on_failure (optional): If True, return None if the build name cannot be determined instead of raising an Exception. Default False.

Example:
>>> await build_name("0x6B175474E89094C44Da98b954EedeAC495271d0F")
'Dai Stablecoin'

Since build_name is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
return_None_on_failure (bool)

Return type:

str

Determine the block when a contract was created using binary search. NOTE Requires access to historical state. Doesn’t account for CREATE2 or SELFDESTRUCT.
Args:
address: The address of the contract. when_no_history_return_0: If True, return 0 when no history is found instead of raising a NodeNotSynced exception. Default False.

Raises:
NodeNotSynced: If the node is not fully synced. ValueError: If the contract creation block cannot be determined.

Example:
>>> block = await contract_creation_block_async("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> print(block)
1234567

Since contract_creation_block_async is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
when_no_history_return_0 (bool)

Return type:

ASyncFunctiony.contracts.has_method(address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress, method: str, return_response: bool = False) → bool | Any[source]

Checks to see if a contract has a method view method with no inputs. return_response=True will return response in bytes if response else False
Args:
address: The address of the contract. method: The name of the method to check for. return_response: If True, return the response of the method call instead of a boolean. Default False.

Example:
>>> await has_method("0x6B175474E89094C44Da98b954EedeAC495271d0F", "name()")
True

Since has_method is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress)
method (str)
return_response (bool)

Return type:

async y.contracts.has_methods(address, methods, _func=<built-in function all>)[source]

Checks to see if a contract has each view method (with no inputs) in methods. Pass at_least_one=True to only verify a contract has at least one of the methods.
Args:
address: The address of the contract. methods: A tuple of method names to check for. _func: The function to use for combining the results (either all() or any()).

Example:
>>> await has_methods("0x6B175474E89094C44Da98b954EedeAC495271d0F", ("name()", "symbol()"))
True

Since has_methods is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
methods (Iterable[str])
_func (Callable)

Return type:

async y.contracts.probe(address, methods, block=None, return_method=False)[source]

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
methods (Iterable[str])
block (int | BlockNumber | None)
return_method (bool)

Return type:

Any

async y.contracts.proxy_implementation(address, block)[source]

Get the implementation address for a proxy contract.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the proxy contract.
block (int | BlockNumber | None) – The block number at which to get the implementation address. Defaults to latest block.

Return type:

str | HexBytes | AnyAddress | EthAddress

Example

>>> await proxy_implementation("0x6B175474E89094C44Da98b954EedeAC495271d0F")
'0x1234567890abcdef1234567890abcdef12345678'

y.convert module

y.convert.checksum(address)[source]

Convert an address to its checksummed format.

This function uses the to_checksum_address function to convert an Ethereum address to its checksummed format. If the address is invalid, a ValueError is raised.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The Ethereum address to be checksummed.
Raises:: ValueError – If the provided address is not a valid Ethereum address.
Returns:: The checksummed Ethereum address.
Return type:: ChecksumAddress

Examples

>>> checksum("0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe")
'0xDe0B295669a9FD93d5F28D9Ec85E40f4cb697BAe'

See also

to_checksum_address() for the underlying implementation.

y.convert.to_address(address)[source]

Convert an address to a checksummed address, with caching.

This function normalizes the input address to a string, checks if it is already checksummed and cached, and if not, computes the checksummed address and caches it.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The Ethereum address to be converted.
Returns:: The checksummed Ethereum address.
Return type:: ChecksumAddress

Examples

>>> to_address("0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe")
'0xDe0B295669a9FD93d5F28D9Ec85E40f4cb697BAe'

See also

checksum() for the checksumming process.
__normalize_input_to_string() for input normalization.

async y.convert.to_address_async(address)[source]

Asynchronously convert an address to a checksummed address, with caching.

This function normalizes the input address to a string, checks if it is already checksummed and cached, and if not, computes the checksummed address asynchronously and caches it.

Parameters:: address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The Ethereum address to be converted.
Returns:: The checksummed Ethereum address.
Return type:: ChecksumAddress

Examples

>>> await to_address_async("0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe")
'0xDe0B295669a9FD93d5F28D9Ec85E40f4cb697BAe'

See also

checksum() for the checksumming process.
__normalize_input_to_string() for input normalization.

y.datatypes module

class y.datatypes.UsdPrice[source]

Bases: UsdValue

Represents a USD price.

Examples

>>> price = UsdPrice(1234.5678)
>>> str(price)
'$1234.56780000'

__add__(value, /): Return self+value.

__bool__(): True if self else False

__eq__(value, /): Return self==value.

__floordiv__(value, /): Return self//value.

__ge__(value, /): Return self>=value.

__getattribute__(name, /): Return getattr(self, name).

__gt__(value, /): Return self>value.

__le__(value, /): Return self<=value.

__lt__(value, /): Return self<value.

__mul__(value, /): Return self*value.

__pow__(value, mod=None, /): Return pow(self, value, mod).

__radd__(value, /): Return value+self.

__str__()

Return a string representation of the USD value. The value is formatted as a USD value with 8 decimal places.

Return type:: str

__sub__(value, /): Return self-value.

__truediv__(value, /): Return self/value.

fromhex()

Create a floating-point number from a hexadecimal string.

>>> float.fromhex('0x1.ffffp10')
2047.984375
>>> float.fromhex('-0x1p-1074')
-5e-324

hex()

Return a hexadecimal representation of a floating-point number.

>>> (-0.1).hex()
'-0x1.999999999999ap-4'
>>> 3.14159.hex()
'0x1.921f9f01b866ep+1'

class y.datatypes.UsdValue[source]

Bases: float

Represents a USD value with custom string representation.

Examples

>>> value = UsdValue(1234.5678)
>>> str(value)
'$1234.56780000'

__add__(value, /): Return self+value.

__bool__(): True if self else False

__eq__(value, /): Return self==value.

__floordiv__(value, /): Return self//value.

__ge__(value, /): Return self>=value.

__getattribute__(name, /): Return getattr(self, name).

__gt__(value, /): Return self>value.

__le__(value, /): Return self<=value.

__lt__(value, /): Return self<value.

__mul__(value, /): Return self*value.

__pow__(value, mod=None, /): Return pow(self, value, mod).

__radd__(value, /): Return value+self.

__str__()[source]

Return a string representation of the USD value. The value is formatted as a USD value with 8 decimal places.

Return type:: str

__sub__(value, /): Return self-value.

__truediv__(value, /): Return self/value.

fromhex()

Create a floating-point number from a hexadecimal string.

>>> float.fromhex('0x1.ffffp10')
2047.984375
>>> float.fromhex('-0x1p-1074')
-5e-324

hex()

Return a hexadecimal representation of a floating-point number.

>>> (-0.1).hex()
'-0x1.999999999999ap-4'
>>> 3.14159.hex()
'0x1.921f9f01b866ep+1'

y.datatypes.Address

A union of types used to represent Ethereum addresses.

Examples

>>> address_str = "0x1234567890abcdef1234567890abcdef12345678"
>>> address_hex = HexBytes("0x1234567890abcdef1234567890abcdef12345678")
>>> address_any = AnyAddress("0x1234567890abcdef1234567890abcdef12345678")
>>> address_eth = EthAddress("0x1234567890abcdef1234567890abcdef12345678")

alias of str | HexBytes | AnyAddress | EthAddress

y.datatypes.AddressOrContract

A type alias representing either an Ethereum address or a contract object. This can be an Address, a Contract, or its subclasses such as Contract and Contract.

Examples

>>> address = "0x1234567890abcdef1234567890abcdef12345678"
>>> contract = Contract.from_abi("MyContract", address, abi)

alias of str | HexBytes | AnyAddress | EthAddress | Contract

y.datatypes.AnyAddressType

A type alias representing any valid representation of an Ethereum address. This can be an Address, a Contract, or an integer.

Examples

>>> any_address_str = "0x1234567890abcdef1234567890abcdef12345678"
>>> any_address_contract = Contract.from_abi("MyContract", any_address_str, abi)
>>> any_address_int = 12345678

y.datatypes.Block

A union of types used to represent block numbers as integers.

Examples

>>> block_int = 12345678
>>> block_number = BlockNumber(12345678)

alias of int | BlockNumber

y.datatypes.Pool

A union of types representing liquidity pools.

Examples

>>> uniswap_pool = UniswapV2Pool("0xUniswapPoolAddress")
>>> curve_pool = CurvePool("0xCurvePoolAddress")

See also

alias of UniswapV2Pool | CurvePool

y.exceptions module

exception y.exceptions.CallReverted[source]

Bases: Exception

Raised when a contract call is reverted.

exception y.exceptions.CalldataPreparationError[source]

Bases: Exception

Raised when there’s an error in preparing calldata for a contract interaction.

exception y.exceptions.CantFetchParam[source]: Bases: Exception

exception y.exceptions.CantFindSwapPath[source]: Bases: Exception

exception y.exceptions.ContractNotVerified[source]

Bases: _ExplorerError

Raised when attempting to fetch the ABI for an unverified contract from a block explorer.

exception y.exceptions.InvalidAPIKeyError[source]

Bases: _ExplorerError

Raised when the API key for the block explorer has been rejected. This typically occurs when making requests to a block explorer API with a missing, incorrect, or banned key.

__init__(msg='')[source]

Parameters:: msg (str)

exception y.exceptions.MessedUpBrownieContract[source]

Bases: Exception

Raised when there’s an issue initialized a Brownie contract instance, typically in the compilation step.

__init__(address, *args)[source]

Parameters:: args (object)
Return type:: None

exception y.exceptions.NoProxyImplementation[source]

Bases: Exception

Raised when the implementation address of a proxy contract cannot be determined.

This may occur when trying to interact with proxy contracts that don’t follow standard patterns.

exception y.exceptions.NodeNotSynced[source]: Bases: Exception

exception y.exceptions.NonStandardERC20[source]

Bases: Exception

Raised when an ERC20 token contract is expected but the provided address is not a standard ERC20 token.

exception y.exceptions.NotABalancerV2Pool[source]

Bases: Exception

Raised when a contract is incorrectly identified as a Balancer V2 pool.

exception y.exceptions.NotAUniswapV2Pool[source]

Bases: Exception

Raised when a contract is incorrectly identified as a Uniswap V2 pool.

__init__(non_pool)[source]

Parameters:: non_pool (UniswapV2Pool)

exception y.exceptions.PriceError[source]

Bases: Exception

Raised when a queried price is not found.

__init__(logger, symbol)[source]

Parameters:

logger (Logger)
symbol (str)

exception y.exceptions.TokenError[source]

Bases: ValueError

Raised when a token contract is not the correct contract type for the desired operation.

__init__(token, desired_type, *optional_extra_args)[source]

Parameters:

token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
desired_type (str)
optional_extra_args (Any)

exception y.exceptions.TokenNotFound[source]

Bases: ValueError

Raised when a specified token cannot be found in a given container.

This is usually used when searching for a token in a liquidity pool.

__init__(token, container)[source]

exception y.exceptions.UnsupportedNetwork[source]

Bases: ValueError

Raised when an operation is attempted on an unsupported blockchain network.

exception y.exceptions.yPriceMagicError[source]

Bases: ValueError

Custom exception for ypricemagic-related errors.

This exception is raised when any error occurs inside of ypricemagic. For example, if a price fails to fetch or if there are unexpected Exceptions while calculating prices.

__init__(exc, token_address, block, symbol)[source]

Parameters:

exc (Exception)
token_address (str)
block (int | None)
symbol (str)

block: The block that was queried when the error occurred.

exception: The original Exception that was raised and wrapped with the yPriceMagicError.

token: The token that caused the error.

y.exceptions.call_reverted(e)[source]

Parameters:: e (Exception)
Return type:: bool

y.exceptions.continue_if_call_reverted(e)[source]

Parameters:: e (Exception)
Return type:: None

y.exceptions.contract_not_verified(e)[source]

Parameters:: e (Exception)
Return type:: bool

y.exceptions.out_of_gas(e)[source]

Parameters:: e (Exception)
Return type:: bool

y.exceptions.reraise_excs_with_extra_context(*extra_context, after=True)[source]

Parameters:

extra_context (Any)
after (bool)

y.networks module

class y.networks.Network[source]

Bases: IntEnum

A lightweight enum that enables lookup of chain IDs for popular blockchain networks.

Each network is associated with its unique integer Chain ID.

Examples

>>> Network.Mainnet
<Network.Mainnet: 1>
>>> Network.Mainnet.value
1

__add__(value, /): Return self+value.

__and__(value, /): Return self&value.

__bool__(): True if self else False

__eq__(value, /): Return self==value.

__floordiv__(value, /): Return self//value.

__ge__(value, /): Return self>=value.

__getattribute__(name, /): Return getattr(self, name).

__gt__(value, /): Return self>value.

__index__(): Return self converted to an integer, if self is suitable for use as an index into a list.

__invert__(): ~self

__le__(value, /): Return self<=value.

__lshift__(value, /): Return self<<value.

__lt__(value, /): Return self<value.

__mul__(value, /): Return self*value.

__or__(value, /): Return self|value.

__pow__(value, mod=None, /): Return pow(self, value, mod).

__radd__(value, /): Return value+self.

__rand__(value, /): Return value&self.

__rlshift__(value, /): Return value<<self.

__ror__(value, /): Return value|self.

__rrshift__(value, /): Return value>>self.

__rshift__(value, /): Return self>>value.

__rxor__(value, /): Return value^self.

__sizeof__(): Returns size in memory, in bytes.

__sub__(value, /): Return self-value.

__truediv__(value, /): Return self/value.

__xor__(value, /): Return self^value.

from_bytes(byteorder, *, signed=False)

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Indicates whether two’s complement is used to represent the integer.

static label(chain_id=None)[source]

Get the label (abbreviation) for a given chain ID.

If no chain ID is provided, it defaults to the current active chain.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.label(Network.Mainnet)
'ETH'
>>> Network.label()
'ETH'  # Assuming the current active chain is Mainnet

See also

Network.name() for getting the full name of the network.

static name(chain_id=None)[source]

Get the full name of a network for a given chain ID.

If no chain ID is provided, it defaults to the current active chain.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.name(Network.Mainnet)
'Mainnet'
>>> Network.name()
'Mainnet'  # Assuming the current active chain is Mainnet

See also

Network.label() for getting the abbreviation of the network.

static printable(chain_id=None)[source]

Get a printable string that identifies the network.

If the network is not supported, it returns a generic string with the chain ID.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.printable(Network.Mainnet)
'Mainnet'
>>> Network.printable(9999)
'chain 9999'

to_bytes(length, byteorder, *, signed=False)

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

Arbitrum = 42161: The Chain ID for Arbitrum One

Aurora = 1313161554: The Chain ID for Aurora

Avalanche = 43114: The Chain ID for Avalanche C-Chain

Base = 8453: The Chain ID for Base

BinanceSmartChain = 56: The Chain ID for Binance Smart Chain

Cronos = 25: The Chain ID for Cronos Mainnet

Fantom = 250: The Chain ID for Fantom Opera Network

Gnosis = 100: The Chain ID for xDai Chain (now Gnosis Chain)

Harmony = 1666600000: The Chain ID for Harmony Mainnet Shard 0

Heco = 128: The Chain ID for Heco

Mainnet = 1: The Chain ID for Ethereum Mainnet

Moonriver = 1285: The Chain ID for Moonriver Network

OKEx = 66: The Chain ID for OKEx Chain

Optimism = 10: The Chain ID for Optimism

Polygon = 137: The Chain ID for Polygon (formerly Matic) Network

denominator: the denominator of a rational number in lowest terms

numerator: the numerator of a rational number in lowest terms

xDai = 100: The Chain ID for xDai Chain (now Gnosis Chain)

y.time module

exception y.time.NoBlockFound[source]

Bases: Exception

Raised when no block is found for a specified timestamp because the timestamp is in the future.

Parameters:: timestamp – The timestamp for which no block was found.

__init__(timestamp)[source]

Parameters:: timestamp (UnixTimestamp | datetime)

y.time.check_node()[source]

Check if the Ethereum node is synced.

Raises:: y.exceptions.NodeNotSynced – If the node is not synced.
Return type:: None

Examples

>>> check_node()

y.time.check_node_async()[source]

Asynchronously check if the Ethereum node is synced.

Raises:: y.exceptions.NodeNotSynced – If the node is not synced.
Return type:: None

Examples

>>> await check_node_async()

y.time.closest_block_after_timestamp(timestamp, wait_for_block_if_needed=False)[source]

Get the closest block after a given timestamp.

Parameters:

timestamp (UnixTimestamp | datetime) – The timestamp to find the closest block after.
wait_for_block_if_needed (bool) – Whether to wait for a block if needed.

Returns:

The block number closest after the given timestamp.

Raises:

NoBlockFound – If no block is found after the timestamp.

Return type:

Example

>>> closest_block_after_timestamp(1672531199)
12345678

See also

get_block_at_timestamp()
get_block_timestamp()

ASyncFunctiony.time.closest_block_after_timestamp_async(timestamp: y.time.UnixTimestamp | datetime.datetime, wait_for_block_if_needed: bool = False) → int[source]

Asynchronously get the closest block after a given timestamp.
Args:
timestamp: The timestamp to find the closest block after. wait_for_block_if_needed: Whether to wait for a block if needed.

Returns:
The block number closest after the given timestamp.

Raises:
NoBlockFound: If no block is found after the timestamp.

Example:
>>> await closest_block_after_timestamp_async(1672531199)
12345678
See Also:

get_block_at_timestamp()

get_block_timestamp_async()

Since closest_block_after_timestamp_async is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

timestamp (UnixTimestamp | datetime)
wait_for_block_if_needed (bool)

Return type:

ASyncFunctiony.time.get_block_at_timestamp(timestamp: datetime.datetime) → int[source]

Get the block number just before a specific timestamp.

This function returns the block number at the given timestamp, which is the block number just before the specified timestamp.
Args:
timestamp: The timestamp to find the block for.

Returns:
The block number just before the given timestamp.

Example:
>>> await get_block_at_timestamp(datetime.datetime(2023, 1, 1))
12345678
See Also:

get_block_timestamp()

get_block_timestamp_async()

Since get_block_at_timestamp is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: timestamp (datetime)
Return type:: int

ASyncFunctiony.time.get_block_timestamp_async(height: int) → int[source]

Asynchronously get the timestamp of a block by its height.
Args:
height: The block height.

Returns:
The timestamp of the block.

Examples:
>>> await get_block_timestamp_async(12345678)
1609459200
See Also:

get_block_timestamp()

get_block_at_timestamp()

Since get_block_timestamp_async is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: height (int)
Return type:: int

Module contents

class y.Contract[source]

Bases: Contract

A Contract object with several modifications for enhanced functionality.

This class provides a modified contract object with additional features and optimizations:

Contracts will not be compiled. This allows you to more quickly fetch contracts from the block explorer and prevents you from having to download and install compilers.
NOTE: You must set autofetch_sources=false in your project’s brownie-config.yaml for this to work correctly.
To each contract method, a coroutine property has been defined which allows you to make asynchronous calls. This is enabled by inheriting from Contract, which provides the coroutine method for each ContractCall object. These asynchronous calls are intelligently batched in the background by dank_mids to reduce overhead.
New methods:
A few attributes were removed in order to minimize the size of a Contract object in memory:
- ast
- bytecode
- coverageMap
- deployedBytecode
- deployedSourceMap
- natspec
- opcodes
- pcMap

Examples

>>> contract = Contract("0xAddress")
>>> contract.methodName(*args, block_identifier=12345678)
1000000000000000000
>>> coro = contract.methodName.coroutine(*args, block_identifier=12345678)
>>> coro
<coroutine coroutine object at 0x12345678>
>>> contract.methodName(*args, block_identifier=12345678) == await coro
True

See also

__eq__(other)

Return self==value.

Parameters:: other (object)
Return type:: bool

__getattribute__(name)[source]

Get a contract method attribute.

This method implements lazy initialization of contract methods. If a method object does not yet exist, it is created and cached.

Parameters:: name (str) – The name of the attribute to get.
Returns:: The contract method object.
Return type:: DankContractCall | DankContractTx | DankOverloadedMethod

__init__(address, owner=None, require_success=True, cache_ttl=<EnvironmentVariable[name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, type=int, default_value=3600, current_value=3600, using_default=True]>)[source]

Initialize a Contract instance.

Note

autofetch-sources: false

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the contract.
owner (optional) – The owner of the contract. Default None.
require_success (optional) – If True, require successful contract verification. Default True.
cache_ttl (optional) – The time-to-live for the contract cache in seconds. Default set in ENVIRONMENT_VARIABLES.

Raises:

ContractNotFound – If the address is not a contract.
ContractNotVerified – If the contract is not verified and require_success is True.

Return type:

None

__post_init__(cache_ttl=None)[source]

Get rid of the contract call objects so we can materialize them on a JIT basis.

This method sets up lazy initialization for contract methods.

Parameters:: cache_ttl (int | None)
Return type:: None

__str__()

Return str(self).

Return type:: str

balance()

Returns the current ether balance of the contract, in wei.

Return type:: Wei

async build_name(return_None_on_failure=False)[source]

Get the build name of the contract.

Parameters:: return_None_on_failure (optional) – If True, return None if the build name cannot be determined instead of raising an Exception. Default False.
Return type:: str | None

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> await contract.build_name()
'Dai Stablecoin'

async classmethod coroutine(address, owner=None, persist=True, require_success=True, cache_ttl=<EnvironmentVariable[name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, type=int, default_value=3600, current_value=3600, using_default=True]>)[source]

Create a Contract instance asynchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address of the contract.
owner (optional) – The owner of the contract. Default None.
persist (optional) – If True, persist the contract in brownie’s local contract database. Default True.
require_success (optional) – If True, require successful contract verification. Default True.
cache_ttl (optional) – The time-to-live for the contract cache in seconds. Default set in ENVIRONMENT_VARIABLES.

Return type:

Self

Example

>>> contract = await Contract.coroutine("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> contract.symbol()
'DAI'

decode_input(calldata)

Decode input calldata for this contract.

Parameters:

calldata (str | bytes) – Calldata for a call to this contract

Returns:

str – Signature of the function that was called
Any – Decoded input arguments

Return type:

Tuple[str, Any]

classmethod from_abi(name, address, abi, owner=None, persist=True, cache_ttl=<EnvironmentVariable[name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, type=int, default_value=3600, current_value=3600, using_default=True]>)[source]

Create a Contract instance from an ABI.
Args:
name: The name of the contract. address: The address of the contract. abi: The ABI of the contract. owner (optional): The owner of the contract. Default None. persist (optional): If True, persist the contract in brownie’s local contract database. Default True. cache_ttl (optional): The time-to-live for the contract cache in seconds. Default set in ENVIRONMENT_VARIABLES.

Example:
>>> abi = [{"constant":True,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"payable":False,"stateMutability":"view","type":"function"}]
>>> contract = Contract.from_abi("MyContract", "0x6B175474E89094C44Da98b954EedeAC495271d0F", abi)
>>> contract.name()
'Dai Stablecoin'

Since from_abi is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

name (str)
address (str)
abi (List)
owner (AccountsType | None)
persist (bool)
cache_ttl (int | None)

Return type:

Self

classmethod from_ethpm(name, manifest_uri, address=None, owner=None, persist=True)[source]

Create a new Contract instance from an ethPM manifest.

This method allows for the creation of a Contract instance using an ethPM manifest, which is a standardized format for Ethereum smart contract packages.

Parameters:

name (str) – The name of the contract.
manifest_uri (str) – The URI of the ethPM manifest.
address (str | None) – The address of the deployed contract (optional).
owner (AccountsType | None) – The account that owns this contract instance.
persist (bool) – Whether to persist the contract data to brownie’s local db for future use.

Returns:

A new Contract instance.

Return type:

classmethod from_explorer(address, as_proxy_for=None, owner=None, silent=False, persist=True)[source]

Create a new Contract instance by fetching the ABI from a block explorer.

This method is useful for interacting with contracts that are not part of the current project, as it automatically fetches the contract’s ABI from a block explorer.

Parameters:

address (str) – The address of the deployed contract.
as_proxy_for (str | None) – The address of the implementation contract if this is a proxy contract.
owner (AccountsType | None) – The account that owns this contract instance.
silent (bool) – Whether to suppress console output during the process.
persist (bool) – Whether to persist the contract data to brownie’s db for future use.

Returns:

A new Contract instance.

Return type:

async get_code(block=None)[source]

Get the bytecode of the contract at a specific block.

Parameters:: block (optional) – The block number at which to get the bytecode. Defaults to latest block.
Return type:: HexBytes

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> await contract.get_code()
HexBytes('0x...')

get_method(calldata)

Parameters:: calldata (str)
Return type:: str | None

get_method_object(calldata)

Given a calldata hex string, returns a ContractMethod object.

Parameters:: calldata (str)
Return type:: _ContractMethod | None

classmethod get_solc_version(compiler_str, address)[source]

Return the solc compiler version either from the passed compiler string or try to find the latest available patch semver compiler version.

Parameters:

compiler_str (str) – The compiler string passed from the contract metadata.
address (str) – The contract address to check for.

Return type:

Version

has_method(method, return_response=False)[source]

Check if the contract has a specific method.

Parameters:

method (str) – The name of the method to check for.
return_response (optional) – If True, return the response of the method call instead of a boolean. Default False.

Return type:

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> contract.has_method("name()")
True

async has_methods(methods, _func=<built-in function all>)[source]

Check if the contract has all the specified methods.

Parameters:

methods (List[str]) – A list of method names to check for.
_func (optional) – The function to use for combining the results (either all() or any()). Default all().

Return type:

Example

>>> contract = Contract("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> await contract.has_methods(["name()", "symbol()"])
True

info()

Display NatSpec documentation for this contract.

Return type:: None

classmethod remove_deployment(address=None, alias=None)[source]

Removes this contract from the internal deployments db with the passed address or alias.

Parameters:

address (str | None) – An address to apply
alias (str | None) – An alias to apply

Return type:

Tuple[Dict | None, Dict | None]

set_alias(alias, persist=True)[source]

Apply a unique alias this object. The alias can be used to restore the object in future sessions.

Parameters:

alias (str | None) – An alias to apply. If None, any existing alias is removed.
persist (bool)

Return type:

None

property abi: List

property alias: str | None

events: ContractEvents

A container for various event types associated with this contract.

Provides a convenient way to query contract events with minimal code.

signatures: Dict[Method, Signature]: A dictionary mapping method names to their corresponding signatures.

topics: Dict[str, str]: A dictionary mapping event names to their corresponding topics.

verified = True: True if the contract is verified on this network’s block explorer. False otherwise.

class y.ERC20[source]

Bases: ContractBase

Represents an ERC20 token.

ASyncFunction_decimals(block: int | eth_typing.evm.BlockNumber | NoneType = None) → int[source]

used to fetch decimals at specific block

Since _decimals is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: block (int | BlockNumber | None)
Return type:: int

ASyncFunction_scale(block: int | eth_typing.evm.BlockNumber | NoneType = None) → int[source]

Since _scale is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: block (int | BlockNumber | None)
Return type:: int

__eq__(_ContractBase__o)

Return self==value.

Parameters:: _ContractBase__o (object)
Return type:: bool

__init__(self)

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
asynchronous (bool)
_deploy_block (int | None)

Return type:

None

__str__()

Return the contract address as a string.

Returns:: The contract address as a string.
Return type:: str

__build_name__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__name__: HiddenMethodDescriptor[Self, str] = 'ERC20'

__scale__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

The original docstring for get() is shown below:

_ASyncPropertyDescriptorBase.get(self, instance: I, owner: Optional[Type[I]] = None) -> T Asynchronously retrieves the property value.

Args:
instance: The instance from which the property is accessed. owner: The owner class of the property.

Returns:
The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Parameters:

instance (I)
owner (Type[I] | None)

Return type:

T

address: str | HexBytes | AnyAddress | EthAddress: The contract address of the token.

asynchronous: bool = False

balance_of[source]

Query the balance of the token for a given address at a specific block.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int) – The address to query.
block (optional) – The block number to query. Defaults to latest block.

Returns:

The balance of the token held by address at block block.

Return type:

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.balance_of("0xabcdefabcdefabcdefabcdefabcdefabcdef")
500000000000000000000

balance_of_readable[source]

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)

Return type:

float

build_name

Get the contract’s build name.

Returns:: The contract’s build name.

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.build_name
'MyContract'

property contract: Contract

decimals[source]

The number of decimal places for the token.

Returns:: The number of decimal places for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.decimals
18

deploy_block: ASyncBoundMethod[Self, Any, int]

Get the block number when the contract was deployed.

Parameters:: when_no_history_return_0 (bool) – If True, return 0 when no history is found instead of raising an exception.
Returns:: The block number when the contract was deployed, or 0 if when_no_history_return_0 is True and the deploy block cannot be determined.
Return type:: int

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.deploy_block()
1234567

See also

contract_creation_block_async()

has_method

Check if the contract has a specific method.

Parameters:

method (str) – The name of the method to check for.
return_response (bool) – If True, return the response of the method call instead of a boolean.

Returns:

A boolean indicating whether the contract has the method, or the response of the method call if return_response is True.

Return type:

ASyncIterator[_EventItem]

Examples

>>> contract = ContractBase("0x1234567890abcdef1234567890abcdef12345678")
>>> await contract.has_method("name()")
True

See also

has_method()

name[source]

The token’s name.

Returns:: The token’s name.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.name
'TokenName'

price[source]

Get the price of the token in USD.

Parameters:

block (optional) – The block number to query. Defaults to latest block.
return_None_on_failure (bool) – If True, return None instead of raising a yPriceMagicError on failure.
skip_cache (bool) – If True, skip using the cache while fetching price data.
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...]) – An optional tuple of pools to ignore when calculating the price.

Returns:

The price of the token in USD, or None if return_None_on_failure is True and the price cannot be retrieved.

Raises:

yPriceMagicError – If return_None_on_failure is False and the price cannot be retrieved.

Return type:

UsdPrice | None

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.price()
1.23

See also

y.prices.magic.get_price()

scale[source]

Get the scaling factor for the token.

Returns:: The scaling factor for the token.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.scale
1000000000000000000

symbol[source]

The token’s symbol.

Returns:: The token’s symbol.

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.symbol
'TKN'

total_supply[source]

Get the total supply of the token.

Parameters:: block (optional) – The block number to query. Defaults to latest block.
Returns:: The total supply of the token.
Return type:: int

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.total_supply()
1000000000000000000000

total_supply_readable[source]

Get the total supply of the token scaled to a human-readable decimal.

Parameters:: block (optional) – The block number to query.
Returns:: The total supply of the token scaled to a decimal.
Return type:: float

Examples

>>> token = ERC20("0x1234567890abcdef1234567890abcdef12345678")
>>> await token.total_supply_readable()
1000.0

class y.Events[source]

Bases: LogFilter

A class for fetching and processing events.

This class extends LogFilter to provide additional functionality for handling events.

When awaited, a list of all Log will be returned.

Example

>>> my_object = Events(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], Log)
True

obj_type: alias of _EventItem

__aiter__()

Returns an async iterator over the stored objects, yielding new ones as they are fetched.

Return type:: AsyncIterator[T]

__await__()

Asynchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.

__init__(*, addresses=[], topics=[], from_block=None, chunk_size=10000, chunks_per_batch=None, semaphore=None, executor=None, is_reusable=True, verbose=False)

Initialize a LogFilter instance.

Parameters:

addresses – List of contract addresses to fetch logs from.
topics – List of event topics to filter logs by.
from_block (int | BlockNumber | None) – The starting block to fetch logs from.
chunk_size (int) – The number of blocks to fetch in each chunk.
chunks_per_batch (int | None) – The number of chunks to fetch in each batch.
semaphore (BlockSemaphore | None) – A semaphore for limiting concurrent requests.
executor (_AsyncExecutorMixin | None) – An executor for running tasks asynchronously.
is_reusable (bool) – Whether the filter is reusable.
verbose (bool) – Verbosity level for logging.

Return type:

None

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the _DiskCachedMixin.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:: Iterator[T]

_objects_thru(block, from_block=None)

Generates objects up to a specified block, or indefinitely if none is given.

Parameters:

block (Block | None) – Maximum block number to generate objects. If None, yields continuously.
from_block (Block | None) – Minimum block from which to start yielding objects. Only valid if reusable.

Yields:

Objects that fall within the requested block range, if any.

Return type:

AsyncIterator[T]

events(to_block, from_block=None)[source]

Get events up to a given block.

Parameters:

to_block (int | BlockNumber) – The ending block to fetch events to.
from_block (int | BlockNumber | None)

Yields:

A decoded event.

Return type:

Examples

>>> events = Events(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in events.events(1000100):
...     print(event)

filter(self, function: ViewFn[T])

Filters the T objects yielded by the _ASyncView based on a function.

Parameters:: function (ViewFn[T]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered T objects from the _ASyncView.

logs(to_block)

Get logs up to a given block.

Parameters:: to_block (int | BlockNumber | None) – The ending block to fetch logs to.
Yields:: A raw log.
Return type:: ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the T objects yielded by the _ASyncView.

Parameters:

key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this _ASyncView, but sorted.

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:: wrapped (AsyncIterable[T])
Return type:: ASyncIterable[T]

__pyx_vtable__ = <capsule object NULL>

__wrapped__

addresses

property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:: A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)

property cache: LogCache: Returns the associated cache object, which must be defined in subclasses.

property executor: AsyncThreadPoolExecutor: Returns the executor used for disk operations, creating one if necessary.

from_block

property insert_to_db: Callable[[Log], None]

Get the function for inserting logs into the database.

Raises:: NotImplementedError – If this method is not implemented in the subclass.

is_reusable

materialized

List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.
Type:: _AwaitableAsyncIterableMixin.materialized

property semaphore: BlockSemaphore

to_block

topics

class y.LogFilter[source]

Bases: Filter[Log, LogCache]

A filter for fetching and processing event logs.

This class provides methods to fetch logs from the blockchain and process them.

When awaited, a list of all Log will be returned.

Example

>>> my_object = LogFilter(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], Log)
True

__aiter__()

Returns an async iterator over the stored objects, yielding new ones as they are fetched.

Return type:: AsyncIterator[T]

__await__()

Asynchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.

__init__(*, addresses=[], topics=[], from_block=None, chunk_size=10000, chunks_per_batch=None, semaphore=None, executor=None, is_reusable=True, verbose=False)[source]

Initialize a LogFilter instance.

Parameters:

addresses – List of contract addresses to fetch logs from.
topics – List of event topics to filter logs by.
from_block (int | BlockNumber | None) – The starting block to fetch logs from.
chunk_size (int) – The number of blocks to fetch in each chunk.
chunks_per_batch (int | None) – The number of chunks to fetch in each batch.
semaphore (BlockSemaphore | None) – A semaphore for limiting concurrent requests.
executor (_AsyncExecutorMixin | None) – An executor for running tasks asynchronously.
is_reusable (bool) – Whether the filter is reusable.
verbose (bool) – Verbosity level for logging.

Return type:

None

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the _DiskCachedMixin.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:: Iterator[T]

_objects_thru(block, from_block=None)

Generates objects up to a specified block, or indefinitely if none is given.

Parameters:

block (Block | None) – Maximum block number to generate objects. If None, yields continuously.
from_block (Block | None) – Minimum block from which to start yielding objects. Only valid if reusable.

Yields:

Objects that fall within the requested block range, if any.

Return type:

AsyncIterator[T]

filter(self, function: ViewFn[T])

Filters the T objects yielded by the _ASyncView based on a function.

Parameters:: function (ViewFn[T]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered T objects from the _ASyncView.

logs(to_block)[source]

Get logs up to a given block.

Parameters:: to_block (int | BlockNumber | None) – The ending block to fetch logs to.
Yields:: A raw log.
Return type:: ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the T objects yielded by the _ASyncView.

Parameters:

key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this _ASyncView, but sorted.

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:: wrapped (AsyncIterable[T])
Return type:: ASyncIterable[T]

__pyx_vtable__ = <capsule object NULL>

__wrapped__

addresses

property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:: A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)

property cache: LogCache: Returns the associated cache object, which must be defined in subclasses.

property executor: AsyncThreadPoolExecutor: Returns the executor used for disk operations, creating one if necessary.

from_block

property insert_to_db: Callable[[Log], None]

Get the function for inserting logs into the database.

Raises:: NotImplementedError – If this method is not implemented in the subclass.

is_reusable

materialized

List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.
Type:: _AwaitableAsyncIterableMixin.materialized

property semaphore: BlockSemaphore

to_block

topics

class y.Network[source]

Bases: IntEnum

A lightweight enum that enables lookup of chain IDs for popular blockchain networks.

Each network is associated with its unique integer Chain ID.

Examples

>>> Network.Mainnet
<Network.Mainnet: 1>
>>> Network.Mainnet.value
1

__add__(value, /): Return self+value.

__and__(value, /): Return self&value.

__bool__(): True if self else False

__eq__(value, /): Return self==value.

__floordiv__(value, /): Return self//value.

__ge__(value, /): Return self>=value.

__getattribute__(name, /): Return getattr(self, name).

__gt__(value, /): Return self>value.

__index__(): Return self converted to an integer, if self is suitable for use as an index into a list.

__invert__(): ~self

__le__(value, /): Return self<=value.

__lshift__(value, /): Return self<<value.

__lt__(value, /): Return self<value.

__mul__(value, /): Return self*value.

__or__(value, /): Return self|value.

__pow__(value, mod=None, /): Return pow(self, value, mod).

__radd__(value, /): Return value+self.

__rand__(value, /): Return value&self.

__rlshift__(value, /): Return value<<self.

__ror__(value, /): Return value|self.

__rrshift__(value, /): Return value>>self.

__rshift__(value, /): Return self>>value.

__rxor__(value, /): Return value^self.

__sizeof__(): Returns size in memory, in bytes.

__sub__(value, /): Return self-value.

__truediv__(value, /): Return self/value.

__xor__(value, /): Return self^value.

from_bytes(byteorder, *, signed=False)

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Indicates whether two’s complement is used to represent the integer.

static label(chain_id=None)[source]

Get the label (abbreviation) for a given chain ID.

If no chain ID is provided, it defaults to the current active chain.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.label(Network.Mainnet)
'ETH'
>>> Network.label()
'ETH'  # Assuming the current active chain is Mainnet

See also

Network.name() for getting the full name of the network.

static name(chain_id=None)[source]

Get the full name of a network for a given chain ID.

If no chain ID is provided, it defaults to the current active chain.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.name(Network.Mainnet)
'Mainnet'
>>> Network.name()
'Mainnet'  # Assuming the current active chain is Mainnet

See also

Network.label() for getting the abbreviation of the network.

static printable(chain_id=None)[source]

Get a printable string that identifies the network.

If the network is not supported, it returns a generic string with the chain ID.

Parameters:: chain_id (int | None) – The chain ID of the network.
Return type:: str

Examples

>>> Network.printable(Network.Mainnet)
'Mainnet'
>>> Network.printable(9999)
'chain 9999'

to_bytes(length, byteorder, *, signed=False)

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

Arbitrum = 42161: The Chain ID for Arbitrum One

Aurora = 1313161554: The Chain ID for Aurora

Avalanche = 43114: The Chain ID for Avalanche C-Chain

Base = 8453: The Chain ID for Base

BinanceSmartChain = 56: The Chain ID for Binance Smart Chain

Cronos = 25: The Chain ID for Cronos Mainnet

Fantom = 250: The Chain ID for Fantom Opera Network

Gnosis = 100: The Chain ID for xDai Chain (now Gnosis Chain)

Harmony = 1666600000: The Chain ID for Harmony Mainnet Shard 0

Heco = 128: The Chain ID for Heco

Mainnet = 1: The Chain ID for Ethereum Mainnet

Moonriver = 1285: The Chain ID for Moonriver Network

OKEx = 66: The Chain ID for OKEx Chain

Optimism = 10: The Chain ID for Optimism

Polygon = 137: The Chain ID for Polygon (formerly Matic) Network

denominator: the denominator of a rational number in lowest terms

numerator: the numerator of a rational number in lowest terms

xDai = 100: The Chain ID for xDai Chain (now Gnosis Chain)

class y.ProcessedEvents[source]

Bases: Events, ASyncIterable[T]

A class for fetching, processing, and iterating over events.

This class extends Events to provide additional functionality for processing events.

When awaited, a list of all T objects will be returned.

Example

>>> my_object = ProcessedEvents(...)
>>> all_contents = await my_object
>>> isinstance(all_contents, list)
True
>>> isinstance(all_contents[0], T)
True

obj_type: alias of _EventItem

__aiter__()

Returns an async iterator over the stored objects, yielding new ones as they are fetched.

Return type:: AsyncIterator[T]

__await__()

Asynchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.

__init__(*, addresses=[], topics=[], from_block=None, chunk_size=10000, chunks_per_batch=None, semaphore=None, executor=None, is_reusable=True, verbose=False)

Initialize a LogFilter instance.

Parameters:

addresses – List of contract addresses to fetch logs from.
topics – List of event topics to filter logs by.
from_block (int | BlockNumber | None) – The starting block to fetch logs from.
chunk_size (int) – The number of blocks to fetch in each chunk.
chunks_per_batch (int | None) – The number of chunks to fetch in each batch.
semaphore (BlockSemaphore | None) – A semaphore for limiting concurrent requests.
executor (_AsyncExecutorMixin | None) – An executor for running tasks asynchronously.
is_reusable (bool) – Whether the filter is reusable.
verbose (bool) – Verbosity level for logging.

Return type:

None

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

__iter__(self) → Iterator[T]

Return an iterator that yields T objects from the _DiskCachedMixin.

Note

Synchronous iteration leverages ASyncIterator, which uses asyncio.BaseEventLoop.run_until_complete() to fetch items. ASyncIterator.__next__() raises a SyncModeInAsyncContextError if the event loop is already running.

If you encounter a SyncModeInAsyncContextError, you are likely working in an async codebase and should consider asynchronous iteration using __aiter__() and __anext__() instead.

Return type:: Iterator[T]

_objects_thru(block, from_block=None)

Generates objects up to a specified block, or indefinitely if none is given.

Parameters:

block (Block | None) – Maximum block number to generate objects. If None, yields continuously.
from_block (Block | None) – Minimum block from which to start yielding objects. Only valid if reusable.

Yields:

Objects that fall within the requested block range, if any.

Return type:

AsyncIterator[T]

events(to_block, from_block=None)

Get events up to a given block.

Parameters:

to_block (int | BlockNumber) – The ending block to fetch events to.
from_block (int | BlockNumber | None)

Yields:

A decoded event.

Return type:

ASyncIterator[_EventItem]

Examples

>>> events = Events(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in events.events(1000100):
...     print(event)

filter(self, function: ViewFn[T])

Filters the T objects yielded by the _ASyncView based on a function.

Parameters:: function (ViewFn[T]) – A function that returns a boolean that indicates if an item should be included in the filtered result. Can be sync or async.
Returns:: An instance of ASyncFilter that yields the filtered T objects from the _ASyncView.

logs(to_block)

Get logs up to a given block.

Parameters:: to_block (int | BlockNumber | None) – The ending block to fetch logs to.
Yields:: A raw log.
Return type:: ASyncIterator[Log]

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> logs = log_filter.logs(1000100)
>>> print(logs)

objects(to_block, from_block=None)[source]

Get an ASyncIterator that yields all events up to a given block.

Parameters:

to_block (int | BlockNumber) – The ending block to fetch events to.
from_block (int | BlockNumber | None)

Returns:

An ASyncIterator that yields all included events.

Return type:

ASyncIterator[_EventItem]

Examples

>>> processed_events = ProcessedEvents(addresses=["0x1234..."], topics=["0x5678..."])
>>> async for event in processed_events.objects(1000100):
...     print(event)

sort(self, key: SortKey[T] = None, reverse: bool = False)

Sort the T objects yielded by the _ASyncView.

Parameters:

key (optional) – A function of one argument that is used to extract a comparison key from each list element. If None, the elements themselves will be sorted. Defaults to None.
reverse (optional) – If True, the yielded elements will be sorted in reverse order. Defaults to False.

Returns:

An instance of ASyncSorter that will yield the T objects yielded from this _ASyncView, but sorted.

classmethod wrap(cls, wrapped: AsyncIterable[T]) → 'ASyncIterable[T]'

Class method to wrap an AsyncIterable for backward compatibility.

Parameters:: wrapped (AsyncIterable[T])
Return type:: ASyncIterable[T]

__pyx_vtable__ = <capsule object NULL>

__wrapped__

addresses

property bulk_insert: Callable[[List[Log]], Awaitable[None]]

Get the function for bulk inserting logs into the database.

Returns:: A function for bulk inserting logs.

Examples

>>> log_filter = LogFilter(addresses=["0x1234..."], topics=["0x5678..."])
>>> await log_filter.bulk_insert(logs)

property cache: LogCache: Returns the associated cache object, which must be defined in subclasses.

property executor: AsyncThreadPoolExecutor: Returns the executor used for disk operations, creating one if necessary.

from_block

property insert_to_db: Callable[[Log], None]

Get the function for inserting logs into the database.

Raises:: NotImplementedError – If this method is not implemented in the subclass.

is_reusable

materialized

List[T]

Synchronously iterate through the {cls} and return all {obj}.

Returns:: A list of the {obj} yielded by the {cls}.
Type:: _AwaitableAsyncIterableMixin.materialized

property semaphore: BlockSemaphore

to_block

topics

Since balanceOf is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

call_address (str | HexBytes | AnyAddress | EthAddress | Contract)
input_address (str | HexBytes | AnyAddress | EthAddress | Contract)
block (int | BlockNumber | None)
return_None_on_failure (bool)

Return type:

int | None

Determine and return the category or “bucket” of a given token.

This function classifies a token by performing a set of tests in the following order:

It first attempts to retrieve a cached bucket from the database via y._db.utils.token.get_bucket().

Next, it applies simple string-based comparisons defined in the string_matchers dictionary.

If no bucket is determined, it concurrently executes a set of asynchronous “calls-only” tests for further classification. The function will return immediately once any of these tests confirms the token’s membership in a bucket.

If none of the concurrent tests succeed, it falls back to a series of sequential asynchronous checks that involve both contract initializations and blockchain calls.

Note

The fallback sequential tests are executed in order. However, the first two fallback checks (for “solidex” and “uni or uni-like lp”) are evaluated independently; as a result, if both conditions return true, the latter bucket (“uni or uni-like lp”) will overwrite the earlier assignment from “solidex”.

The final determined bucket is stored in the database using db.set_bucket().
Args:
token: The token address to classify.

Examples:
Basic usage with a stable USD token:
>>> bucket = check_bucket("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> print(bucket)
stable usd
Example triggering a concurrent asynchronous test:
>>> bucket = check_bucket("0xSomeLPTokenAddress")
>>> print(bucket)
popsicle
Example where multiple fallback conditions are met (note that if both “solidex” and “uni or uni-like lp” tests pass, the latter overwrites the former):
>>> bucket = check_bucket("0xAnotherTokenAddress")
>>> print(bucket)
uni or uni-like lp
See Also:
y.convert.to_address_async() y.utils.logging.get_price_logger() _check_bucket_helper() (Helper used for asynchronous bucket checks)

Since check_bucket is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:: token (str | HexBytes | AnyAddress | EthAddress | Contract | int)
Return type:: str

Determine the block when a contract was created using binary search. NOTE Requires access to historical state. Doesn’t account for CREATE2 or SELFDESTRUCT.
Args:
address: The address of the contract. when_no_history_return_0: If True, return 0 when no history is found instead of raising a NodeNotSynced exception. Default False.

Raises:
NodeNotSynced: If the node is not fully synced. ValueError: If the contract creation block cannot be determined.

Example:
>>> block = await contract_creation_block_async("0x6B175474E89094C44Da98b954EedeAC495271d0F")
>>> print(block)
1234567

Since contract_creation_block_async is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
when_no_history_return_0 (bool)

Return type:

ASyncFunctiony.fetch_multicall(*calls: Any, block: int | eth_typing.evm.BlockNumber | NoneType = None) → List[Any | None][source]

Since fetch_multicall is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

calls (Any)
block (int | BlockNumber | None)

Return type:

List[Any | None]

ASyncFunctiony.get_block_at_timestamp(timestamp: datetime.datetime) → int[source]

Get the block number just before a specific timestamp.

This function returns the block number at the given timestamp, which is the block number just before the specified timestamp.
Args:
timestamp: The timestamp to find the block for.

Returns:
The block number just before the given timestamp.

Example:
>>> await get_block_at_timestamp(datetime.datetime(2023, 1, 1))
12345678
See Also:

get_block_timestamp()

get_block_timestamp_async()

Since get_block_at_timestamp is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: timestamp (datetime)
Return type:: int

ASyncFunctiony.get_block_timestamp_async(height: int) → int[source]

Asynchronously get the timestamp of a block by its height.
Args:
height: The block height.

Returns:
The timestamp of the block.

Examples:
>>> await get_block_timestamp_async(12345678)
1609459200
See Also:

get_block_timestamp()

get_block_at_timestamp()

Since get_block_timestamp_async is an ASyncFunctionAsyncDefault, you can optionally pass sync=True or asynchronous=False to force it to run synchronously and return a value. Without either kwarg, it will return a coroutine for you to await.

Parameters:: height (int)
Return type:: int

ASyncFunctiony.get_price(token_address: Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, *, fail_to_None: bool = False, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>, ignore_pools: Tuple[Union[ForwardRef('UniswapV2Pool'), ForwardRef('CurvePool')], ...] = (), silent: bool = False) → y.datatypes.UsdPrice | None[source]

Get the price of a token in USD.
Args:
token_address: The address of the token to price. The function accepts hexadecimal strings, Brownie Contract objects, and integers as shorthand for addresses. block (optional): The block number at which to get the price. If None, uses the latest block. fail_to_None (optional): If True, return None instead of raising a yPriceMagicError on failure. Defaults to False. skip_cache (optional): If True, bypass the cache and fetch the price directly. Defaults to ENVS.SKIP_CACHE. ignore_pools (optional): A tuple of pool addresses to ignore when fetching the price. silent: If True, suppress error logging. Defaults to False.

Returns:
The price of the token in USD, or None if the price couldn’t be determined and fail_to_None is True.

Raises:
yPriceMagicError: If the price couldn’t be determined and fail_to_None is False.

Note:
ypricemagic accepts integers as valid token_address values for convenience. For example, you can use y.get_price(0x0bc529c00C6401aEF6D220BE8C6Ea1667F6Ad93e) to save keystrokes while testing in an interactive console.

Examples:
>>> from y import get_price
>>> price = get_price("0x0bc529c00C6401aEF6D220BE8C6Ea1667F6Ad93e", 12345678)
>>> print(price)
See Also:
get_prices()

Since get_price is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token_address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
block (int | BlockNumber | None)
fail_to_None (bool)
skip_cache (bool)
ignore_pools (Tuple[UniswapV2Pool | CurvePool, ...])
silent (bool)

Return type:

UsdPrice | None

ASyncFunctiony.get_prices(token_addresses: Iterable[Union[str, hexbytes.main.HexBytes, ~AnyAddress, brownie.convert.datatypes.EthAddress, brownie.network.contract.Contract, int]], block: Union[int, eth_typing.evm.BlockNumber, NoneType] = None, *, fail_to_None: bool = False, skip_cache: bool = <EnvironmentVariable[name=`YPRICEMAGIC_SKIP_CACHE`, type=bool, default_value=False, current_value=False, using_default=True]>, silent: bool = False) → List[y.datatypes.UsdPrice | None][source]

Get prices for multiple tokens in USD.

This function is optimized for parallel execution and should be preferred over get_price() when querying prices in bulk.
Args:
token_addresses: An iterable of token addresses to price. block (optional): The block number at which to get the prices. If None, defaults to the latest block. fail_to_None (optional): If True, return None for tokens whose price couldn’t be determined. Defaults to False. skip_cache (optional): If True, bypass the cache and fetch prices directly. Defaults to ENVS.SKIP_CACHE. silent (optional): If True, suppress error logging and any progress indicators. Defaults to False.

Returns:
A list of token prices in USD, in the same order as the input token_addresses.

Examples:
>>> from y import get_prices
>>> prices = get_prices(["0x123...", "0x456..."], block=12345678)
>>> print(prices)
See Also:
get_price() and map_prices()

Since get_prices is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

token_addresses (Iterable[str | HexBytes | AnyAddress | EthAddress | Contract | int])
block (int | BlockNumber | None)
fail_to_None (bool)
skip_cache (bool)
silent (bool)

Return type:

List[UsdPrice | None]

ASyncFunctiony.has_method(address: str | hexbytes.main.HexBytes | AnyAddress | brownie.convert.datatypes.EthAddress, method: str, return_response: bool = False) → bool | Any[source]

Checks to see if a contract has a method view method with no inputs. return_response=True will return response in bytes if response else False
Args:
address: The address of the contract. method: The name of the method to check for. return_response: If True, return the response of the method call instead of a boolean. Default False.

Example:
>>> await has_method("0x6B175474E89094C44Da98b954EedeAC495271d0F", "name()")
True

Since has_method is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress)
method (str)
return_response (bool)

Return type:

async y.has_methods(address, methods, _func=<built-in function all>)[source]

Checks to see if a contract has each view method (with no inputs) in methods. Pass at_least_one=True to only verify a contract has at least one of the methods.
Args:
address: The address of the contract. methods: A tuple of method names to check for. _func: The function to use for combining the results (either all() or any()).

Example:
>>> await has_methods("0x6B175474E89094C44Da98b954EedeAC495271d0F", ("name()", "symbol()"))
True

Since has_methods is an ASyncFunctionSyncDefault, you can optionally pass sync=False or asynchronous=True to force it to return a coroutine. Without either kwarg, it will run synchronously.

Parameters:

address (str | HexBytes | AnyAddress | EthAddress | Contract | int)
methods (Iterable[str])
_func (Callable)

Return type: