dank_mids.helpers package

Submodules

dank_mids.helpers.batch_size module

dank_mids.helpers.batch_size.log_check(batch_type, member_type, num_calls)

Parameters:

batch_type (Literal['multicall', 'json'])
member_type (Literal['calls', 'requests'])
num_calls (int)

Return type:

None

dank_mids.helpers.future module

final class dank_mids.helpers.future.DebuggableFuture

Bases: Future[T]

__init__(owner, loop)

Parameters:

owner (_RequestBase)
loop (AbstractEventLoop)

Return type:

None

__iter__(): Implement iter(self).

add_done_callback()

Add a callback to be run when the future becomes done.

The callback is called with a single argument - the future object. If the future is already done when this is called, the callback is scheduled with call_soon.

cancel(msg=None)

Cancel the future and schedule callbacks.

If the future is already done or cancelled, return False. Otherwise, change the future’s state to cancelled, schedule the callbacks and return True.

cancelled(): Return True if the future was cancelled.

done()

Return True if the future is done.

Done means either that a result / exception are available, or that the future was cancelled.

exception()

Return the exception that was set on this future.

The exception (or None if no exception was set) is returned only if the future is done. If the future has been cancelled, raises CancelledError. If the future isn’t done yet, raises InvalidStateError.

get_loop(): Return the event loop the Future is bound to.

remove_done_callback(fn, /)

Remove all instances of a callback from the “call when done” list.

Returns the number of callbacks removed.

result()

Return the result this future represents.

If the future has been cancelled, raises CancelledError. If the future’s result isn’t yet available, raises InvalidStateError. If the future is done and has an exception set, this exception is raised.

set_exception(exc)

Mark the future done and set an exception.

If the future is already done when this method is called, raises InvalidStateError.

Parameters:: exc (type | BaseException)
Return type:: None

set_result(value)

Mark the future done and set its result.

If the future is already done when this method is called, raises InvalidStateError.

Parameters:: value (T)
Return type:: None

property has_waiters: bool

property waiter_count: int

dank_mids.helpers.hashing module

final class dank_mids.helpers.hashing.AttributeDict

Bases: Generic[TKey, TValue]

Provides superficial immutability, someone could hack around it

__getitem__(key)

Parameters:: key (TKey)
Return type:: TValue

__init__(dictionary, *args, **kwargs)

Parameters:

dictionary (dict[TKey, TValue])
args (TKey)
kwargs (TValue)

Return type:

None

__iter__()

Return type:: Iterator[Any]

items()

Return type:: ItemsView[TKey, TValue]

keys()

Return type:: KeysView[TKey]

classmethod recursive(value)

Recursively convert mappings to ReadableAttributeDict instances and process nested collections (e.g., lists, sets, and dictionaries).

Parameters:: value (TValue)
Return type:: Any

values()

Return type:: ValuesView[TValue]

dank_mids.helpers.hashing.make_hashable(obj)

Converts an object into a hashable type if possible.

This function is used internally to ensure that objects can be used as dictionary keys or in sets.

Parameters:: obj (Any) – The object to make hashable.
Return type:: Hashable

dank_mids.helpers.hashing.tupleize_lists_nested(d)

Unhashable types inside dicts will throw an error if attempted to be hashed. This method converts lists to tuples, rendering them hashable. Other unhashable types found will raise a TypeError

Parameters:: d (AttributeDict[TKey, TValue] | Mapping[TKey, TValue])
Return type:: AttributeDict[TKey, TValue]

dank_mids.helpers.lru_cache module

dank_mids.helpers.lru_cache.lru_cache_lite(func)

A faster implementation of functools.lru_cache with no maxsize.

This decorator caches the results of the decorated function without imposing a maximum cache size. It uses a unique internal cache-miss marker to determine if a result is cached.

Example:

@lru_cache_lite
def add(x: int, y: int) -> int:
    print("Computing add...")
    return x + y

# First call with same arguments results in function execution.
print(add(2, 3))  # prints "Computing add..." then 5

# Subsequent calls with the same arguments return the cached result.
print(add(2, 3))  # prints 5 without "Computing add..."

See Also:

functools.lru_cache
lru_cache_lite_nonull() for a variant using None as the cache-miss marker.

Parameters:: func (Callable[[~__P], __T])
Return type:: Callable[[~__P], __T]

dank_mids.helpers.lru_cache.lru_cache_lite_nonull(func)

A faster implementation of functools.lru_cache with no maxsize that uses None as the cache miss indicator.

This decorator behaves similarly to lru_cache_lite() but differs by using None as the default value returned by the cache when a key is missing. Consequently, if the decorated function legitimately returns None for a given input, the cache will interpret the cached None as a miss and will re-evaluate the function.

Example:

@lru_cache_lite_nonull
def might_return_none(x: int) -> int:
    print("Executing might_return_none...")
    return x if x > 0 else None

# First call with x = -1 triggers a cache miss and function execution.
result = might_return_none(-1)
print(result)  # prints "Executing might_return_none..." then None

# Second call with x = -1 also triggers a cache miss because None is used as the cache-miss marker.
result = might_return_none(-1)
print(result)  # prints "Executing might_return_none..." then None

Warnings:

Use lru_cache_lite_nonull() only when it is acceptable that a cached return value of None results in re-execution of the decorated function. If the function may validly return None, consider using lru_cache_lite() instead.

See Also:

lru_cache_lite() which uses a unique cache-miss marker.
functools.lru_cache

Parameters:: func (Callable[[~__P], __T])
Return type:: Callable[[~__P], __T]

dank_mids.helpers.method module

dank_mids.helpers.method.get_len(method)

Parameters:: method (str)
Return type:: int

dank_mids.helpers.method.should_batch(method)

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

dank_mids.helpers.method.BYPASS_METHODS: Final = ('eth_blockNumber', 'eth_getLogs', 'trace_', 'debug_'): A tuple of method names that should bypass batching. These methods are typically handled separately or have special requirements.

Module contents

final class dank_mids.helpers.DebuggableFuture

Bases: Future[T]

__init__(owner, loop)

Parameters:

owner (_RequestBase)
loop (AbstractEventLoop)

Return type:

None

__iter__(): Implement iter(self).

add_done_callback()

Add a callback to be run when the future becomes done.

The callback is called with a single argument - the future object. If the future is already done when this is called, the callback is scheduled with call_soon.

cancel(msg=None)

Cancel the future and schedule callbacks.

If the future is already done or cancelled, return False. Otherwise, change the future’s state to cancelled, schedule the callbacks and return True.

cancelled(): Return True if the future was cancelled.

done()

Return True if the future is done.

Done means either that a result / exception are available, or that the future was cancelled.

exception()

Return the exception that was set on this future.

The exception (or None if no exception was set) is returned only if the future is done. If the future has been cancelled, raises CancelledError. If the future isn’t done yet, raises InvalidStateError.

get_loop(): Return the event loop the Future is bound to.

remove_done_callback(fn, /)

Remove all instances of a callback from the “call when done” list.

Returns the number of callbacks removed.

result()

Return the result this future represents.

If the future has been cancelled, raises CancelledError. If the future’s result isn’t yet available, raises InvalidStateError. If the future is done and has an exception set, this exception is raised.

set_exception(exc)

Mark the future done and set an exception.

If the future is already done when this method is called, raises InvalidStateError.

Parameters:: exc (type | BaseException)
Return type:: None

set_result(value)

Mark the future done and set its result.

If the future is already done when this method is called, raises InvalidStateError.

Parameters:: value (T)
Return type:: None

property has_waiters: bool

property waiter_count: int

async dank_mids.helpers.gatherish(coros, *, name=None)

An implementation of asyncio.gather optimized for use cases that:

Have coroutines that are predomininately sync
Expect occasional Exceptions but do not want to propagate them to the other tasks
Do not need the results of the tasks returned”

Parameters:

coros (Iterable[Coroutine[Any, Any, Any]])
name (str | None)

Return type:

None

dank_mids.helpers.lru_cache_lite(func)

A faster implementation of functools.lru_cache with no maxsize.

This decorator caches the results of the decorated function without imposing a maximum cache size. It uses a unique internal cache-miss marker to determine if a result is cached.

Example:

@lru_cache_lite
def add(x: int, y: int) -> int:
    print("Computing add...")
    return x + y

# First call with same arguments results in function execution.
print(add(2, 3))  # prints "Computing add..." then 5

# Subsequent calls with the same arguments return the cached result.
print(add(2, 3))  # prints 5 without "Computing add..."

See Also:

functools.lru_cache
lru_cache_lite_nonull() for a variant using None as the cache-miss marker.

Parameters:: func (Callable[[~__P], __T])
Return type:: Callable[[~__P], __T]

dank_mids.helpers.lru_cache_lite_nonull(func)

A faster implementation of functools.lru_cache with no maxsize that uses None as the cache miss indicator.

This decorator behaves similarly to lru_cache_lite() but differs by using None as the default value returned by the cache when a key is missing. Consequently, if the decorated function legitimately returns None for a given input, the cache will interpret the cached None as a miss and will re-evaluate the function.

Example:

@lru_cache_lite_nonull
def might_return_none(x: int) -> int:
    print("Executing might_return_none...")
    return x if x > 0 else None

# First call with x = -1 triggers a cache miss and function execution.
result = might_return_none(-1)
print(result)  # prints "Executing might_return_none..." then None

# Second call with x = -1 also triggers a cache miss because None is used as the cache-miss marker.
result = might_return_none(-1)
print(result)  # prints "Executing might_return_none..." then None

Warnings:

Use lru_cache_lite_nonull() only when it is acceptable that a cached return value of None results in re-execution of the decorated function. If the function may validly return None, consider using lru_cache_lite() instead.

See Also:

lru_cache_lite() which uses a unique cache-miss marker.
functools.lru_cache

Parameters:: func (Callable[[~__P], __T])
Return type:: Callable[[~__P], __T]

dank_mids.helpers.make_hashable(obj)

Converts an object into a hashable type if possible.

This function is used internally to ensure that objects can be used as dictionary keys or in sets.

Parameters:: obj (Any) – The object to make hashable.
Return type:: Hashable

dank_mids.helpers.setup_dank_w3(async_w3)

Sets up a DankWeb3 instance from a given Web3 instance.

Parameters:: async_w3 (Web3) – The Web3 instance to be wrapped.
Returns:: A new DankWeb3 instance with Dank Middleware injected.
Return type:: DankWeb3

dank_mids.helpers.setup_dank_w3_from_sync(sync_w3)

Sets up a DankWeb3 instance from a given synchronous Web3 instance.

Parameters:: sync_w3 (Web3) – The synchronous Web3 instance to be wrapped.
Returns:: A new DankWeb3 instance with Dank Middleware injected, supporting batched asynchronous operations.
Return type:: DankWeb3