multicall.py

Install the multicall package using pip to use the Python interface for MakerDAO's multicall.

pip install multicall

The Multicall implementation uses an internal NotSoBrightBatcher to manage large volumes of calls.

Batching Logic:

• Calls are split into chunks based on batcher.step (defaulting to 10,000).
• If a batch fails due to size or connectivity issues, the rebatch mechanism automatically reduces the step size (splitting the failed batch in half) and retries.

Error Handling:

• The library automatically retries on specific recoverable errors, such as:
  • aiohttp.ClientResponseError with "request entity too large" or "connection reset by peer".
  • requests.HTTPError with "request entity too large", "payload too large", "time-out", or "520 server error".
  • ValueError containing "out of gas" (unless it's a single-call batch).
  • TimeoutError.
• If an error is not in the recoverable list, it is raised immediately.

You can tune the library behavior using the following environment variables:

The Call(target, function, returns) class represents a single contract call.

Parameters:

• target: The to address for the eth_call.
• function: Either a seth-style signature string method(input,types)(output,types) or a list containing [signature, *args].
• returns: A list of tuples (name, handler) for processing return values.
  • If omitted, the call returns a tuple.
  • If provided, the call returns a dictionary.
  • To skip processing a specific value, pass None as the handler.

Usage Patterns:

• Predefined args: Call(... )() executes the call with arguments provided during initialization.
• Dynamic args: Call(...)(args) allows you to reuse a prepared call object with different arguments.
• Decoding: Use decode_output(output) to decode raw output and apply the returns handlers.

The Multicall(calls) class aggregates a list of Call objects.

Parameters:

• calls: A list of Call objects that have already had their arguments prepared.

Usage:

• Call the instance Multicall(...)() to execute the batch and retrieve the results.

The Signature(signature) class handles seth-style function signatures in the format function_name(input,types)(output,types). It supports complex types like structs, provided they are broken down into basic parts (e.g., (address,bytes)[]).

Use the following methods for data manipulation:

• encode_data(args): Encodes input arguments into calldata.
• decode_data(output): Decodes the output data.

The multicall package provides a structured way to define and execute multiple blockchain calls using Signature, Call, and Multicall abstractions.

• Signature: Defines the cryptographic signature or identity used for the calls.
• Call: Represents a single target function call, including the target address, the function selector, and the expected return types.
• Multicall: Orchestrates a collection of Call objects into a single batch execution.

from multicall import Signature, Call, Multicall

# Example workflow (conceptual):
signature = Signature("...")
calls = [
    Call(target="0x...", function="0x...", returns=["uint256"]),
    Call(target="0x...", function="0x...", returns=["bool"])
]
multicall = Multicall(calls)
# Execute using the multicall instance...

The setup_logger function initializes a logging.Logger instance for a given name. If the environment variable MULTICALL_DEBUG is set to a truthy value, the logger is automatically configured to output to a StreamHandler (stdout/stderr) and its level is set to logging.DEBUG.

import os
from multicall.loggers import setup_logger

# To enable debug logging, ensure the environment variable is set
os.environ["MULTICALL_DEBUG"] = "True"

logger = setup_logger("my_app")
logger.debug("This will be visible if MULTICALL_DEBUG is set")

Once a Call object is instantiated, you can execute it using several methods:

1. Synchronous execution: Use call() (via __call__). This method includes automatic retries using eth_retry.
2. Asynchronous execution: Use await call.coroutine() for non-blocking execution. This also supports automatic retries and respects semaphore limits for concurrency control.

Both methods allow you to override the Web3 instance or the block_id at runtime.

# Synchronous
result = call()

# Asynchronous
result = await call.coroutine(block_id=12345)