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[ABIFunction | ABIConstructor | ABIFallback | ABIReceive | ABIEvent | ABIError]

address: Final

property alias: str | None

bytecode: Final[HexStr]

events: ContractEvents

A container for various event types associated with this contract.

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

selectors: Final[Dict[Selector, FunctionName]]

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.

tx: Final

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[T]

__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 occurred, 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 | Address | 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 | Address | 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 | Address | 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 | Address | EthAddress | Contract | int)
when_no_history_return_0 (bool)

Return type:

int

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 | Address | 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 | Address | 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 | Address | 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 | Address | 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 | Address | EthAddress

Example

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

y.convert module

y.convert.checksum(address)

y.convert.to_address(address)

y.convert.to_address_async(address)

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 | Address | 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)

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

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.CalldataPreparationError

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.CantFetchParam

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.CantFindSwapPath

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.ContractNotVerified

Bases: _ExplorerError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.InvalidAPIKeyError

Bases: _ExplorerError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.MessedUpBrownieContract

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.NoProxyImplementation

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.NodeNotSynced

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.NonStandardERC20

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.NotABalancerV2Pool

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.NotAUniswapV2Pool

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.PriceError

Bases: Exception

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.TokenError

Bases: ValueError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.TokenNotFound

Bases: ValueError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.UnsupportedNetwork

Bases: ValueError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

exception y.exceptions.yPriceMagicError

Bases: ValueError

__getstate__()

__mypyc_attrs__ = ('__dict__',)

y.exceptions.call_reverted(e)

y.exceptions.continue_if_call_reverted(e)

y.exceptions.contract_not_verified(e)

y.exceptions.out_of_gas(e)

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

y.monkey_patches module

y.monkey_patches.monkey_patch_checksum_cache()[source]

Monkey patch dependency checksums with faster versions.

This function replaces the default checksumming functions in various libraries with a faster implementation using to_address. This improves performance by utilizing caching and optimized checksumming.

See also

to_address() for the checksumming process.

y.networks module

class y.networks.Network

Bases: IntEnum

mypyc filler docstring

__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.

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: int = 42161

Aurora: int = 1313161554

Avalanche: int = 43114

Base: int = 8453

Berachain: int = 80094

BinanceSmartChain: int = 56

Cronos: int = 25

Fantom: int = 250

Gnosis: int = 100

Harmony: int = 1666600000

Heco: int = 128

Katana: int = 747474

Mainnet: int = 1

Moonriver: int = 1285

OKEx: int = 66

Optimism: int = 10

Polygon: int = 137

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

label

name

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

printable

xDai: int = 100

y.time module

final 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:

BlockNumber

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) → eth_typing.evm.BlockNumber[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:

BlockNumber

ASyncFunctiony.time.get_block_at_timestamp(timestamp: datetime.datetime) → eth_typing.evm.BlockNumber[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:: BlockNumber

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[int](name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, default_value=3600, using_default=True))[source]

Initialize a Contract instance.

Note

autofetch-sources: false

Parameters:

address (str | HexBytes | AnyAddress | Address | 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[int](name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, default_value=3600, using_default=True))[source]

Create a Contract instance asynchronously.

Parameters:

address (str | HexBytes | AnyAddress | Address | 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[int](name=`YPRICEMAGIC_CONTRACT_CACHE_TTL`, default_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[ABIFunction | ABIConstructor | ABIFallback | ABIReceive | ABIEvent | ABIError]

address: Final

property alias: str | None

bytecode: Final[HexStr]

events: ContractEvents

A container for various event types associated with this contract.

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

selectors: Final[Dict[Selector, FunctionName]]

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.

tx: Final

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.

__eq__(_ContractBase__o)

Return self==value.

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

__init__(self)

Parameters:

address (str | HexBytes | AnyAddress | Address | 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.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

The original docstring for get() is shown below:

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

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Return type:

T

__decimals__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

The original docstring for get() is shown below:

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

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Return type:

T

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

__scale__: HiddenMethodDescriptor[Self, int]

Optional[Type[I]] = None) -> T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

The original docstring for get() is shown below:

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

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Return type:

T

__symbol__: HiddenMethodDescriptor[Self, str]

Optional[Type[I]] = None) -> T

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Return type:

T

The original docstring for get() is shown below:

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

Asynchronously retrieves the property value.

Parameters:

instance (I) – The instance from which the property is accessed.
owner (Type[I] | None) – The owner class of the property.

Returns:

The property value.

Type:

_ASyncPropertyDescriptorBase.get(self, instance

Type:

I, owner

Return type:

T

address: ChecksumAddress: The contract address of the token.

asynchronous: bool = False: A boolean indicating whether the instance will be used asynchronously.

balance_of[source]

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

Parameters:

address (str | HexBytes | AnyAddress | Address | 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:

int

Examples

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

balance_of_readable[source]

Parameters:

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

Return type:

Decimal

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:

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 T objects will be returned.

Example

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

__origin__: alias of ASyncIterable

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__(self) → Generator[Any, Any, List[T]]

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) → ASyncIterator[T]: Return an iterator that yields {obj} from the {cls}.

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.

_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:

ASyncIterator[_EventItem][_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 ASyncIterator[T] 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 ASyncIterator[T].

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][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 ASyncIterator[T].

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 ASyncIterator[T], 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]

__args__ = (~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

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 T objects will be returned.

Example

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

__origin__: alias of ASyncIterable

__aiter__()

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

Return type:: AsyncIterator[T]

__await__(self) → Generator[Any, Any, List[T]]

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) → ASyncIterator[T]: Return an iterator that yields {obj} from the {cls}.

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.

_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 ASyncIterator[T] 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 ASyncIterator[T].

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][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 ASyncIterator[T].

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 ASyncIterator[T], 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]

__args__ = (~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

Bases: IntEnum

mypyc filler docstring

__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.

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: int = 42161

Aurora: int = 1313161554

Avalanche: int = 43114

Base: int = 8453

Berachain: int = 80094

BinanceSmartChain: int = 56

Cronos: int = 25

Fantom: int = 250

Gnosis: int = 100

Harmony: int = 1666600000

Heco: int = 128

Katana: int = 747474

Mainnet: int = 1

Moonriver: int = 1285

OKEx: int = 66

Optimism: int = 10

Polygon: int = 137

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

label

name

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

printable

xDai: int = 100

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

__origin__: alias of ASyncIterable

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__(self) → Generator[Any, Any, List[T]]

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) → ASyncIterator[T]: Return an iterator that yields {obj} from the {cls}.

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.

_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][_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 ASyncIterator[T] 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 ASyncIterator[T].

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][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][_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 ASyncIterator[T].

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 ASyncIterator[T], 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]

__args__ = (~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 | Address | EthAddress | Contract)
input_address (str | HexBytes | AnyAddress | Address | 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 | Address | 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 | Address | EthAddress | Contract | int)
when_no_history_return_0 (bool)

Return type:

int

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) → eth_typing.evm.BlockNumber[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:: BlockNumber

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, faster_hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, 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[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_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 | Address | 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, faster_hexbytes.main.HexBytes, ~AnyAddress, evmspec.data._main.Address, 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[bool](name=`YPRICEMAGIC_SKIP_CACHE`, default_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 | Address | EthAddress | Contract | int])
block (int | BlockNumber | None)
fail_to_None (bool)
skip_cache (bool)
silent (bool)

Return type:

List[UsdPrice | None]

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 | Address | 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 | Address | EthAddress | Contract | int)
methods (Iterable[str])
_func (Callable)

Return type: