y.utils package

Submodules

y.utils.cache module

y.utils.cache.optional_async_diskcache(fn)[source]

Decorator to optionally apply disk caching to asynchronous functions.

If toolcache is not installed, this decorator will return the function as is, without any caching applied.

Parameters:: fn (Callable[[~P], Awaitable[T]] | Callable[[~P], T]) – The function to be decorated.
Return type:: Callable[[~P], Awaitable[T]] | Callable[[~P], T]

Examples

Using the decorator with an asynchronous function:

>>> async def fetch_data():
...     return "data"
>>> fetch_data = optional_async_diskcache(fetch_data)

See also

toolcache

y.utils.checks module

Utility functions for performing various checks.

y.utils.checks.hasall(obj, attrs)[source]

Check if an object has all the specified attributes.

Parameters:

obj (Any) – The object to check.
attrs (Iterable[str]) – An iterable of attribute names to check for.

Returns:

True if the object has all the specified attributes, False otherwise.

Return type:

bool

Example

>>> class TestClass:
...     attr1 = 1
...     attr2 = 2
>>> test_obj = TestClass()
>>> hasall(test_obj, ['attr1', 'attr2'])
True
>>> hasall(test_obj, ['attr1', 'attr3'])
False

y.utils.client module

Utility functions for retrieving Ethereum client information.

y.utils.client.get_ethereum_client()[source]

Get the Ethereum client type for the current connection.

Returns:: A string representing the Ethereum client type, such as ‘geth’, ‘erigon’, or ‘tg’.
Return type:: str

Examples

>>> from y.utils.client import get_ethereum_client
>>> client = get_ethereum_client()
>>> print(client)
'geth'

>>> # Example with a different client
>>> client = get_ethereum_client()
>>> print(client)
'erigon'

See also

get_ethereum_client_async() for the asynchronous version of this function.

y.utils.events module

y.utils.fakes module

y.utils.gather module

async y.utils.gather.gather_methods()

y.utils.logging module

final class y.utils.logging.PriceLogger[source]

Bases: Logger

__init__(name, level=0): Initialize the logger with a name and an optional level.

addFilter(filter): Add the specified filter to this handler.

addHandler(hdlr): Add the specified handler to this logger.

callHandlers(record)

Pass a record to all relevant handlers.

Loop through all handlers for this logger and its parents in the logger hierarchy. If no handler was found, output a one-off error message to sys.stderr. Stop searching up the hierarchy whenever a logger with the “propagate” attribute set to zero is found - that will be the last logger whose handlers are called.

close()[source]

Return type:: None

critical(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘CRITICAL’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.critical(“Houston, we have a %s”, “major disaster”, exc_info=True)

debug(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘DEBUG’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.debug(“Houston, we have a %s”, “thorny problem”, exc_info=True)

error(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘ERROR’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.error(“Houston, we have a %s”, “major problem”, exc_info=True)

exception(msg, *args, exc_info=True, **kwargs): Convenience method for logging an ERROR with exception information.

fatal(msg, *args, **kwargs): Don’t use this method, use critical() instead.

filter(record)

Determine if a record is loggable by consulting all the filters.

The default is to allow the record to be logged; any filter can veto this by returning a false value. If a filter attached to a handler returns a log record instance, then that instance is used in place of the original log record in any further processing of the event by that handler. If a filter returns any other true value, the original log record is used in any further processing of the event by that handler.

If none of the filters return false values, this method returns a log record. If any of the filters return a false value, this method returns a false value.

Changed in version 3.2: Allow filters to be just callables.

Changed in version 3.12: Allow filters to return a LogRecord instead of modifying it in place.

findCaller(stack_info=False, stacklevel=1): Find the stack frame of the caller so that we can note the source file name, line number and function name.

getChild(suffix)

Get a logger which is a descendant to this one.

This is a convenience method, such that

logging.getLogger(‘abc’).getChild(‘def.ghi’)

is the same as

logging.getLogger(‘abc.def.ghi’)

It’s useful, for example, when the parent logger is named using __name__ rather than a literal string.

getChildren()

getEffectiveLevel()

Get the effective level for this logger.

Loop through this logger and its parents in the logger hierarchy, looking for a non-zero logging level. Return the first one found.

handle(record)

Call the handlers for the specified record.

This method is used for unpickled records received from a socket, as well as those created locally. Logger-level filtering is applied.

hasHandlers()

See if this logger has any handlers configured.

Loop through all handlers for this logger and its parents in the logger hierarchy. Return True if a handler was found, else False. Stop searching up the hierarchy whenever a logger with the “propagate” attribute set to zero is found - that will be the last logger which is checked for the existence of handlers.

info(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘INFO’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.info(“Houston, we have a %s”, “notable problem”, exc_info=True)

isEnabledFor(level): Is this logger enabled for level ‘level’?

log(level, msg, *args, **kwargs)

Log ‘msg % args’ with the integer severity ‘level’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.log(level, “We have a %s”, “mysterious problem”, exc_info=True)

makeRecord(name, level, fn, lno, msg, args, exc_info, func=None, extra=None, sinfo=None): A factory method which can be overridden in subclasses to create specialized LogRecords.

removeFilter(filter): Remove the specified filter from this handler.

removeHandler(hdlr): Remove the specified handler from this logger.

setLevel(level): Set the logging level of this logger. level must be an int or a str.

warn(msg, *args, **kwargs)

warning(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘WARNING’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.warning(“Houston, we have a %s”, “bit of a problem”, exc_info=True)

__final__ = True

address: ChecksumAddress

block: BlockNumber

debug_task: Task[None] | None

enabled: bool

key: tuple[str | HexBytes | AnyAddress | Address | EthAddress | Contract | int, int | BlockNumber, str | None, str]

manager = <logging.Manager object>

root = <RootLogger root (WARNING)>

y.utils.logging.enable_debug_logging(logger='y')[source]

Enables ypricemagic’s debugging mode. Very verbose.

Parameters:: logger (str) – The name of the logger to enable debugging for. Defaults to “y”.
Return type:: None

Example

>>> enable_debug_logging("y")

y.utils.logging.get_price_logger(token_address, block, *, symbol=None, extra='', start_task=False)[source]

Create or retrieve a PriceLogger instance for a given token address and block.

This function manages a cache of loggers to ensure they have the proper members for ypricemagic. If a logger is enabled for DEBUG, it will start a debug task if specified.

Parameters:

token_address (str | HexBytes | AnyAddress | Address | EthAddress | Contract | int) – The address of the token.
block (int | BlockNumber) – The block number.
symbol (str) – An optional symbol for the token.
extra (str) – An optional extra string to append to the logger name.
start_task (bool) – Whether to start a debug task. Defaults to False.

Return type:

PriceLogger

Example

>>> logger = get_price_logger("0xTokenAddress", 123456)
>>> logger.debug("This is a debug message.")

See also

enable_debug_logging()

y.utils.middleware module

y.utils.multicall module

y.utils.raw_calls module

Module contents

Provides utility functions for the ypricemagic library.

This module includes functions for caching, attribute checking, and gathering results from multiple contract method calls. These utilities are used throughout the library to enhance performance and ensure code reliability.

Imported Functions:

y.utils.cache.a_sync_ttl_cache(): Provides a caching mechanism with a time-to-live (TTL) feature.
y.utils.checks.hasall(): Checks if an object has all specified attributes.
y.utils.gather.gather_methods(): Asynchronously gathers results by calling multiple contract method signatures on a given contract address.

Examples

Using the caching utility:

>>> from y.utils.cache import a_sync_ttl_cache
>>> @a_sync_ttl_cache
... def expensive_function(x):
...     return x * x
>>> result = expensive_function(4)

Checking for attributes:

>>> from y.utils.checks import hasall
>>> class MyClass:
...     attr1 = 1
...     attr2 = 2
>>> obj = MyClass()
>>> hasall(obj, ['attr1', 'attr2'])
True

Gathering contract method results:

>>> from y.utils.gather import gather_methods
>>> contract_address = "0x6B175474E89094C44Da98b954EedeAC495271d0F"  # Example contract address
>>> methods = ["name()", "symbol()", "decimals()"]
>>> results = await gather_methods(contract_address, methods)
>>> print(results)
('Dai Stablecoin', 'DAI', 18)

See also

y.utils.cache for caching utilities.
y.utils.checks for attribute checking utilities.
y.utils.gather for gathering results from multiple contract method calls.

async y.utils.gather_methods()