dank_mids package

Subpackages

Submodules

dank_mids.ENVIRONMENT_VARIABLES module

dank_mids.ENVIRONMENT_VARIABLES.AIOHTTP_TIMEOUT = <EnvironmentVariable[name=`DANKMIDS_AIOHTTP_TIMEOUT`, type=int, default_value=1200, current_value=1200, using_default=True]>

Timeout value in seconds for aiohttp requests.

dank_mids.ENVIRONMENT_VARIABLES.BROWNIE_CALL_SEMAPHORE = <EnvironmentVariable[name=`DANKMIDS_BROWNIE_CALL_SEMAPHORE`, type=BlockSemaphore, default_value=100000, current_value=100000, using_default=True]>

Semaphore for limiting concurrent Brownie calls.

See also

dank_mids.semaphores.BlockSemaphore: The semaphore class used for concurrency control.

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

dank_mids.ENVIRONMENT_VARIABLES.BROWNIE_DECODER_PROCESSES = <EnvironmentVariable[name=`DANKMIDS_BROWNIE_DECODER_PROCESSES`, type=AsyncProcessPoolExecutor, default_value=0, current_value=0, using_default=True]>

Process pool for Brownie decoding operations.

See also

a_sync.AsyncProcessPoolExecutor: The executor class used for managing asynchronous processes.

dank_mids.ENVIRONMENT_VARIABLES.BROWNIE_ENCODER_PROCESSES = <EnvironmentVariable[name=`DANKMIDS_BROWNIE_ENCODER_PROCESSES`, type=AsyncProcessPoolExecutor, default_value=0, current_value=0, using_default=True]>

Process pool for Brownie encoding operations.

See also

a_sync.AsyncProcessPoolExecutor: The executor class used for managing asynchronous processes.

dank_mids.ENVIRONMENT_VARIABLES.BROWNIE_ENCODER_SEMAPHORE = <EnvironmentVariable[name=`DANKMIDS_BROWNIE_ENCODER_SEMAPHORE`, type=BlockSemaphore, default_value=200000, current_value=200000, using_default=True]>

Semaphore for limiting concurrent Brownie encoding operations. This limits memory consumption.

See also

dank_mids.semaphores.BlockSemaphore: The semaphore class used for concurrency control.

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

dank_mids.ENVIRONMENT_VARIABLES.COLLECTION_FACTOR = <EnvironmentVariable[name=`DANKMIDS_COLLECTION_FACTOR`, type=int, default_value=10, current_value=10, using_default=True]>

Factor determining the size of data collection operations.

dank_mids.ENVIRONMENT_VARIABLES.GANACHE_FORK = <EnvironmentVariable[name=`DANKMIDS_GANACHE_FORK`, type=bool, default_value=False, current_value=False, using_default=True]>

Flag indicating whether the current environment is a Ganache fork.

dank_mids.ENVIRONMENT_VARIABLES.MULTICALL_DECODER_PROCESSES = <EnvironmentVariable[name=`DANKMIDS_MULTICALL_DECODER_PROCESSES`, type=AsyncProcessPoolExecutor, default_value=0, current_value=0, using_default=True]>

Process pool for Multicall decoding operations.

See also

a_sync.AsyncProcessPoolExecutor: The executor class used for managing asynchronous processes.

dank_mids.ENVIRONMENT_VARIABLES.USE_FULL_REQUEST = <EnvironmentVariable[name=`DANKMIDS_USE_FULL_REQUEST`, type=bool, default_value=False, current_value=False, using_default=True]>

Flag indicating whether to use the full request specification.

dank_mids.constants module

dank_mids.constants.GAS_LIMIT = 50000000

The gas limit constant imported from the multicall library.

This value is used as the default gas limit for multicall operations.

dank_mids.constants.MULTICALL2_DEPLOY_BLOCKS: Dict[Network, BlockNumber] = {Network.Mainnet: 12336033, Network.Optimism: 722566, Network.Fantom: 16572242, Network.Arbitrum: 821923}

A dictionary mapping networks to the block numbers where Multicall2 was deployed.

dank_mids.constants.MULTICALL2_OVERRIDE_CODE = '0x608060405234801561001057600080fd5b50600436106100b45760003560e01c806372425d9d1161007157806372425d9d1461013d57806386d516e814610145578063a8b0574e1461014d578063bce38bd714610162578063c3077fa914610182578063ee82ac5e14610195576100b4565b80630f28c97d146100b9578063252dba42146100d757806327e86d6e146100f8578063399542e91461010057806342cbb15c146101225780634d2301cc1461012a575b600080fd5b6100c16101a8565b6040516100ce919061083b565b60405180910390f35b6100ea6100e53660046106bb565b6101ac565b6040516100ce9291906108ba565b6100c1610340565b61011361010e3660046106f6565b610353565b6040516100ce93929190610922565b6100c161036b565b6100c161013836600461069a565b61036f565b6100c161037c565b6100c1610380565b610155610384565b6040516100ce9190610814565b6101756101703660046106f6565b610388565b6040516100ce9190610828565b6101136101903660046106bb565b610533565b6100c16101a3366004610748565b610550565b4290565b8051439060609067ffffffffffffffff8111156101d957634e487b7160e01b600052604160045260246000fd5b60405190808252806020026020018201604052801561020c57816020015b60608152602001906001900390816101f75790505b50905060005b835181101561033a5760008085838151811061023e57634e487b7160e01b600052603260045260246000fd5b6020026020010151600001516001600160a01b031686848151811061027357634e487b7160e01b600052603260045260246000fd5b60200260200101516020015160405161028c91906107f8565b6000604051808303816000865af19150503d80600081146102c9576040519150601f19603f3d011682016040523d82523d6000602084013e6102ce565b606091505b5091509150816102f95760405162461bcd60e51b81526004016102f090610885565b60405180910390fd5b8084848151811061031a57634e487b7160e01b600052603260045260246000fd5b602002602001018190525050508080610332906109c2565b915050610212565b50915091565b600061034d60014361097b565b40905090565b43804060606103628585610388565b90509250925092565b4390565b6001600160a01b03163190565b4490565b4590565b4190565b6060815167ffffffffffffffff8111156103b257634e487b7160e01b600052604160045260246000fd5b6040519080825280602002602001820160405280156103eb57816020015b6103d8610554565b8152602001906001900390816103d05790505b50905060005b825181101561052c5760008084838151811061041d57634e487b7160e01b600052603260045260246000fd5b6020026020010151600001516001600160a01b031685848151811061045257634e487b7160e01b600052603260045260246000fd5b60200260200101516020015160405161046b91906107f8565b6000604051808303816000865af19150503d80600081146104a8576040519150601f19603f3d011682016040523d82523d6000602084013e6104ad565b606091505b509150915085156104d557816104d55760405162461bcd60e51b81526004016102f090610844565b604051806040016040528083151581526020018281525084848151811061050c57634e487b7160e01b600052603260045260246000fd5b602002602001018190525050508080610524906109c2565b9150506103f1565b5092915050565b6000806060610543600185610353565b9196909550909350915050565b4090565b60408051808201909152600081526060602082015290565b80356001600160a01b038116811461058357600080fd5b919050565b600082601f830112610598578081fd5b8135602067ffffffffffffffff808311156105b5576105b56109f3565b6105c2828385020161094a565b83815282810190868401865b8681101561068c57813589016040601f198181848f030112156105ef578a8bfd5b6105f88261094a565b6106038a850161056c565b81528284013589811115610615578c8dfd5b8085019450508d603f850112610629578b8cfd5b898401358981111561063d5761063d6109f3565b61064d8b84601f8401160161094a565b92508083528e84828701011115610662578c8dfd5b808486018c85013782018a018c9052808a01919091528652505092850192908501906001016105ce565b509098975050505050505050565b6000602082840312156106ab578081fd5b6106b48261056c565b9392505050565b6000602082840312156106cc578081fd5b813567ffffffffffffffff8111156106e2578182fd5b6106ee84828501610588565b949350505050565b60008060408385031215610708578081fd5b82358015158114610717578182fd5b9150602083013567ffffffffffffffff811115610732578182fd5b61073e85828601610588565b9150509250929050565b600060208284031215610759578081fd5b5035919050565b60008282518085526020808601955080818302840101818601855b848110156107bf57858303601f19018952815180511515845284015160408585018190526107ab818601836107cc565b9a86019a945050509083019060010161077b565b5090979650505050505050565b600081518084526107e4816020860160208601610992565b601f01601f19169290920160200192915050565b6000825161080a818460208701610992565b9190910192915050565b6001600160a01b0391909116815260200190565b6000602082526106b46020830184610760565b90815260200190565b60208082526021908201527f4d756c746963616c6c32206167677265676174653a2063616c6c206661696c656040820152601960fa1b606082015260800190565b6020808252818101527f4d756c746963616c6c206167677265676174653a2063616c6c206661696c6564604082015260600190565b600060408201848352602060408185015281855180845260608601915060608382028701019350828701855b8281101561091457605f198887030184526109028683516107cc565b955092840192908401906001016108e6565b509398975050505050505050565b6000848252836020830152606060408301526109416060830184610760565b95945050505050565b604051601f8201601f1916810167ffffffffffffffff81118282101715610973576109736109f3565b604052919050565b60008282101561098d5761098d6109dd565b500390565b60005b838110156109ad578181015183820152602001610995565b838111156109bc576000848401525b50505050565b60006000198214156109d6576109d66109dd565b5060010190565b634e487b7160e01b600052601160045260246000fd5b634e487b7160e01b600052604160045260246000fdfea2646970667358221220c1152f751f29ece4d7bce5287ceafc8a153de9c2c633e3f21943a87d845bd83064736f6c63430008010033'

The bytecode for the Multicall2 contract.

This is used for state override on blocks before the Multicall2 contract was deployed.

dank_mids.constants.MULTICALL3_DEPLOY_BLOCKS: Dict[Network, BlockNumber] = {Network.Mainnet: 14353601, Network.Optimism: 4286263, Network.Fantom: 33001987, Network.Base: 5022, Network.Arbitrum: 7654707}

A dictionary mapping networks to the block numbers where Multicall3 was deployed.

dank_mids.constants.MULTICALL3_OVERRIDE_CODE = '0x6080604052600436106100f35760003560e01c80634d2301cc1161008a578063a8b0574e11610059578063a8b0574e1461025a578063bce38bd714610275578063c3077fa914610288578063ee82ac5e1461029b57600080fd5b80634d2301cc146101ec57806372425d9d1461022157806382ad56cb1461023457806386d516e81461024757600080fd5b80633408e470116100c65780633408e47014610191578063399542e9146101a45780633e64a696146101c657806342cbb15c146101d957600080fd5b80630f28c97d146100f8578063174dea711461011a578063252dba421461013a57806327e86d6e1461015b575b600080fd5b34801561010457600080fd5b50425b6040519081526020015b60405180910390f35b61012d610128366004610a85565b6102ba565b6040516101119190610bbe565b61014d610148366004610a85565b6104ef565b604051610111929190610bd8565b34801561016757600080fd5b50437fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff0140610107565b34801561019d57600080fd5b5046610107565b6101b76101b2366004610c60565b610690565b60405161011193929190610cba565b3480156101d257600080fd5b5048610107565b3480156101e557600080fd5b5043610107565b3480156101f857600080fd5b50610107610207366004610ce2565b73ffffffffffffffffffffffffffffffffffffffff163190565b34801561022d57600080fd5b5044610107565b61012d610242366004610a85565b6106ab565b34801561025357600080fd5b5045610107565b34801561026657600080fd5b50604051418152602001610111565b61012d610283366004610c60565b61085a565b6101b7610296366004610a85565b610a1a565b3480156102a757600080fd5b506101076102b6366004610d18565b4090565b60606000828067ffffffffffffffff8111156102d8576102d8610d31565b60405190808252806020026020018201604052801561031e57816020015b6040805180820190915260008152606060208201528152602001906001900390816102f65790505b5092503660005b8281101561047757600085828151811061034157610341610d60565b6020026020010151905087878381811061035d5761035d610d60565b905060200281019061036f9190610d8f565b6040810135958601959093506103886020850185610ce2565b73ffffffffffffffffffffffffffffffffffffffff16816103ac6060870187610dcd565b6040516103ba929190610e32565b60006040518083038185875af1925050503d80600081146103f7576040519150601f19603f3d011682016040523d82523d6000602084013e6103fc565b606091505b50602080850191909152901515808452908501351761046d577f08c379a000000000000000000000000000000000000000000000000000000000600052602060045260176024527f4d756c746963616c6c333a2063616c6c206661696c656400000000000000000060445260846000fd5b5050600101610325565b508234146104e6576040517f08c379a000000000000000000000000000000000000000000000000000000000815260206004820152601a60248201527f4d756c746963616c6c333a2076616c7565206d69736d6174636800000000000060448201526064015b60405180910390fd5b50505092915050565b436060828067ffffffffffffffff81111561050c5761050c610d31565b60405190808252806020026020018201604052801561053f57816020015b606081526020019060019003908161052a5790505b5091503660005b8281101561068657600087878381811061056257610562610d60565b90506020028101906105749190610e42565b92506105836020840184610ce2565b73ffffffffffffffffffffffffffffffffffffffff166105a66020850185610dcd565b6040516105b4929190610e32565b6000604051808303816000865af19150503d80600081146105f1576040519150601f19603f3d011682016040523d82523d6000602084013e6105f6565b606091505b5086848151811061060957610609610d60565b602090810291909101015290508061067d576040517f08c379a000000000000000000000000000000000000000000000000000000000815260206004820152601760248201527f4d756c746963616c6c333a2063616c6c206661696c656400000000000000000060448201526064016104dd565b50600101610546565b5050509250929050565b43804060606106a086868661085a565b905093509350939050565b6060818067ffffffffffffffff8111156106c7576106c7610d31565b60405190808252806020026020018201604052801561070d57816020015b6040805180820190915260008152606060208201528152602001906001900390816106e55790505b5091503660005b828110156104e657600084828151811061073057610730610d60565b6020026020010151905086868381811061074c5761074c610d60565b905060200281019061075e9190610e76565b925061076d6020840184610ce2565b73ffffffffffffffffffffffffffffffffffffffff166107906040850185610dcd565b60405161079e929190610e32565b6000604051808303816000865af19150503d80600081146107db576040519150601f19603f3d011682016040523d82523d6000602084013e6107e0565b606091505b506020808401919091529015158083529084013517610851577f08c379a000000000000000000000000000000000000000000000000000000000600052602060045260176024527f4d756c746963616c6c333a2063616c6c206661696c656400000000000000000060445260646000fd5b50600101610714565b6060818067ffffffffffffffff81111561087657610876610d31565b6040519080825280602002602001820160405280156108bc57816020015b6040805180820190915260008152606060208201528152602001906001900390816108945790505b5091503660005b82811015610a105760008482815181106108df576108df610d60565b602002602001015190508686838181106108fb576108fb610d60565b905060200281019061090d9190610e42565b925061091c6020840184610ce2565b73ffffffffffffffffffffffffffffffffffffffff1661093f6020850185610dcd565b60405161094d929190610e32565b6000604051808303816000865af19150503d806000811461098a576040519150601f19603f3d011682016040523d82523d6000602084013e61098f565b606091505b506020830152151581528715610a07578051610a07576040517f08c379a000000000000000000000000000000000000000000000000000000000815260206004820152601760248201527f4d756c746963616c6c333a2063616c6c206661696c656400000000000000000060448201526064016104dd565b506001016108c3565b5050509392505050565b6000806060610a2b60018686610690565b919790965090945092505050565b60008083601f840112610a4b57600080fd5b50813567ffffffffffffffff811115610a6357600080fd5b6020830191508360208260051b8501011115610a7e57600080fd5b9250929050565b60008060208385031215610a9857600080fd5b823567ffffffffffffffff811115610aaf57600080fd5b610abb85828601610a39565b90969095509350505050565b6000815180845260005b81811015610aed57602081850181015186830182015201610ad1565b81811115610aff576000602083870101525b50601f017fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffe0169290920160200192915050565b600082825180855260208086019550808260051b84010181860160005b84811015610bb1578583037fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffe001895281518051151584528401516040858501819052610b9d81860183610ac7565b9a86019a9450505090830190600101610b4f565b5090979650505050505050565b602081526000610bd16020830184610b32565b9392505050565b600060408201848352602060408185015281855180845260608601915060608160051b870101935082870160005b82811015610c52577fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa0888703018452610c40868351610ac7565b95509284019290840190600101610c06565b509398975050505050505050565b600080600060408486031215610c7557600080fd5b83358015158114610c8557600080fd5b9250602084013567ffffffffffffffff811115610ca157600080fd5b610cad86828701610a39565b9497909650939450505050565b838152826020820152606060408201526000610cd96060830184610b32565b95945050505050565b600060208284031215610cf457600080fd5b813573ffffffffffffffffffffffffffffffffffffffff81168114610bd157600080fd5b600060208284031215610d2a57600080fd5b5035919050565b7f4e487b7100000000000000000000000000000000000000000000000000000000600052604160045260246000fd5b7f4e487b7100000000000000000000000000000000000000000000000000000000600052603260045260246000fd5b600082357fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff81833603018112610dc357600080fd5b9190910192915050565b60008083357fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffe1843603018112610e0257600080fd5b83018035915067ffffffffffffffff821115610e1d57600080fd5b602001915036819003821315610a7e57600080fd5b8183823760009101908152919050565b600082357fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffc1833603018112610dc357600080fd5b600082357fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa1833603018112610dc357600080fdfea2646970667358221220bb2b5c71a328032f97c676ae39a1ec2148d3e5d6f73d95e9b17910152d61f16264736f6c634300080c0033'

The bytecode for the Multicall3 contract, if supported on the currently connected network.

If Multicall3 is not supported, this will fall back to the Multicall2 bytecode.

dank_mids.constants.RETRY_ERRS = ['connection reset by peer', 'server disconnected', 'execution aborted (timeout =', 'batch limit exceeded', 'request timed out']

A list of error messages that are expected during normal use and are not indicative of any problem(s).

These errors will be automatically retried until success is achieved.

dank_mids.constants.REVERT_SELECTORS = [b'\x08\xc3y\xa0', b'4e487b71']

A list of byte strings representing revert selectors.

These selectors are used to identify specific types of revert errors in Ethereum transactions.

dank_mids.constants.TOO_MUCH_DATA_ERRS = ['payload too large', 'content length too large', 'request entity too large', 'batch limit exceeded']

A list of error messages indicating that the request sent to the RPC was too large and must be split up.

These error messages are used to identify when a request needs to be broken into smaller chunks.

dank_mids.controller module

class dank_mids.controller.DankMiddlewareController

Bases: object

Controller for managing Dank Middleware operations.

This class handles the core functionality of Dank Mids, including request batching, call execution, and error handling.

See also

dank_mids.semaphores.BlockSemaphore: Used for managing concurrency of eth_calls made with the controller.

async __call__(method, params)

Asynchronous method to handle RPC calls.

This method routes different types of RPC calls to appropriate handlers, including specialized handling for eth_call and other methods that may use queues.

Parameters:

  • method (RPCEndpoint) – The RPC method to be called.

  • params (Any) – The parameters for the RPC call.

Returns:

The response from the RPC call.

Return type:

RPCResponse

__init__(w3)

Initialize the DankMiddlewareController.

Parameters:

w3 (Web3) – The Web3 instance used to make RPC requests.

Return type:

None

early_start()

Initiate processing of queued calls when a full batch is available.

This method combines pending eth_calls and other RPC calls into a single batch, clears the pending Ethereum calls queue, and starts processing the combined batch. It’s used to optimize processing by starting as soon as enough calls are queued to form a full batch.

async execute_batch()

Execute a batch of pending calls.

This method collects all pending eth calls and RPC calls, clears the pending queues, and executes them as a single batch.

Return type:

None

async make_request(method, params, request_id=None)

Makes an RPC request to the Ethereum node.

This method creates an RPC request with the given method and parameters, sends it to the node, and returns the raw response.

Parameters:

  • method (str) – The RPC method to call.

  • params (List[Any]) – The parameters for the RPC method.

  • request_id (int | None) – An optional request ID. If not provided, a new ID will be generated.

Returns:

The raw response from the Ethereum node.

Raises:

Exception – If DEBUG environment variable is set, any exception that occurs during the request is logged and re-raised.

Return type:

RawResponse

reduce_batch_size(num_calls)

Decrease the size of JSON-RPC batches in response to failures.

Similar to reduce_multicall_size(), this method is called when a JSON-RPC batch operation fails, allowing for dynamic adjustment of batch sizes.

Parameters:

num_calls (int) – The number of calls in the failed batch operation.

Return type:

None

reduce_multicall_size(num_calls)

Decrease the size of multicall batches in response to failures.

This method is called when a multicall operation fails, allowing the system to dynamically adjust the batch size to prevent future failures.

Parameters:

num_calls (int) – The number of calls in the failed multicall operation.

Return type:

None

set_batch_size_limit(new_limit)

Set a new limit for the JSON-RPC batch size.

Parameters:

new_limit (int) – The new maximum number of calls in a JSON-RPC batch.

Return type:

None

set_multicall_size_limit(new_limit)

Set a new limit for the multicall size.

Parameters:

new_limit (int) – The new maximum number of calls in a multicall.

Return type:

None

batcher

Batcher for RPC calls.

call_uid

Unique identifier generator for individual calls.

chain_id

The chainid for the currently connected rpc.

client_version: str

The client version for the currently connected rpc.

endpoint

The uri for the connected rpc.

eth_call_semaphores: BlockSemaphore

Used for managing concurrency of eth_calls.

method_queues

Queues for different method types.

multicall_uid: UIDGenerator

Unique identifier generator for multicall operations.

no_multicall: Set[ChecksumAddress]

A set of addresses that have issues when called from the multicall contract. Calls to these contracts will not be batched in multicalls.

pending_eth_calls: DefaultDict[BlockId, Multicall]

A dictionary of pending Multicalls by block. The Multicalls hold all pending eth_calls.

pending_rpc_calls

A JSONRPCBatch containing all pending rpc requests.

property queue_is_full: bool

Check if the queue of pending calls is full.

Returns:

True if the queue is full, False otherwise.

request_type

The Struct class the controller will use to encode requests.

request_uid: UIDGenerator

Unique identifier generator for RPC requests.

state_override_not_supported: bool

A boolean that indicates whether the connected rpc supports state override functionality.

sync_w3

A sync Web3 instance connected to the same rpc, used to make calls during init.

w3: Web3

The Web3 instance used to make rpc requests.

dank_mids.eth module

class dank_mids.eth.DankEth

Bases: AsyncEth

__init__(w3)
Parameters:

w3 (AsyncWeb3 | Web3)

Return type:

None

attach_methods(methods)
Parameters:

methods (Dict[str, Method[Callable[[...], Any]]])

Return type:

None

block_id_munger(account, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Tuple[Address | ChecksumAddress | ENS, Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int]

async call(transaction, block_identifier=None, state_override=None, ccip_read_enabled=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

  • state_override (Dict[ChecksumAddress, CallOverrideParams] | None)

  • ccip_read_enabled (bool | None)

Return type:

HexBytes

call_munger(transaction, block_identifier=None, state_override=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

  • state_override (Dict[ChecksumAddress, CallOverrideParams] | None)

Return type:

Tuple[TxParams, Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int] | ~typing.Tuple[~web3.types.TxParams, ~typing.Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int, ~typing.Dict[~eth_typing.evm.ChecksumAddress, ~web3.types.CallOverrideParams]]

contract(address=None, **kwargs)
Parameters:

  • address (Address | ChecksumAddress | ENS | None)

  • kwargs (Any)

Return type:

Type[AsyncContract] | AsyncContract

async create_access_list(transaction, block_identifier=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

CreateAccessListResponse

create_access_list_munger(transaction, block_identifier=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Tuple[TxParams, Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int]

async estimate_gas(transaction, block_identifier=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

int

estimate_gas_munger(transaction, block_identifier=None)
Parameters:

  • transaction (TxParams)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Sequence[TxParams | Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int]

async fee_history(block_count, newest_block, reward_percentiles=None)
Parameters:

  • block_count (int)

  • newest_block (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber)

  • reward_percentiles (List[float] | None)

Return type:

FeeHistory

filter_munger(filter_params=None, filter_id=None)
Parameters:

  • filter_params (str | FilterParams | None)

  • filter_id (HexStr | None)

Return type:

List[FilterParams] | List[HexStr] | List[str]

generate_gas_price(transaction_params=None)
Parameters:

transaction_params (TxParams | None)

Return type:

Wei | None

async get_balance(account, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Wei

async get_block(block_identifier, full_transactions=False)
Parameters:

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int)

  • full_transactions (bool)

Return type:

BlockData

get_block_munger(block_identifier, full_transactions=False)
Parameters:

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int)

  • full_transactions (bool)

Return type:

Tuple[Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int, bool]

async get_block_timestamp(block_identifier)

Retrieves only the timestamp from a specific block.

This method skips decoding the rest of the Block response data.

Parameters:

  • block_identifier (int) – The block number from which to retrieve the transactions.

  • hashes_only – If True, only transaction hashes will be returned.

Returns:

A list of Transaction data objects from the block, or a list of transaction hashes.

Return type:

UnixTimestamp

Example

>>> [print(tx.hash) for tx in await dank_mids.eth.get_transactions(12345678)]
async get_code(account, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

HexBytes

async get_logs(*args, decode_to=typing.Tuple[evmspec.log.Log, ...], decode_hook=<function _decode_hook>, **kwargs)
Parameters:
Return type:

T

async get_raw_transaction(transaction_hash)
Parameters:

transaction_hash (Hash32 | HexBytes | HexStr)

Return type:

HexBytes

async get_raw_transaction_by_block(block_identifier, index)
Parameters:

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int)

  • index (int)

Return type:

HexBytes

async get_storage_at(account, position, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • position (int)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

HexBytes

get_storage_at_munger(account, position, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • position (int)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Tuple[Address | ChecksumAddress | ENS, int, Literal[‘latest’, ‘earliest’, ‘pending’, ‘safe’, ‘finalized’] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int]

async get_transaction(transaction_hash)
Parameters:

transaction_hash (str)

Return type:

TransactionLegacy | Transaction2930 | Transaction1559 | TransactionRLP

async get_transaction_by_block(block_identifier, index)
Parameters:

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int)

  • index (int)

Return type:

TxData

async get_transaction_count(account, block_identifier=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int | None)

Return type:

Nonce

async get_transaction_receipt(*args, decode_to=<class 'evmspec.receipt.TransactionReceipt'>, decode_hook=<function _decode_hook>, **kwargs)
Parameters:
Return type:

T

async get_transaction_status(transaction_hash)
Parameters:

transaction_hash (str)

Return type:

Status

async get_transactions(block_identifier: int | HexStr) → List[TransactionLegacy | Transaction2930 | Transaction1559]
async get_transactions(block_identifier: int | HexStr, hashes_only: Literal[True]) → List[TransactionHash]
async get_transactions(block_identifier: int | HexStr, hashes_only: Literal[False]) → List[TransactionLegacy | Transaction2930 | Transaction1559]

Retrieves only the transactions from a specific block.

This method skips decoding the rest of the Block response data.

Parameters:

  • block_identifier – The block number from which to retrieve the transactions.

  • hashes_only – If True, only transaction hashes will be returned.

Returns:

A list of Transaction data objects from the block, or a list of transaction hashes.

Example

>>> [print(tx.hash) for tx in await dank_mids.eth.get_transactions(12345678)]
async get_uncle_count(block_identifier)
Parameters:

block_identifier (Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | ~eth_typing.evm.BlockNumber | ~eth_typing.evm.Hash32 | ~eth_typing.encoding.HexStr | ~hexbytes.main.HexBytes | int)

Return type:

int

icap_namereg()
Return type:

NoReturn

async modify_transaction(transaction_hash, **transaction_params)
Parameters:

  • transaction_hash (Hash32 | HexBytes | HexStr)

  • transaction_params (Any)

Return type:

HexBytes

namereg()
Return type:

NoReturn

async replace_transaction(transaction_hash, new_transaction)
Parameters:

  • transaction_hash (Hash32 | HexBytes | HexStr)

  • new_transaction (TxParams)

Return type:

HexBytes

async send_raw_transaction(transaction)
Parameters:

transaction (HexStr | bytes)

Return type:

HexBytes

async send_transaction(transaction)
Parameters:

transaction (TxParams)

Return type:

HexBytes

send_transaction_munger(transaction)
Parameters:

transaction (TxParams)

Return type:

Tuple[TxParams]

set_contract_factory(contract_factory)
Parameters:

contract_factory (Type[AsyncContract | AsyncContractCaller])

Return type:

None

set_gas_price_strategy(gas_price_strategy)
Parameters:

gas_price_strategy (Callable[[Web3, TxParams], Wei] | Callable[[AsyncWeb3, TxParams], Wei] | None)

Return type:

None

async sign(account, data=None, hexstr=None, text=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • data (int | bytes | None)

  • hexstr (HexStr | None)

  • text (str | None)

Return type:

HexStr

sign_munger(account, data=None, hexstr=None, text=None)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • data (int | bytes | None)

  • hexstr (HexStr | None)

  • text (str | None)

Return type:

Tuple[Address | ChecksumAddress | ENS, HexStr]

async sign_transaction(transaction)
Parameters:

transaction (TxParams)

Return type:

SignedTx

async sign_typed_data(account, data)
Parameters:

  • account (Address | ChecksumAddress | ENS)

  • data (str)

Return type:

HexStr

async subscribe(subscription_type, subscription_arg=None)
Parameters:

  • subscription_type (Literal['newHeads', 'logs', 'newPendingTransactions', 'syncing'])

  • subscription_arg (LogsSubscriptionArg | bool | None)

Return type:

HexStr

async trace_filter(filter_params, decode_to=typing.List[typing.Union[evmspec.trace.call.Trace, evmspec.trace.create.Trace, evmspec.trace.reward.Trace, evmspec.trace.suicide.Trace]], decode_hook=<function _decode_hook>)
Parameters:
Return type:

T

async trace_transaction(transaction_hash)
Parameters:

transaction_hash (str)

Return type:

List[Trace | Trace | Trace | Trace]

async uninstall_filter(filter_id)
Parameters:

filter_id (HexStr)

Return type:

bool

async unsubscribe(subscription_id)
Parameters:

subscription_id (HexStr)

Return type:

bool

async wait_for_transaction_receipt(transaction_hash, timeout=120, poll_latency=0.1)
Parameters:
Return type:

TxReceipt

account = <eth_account.account.Account object>
property accounts: Tuple[ChecksumAddress]
property block_number: BlockNumber
chain_id
property codec: ABICodec
property coinbase: ChecksumAddress
property default_account: ChecksumAddress | Empty
property default_block: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 | HexStr | HexBytes | int
filter: Method[Callable[[str | FilterParams | HexStr | None], Awaitable[AsyncFilter]]]
property gas_price: Wei
get_block_number: Method[Callable[[], Awaitable[BlockNumber]]]

Bypasses web3py default result formatters.

get_block_transaction_count: Method[Callable[[BlockIdentifier], Awaitable[int]]]
get_filter_changes

Bypasses web3py default result formatters.

get_filter_logs

Bypasses web3py default result formatters.

property hashrate: int
is_async = True
property max_priority_fee: Wei

Try to use eth_maxPriorityFeePerGas but, since this is not part of the spec and is only supported by some clients, fall back to an eth_feeHistory calculation with min and max caps.

meth

Bypasses web3py default result formatters.

property mining: bool
property syncing: SyncStatus | bool
w3: AsyncWeb3
class dank_mids.eth.TraceFilterParams

Bases: TypedDict

__getitem__()

x.__getitem__(y) <==> x[y]

__init__(*args, **kwargs)
__iter__()

Implement iter(self).

clear() → None.  Remove all items from D.
copy() → a shallow copy of D
fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values
after: int
count: int
fromAddress: Sequence[Address | ChecksumAddress | ENS]
fromBlock: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 | HexStr | HexBytes | int
toAddress: Sequence[Address | ChecksumAddress | ENS]
toBlock: Literal['latest', 'earliest', 'pending', 'safe', 'finalized'] | BlockNumber | Hash32 | HexStr | HexBytes | int

dank_mids.exceptions module

exception dank_mids.exceptions.BrownieNotConnectedError

Bases: RuntimeError

RuntimeError raised when brownie is not connected but needs to be in order to access functionality within dank_mids.brownie_patch.

__init__(obj_name)

Initializes a new BrownieNotConnectedError.

Params:

obj_name: The name of the object the user attempted to access, which requires brownie to be connected.

Parameters:

obj_name (str)

exception dank_mids.exceptions.Revert

Bases: ValueError

dank_mids.middleware module

async dank_mids.middleware.dank_middleware(make_request, web3)

Create and return a DankMiddlewareController instance for an asynchronous Web3.

This function is used to create a middleware instance that can be added to an asynchronous Web3 object.

Parameters:

  • make_request (Callable[[RPCEndpoint, Any], Any]) – The next middleware in the chain or the actual request sender.

  • web3 (Web3) – The synchronous Web3 instance this middleware is attached to.

Returns:

An AsyncMiddleware function that can be used with synchronous Web3.

Return type:

Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]

See also

dank_mids.controller.DankMiddlewareController: The controller class that manages RPC requests sent thru the middleware.

dank_mids.semaphores module

class dank_mids.semaphores.BlockSemaphore

Bases: _AbstractPrioritySemaphore[str, _BlockSemaphoreContextManager]

A semaphore for managing concurrency based on block numbers.

This class extends _AbstractPrioritySemaphore to provide block-specific concurrency control.

Parameters:

  • value – The initial value of the semaphore.

  • name – An optional name for the semaphore.

See also

_BlockSemaphoreContextManager: The context manager used by this semaphore.

__call__(fn)

Convenient decorator method to wrap coroutine functions with the semaphore so you can rewrite this pattern:

``` semaphore = Semaphore(5)

async def limited():
async with semaphore:

return 1

```

like this:

``` semaphore = Semaphore(5)

@semaphore async def limited():

return 1

```

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

__getitem__(block)
Parameters:

block (int | str | Literal['latest', None])

Return type:

_BlockSemaphoreContextManager

__init__(value=1, *, name=None)

Initialize the semaphore with a given value and optional name for debugging.

Parameters:

  • value (int) – The initial value for the semaphore.

  • name (optional) – An optional name used only to provide useful context in debug logs.

Return type:

None

async acquire()

Acquire a semaphore.

If the internal counter is larger than zero on entry, decrement it by one and return True immediately. If it is zero on entry, block, waiting until some other coroutine has called release() to make it larger than 0, and then return True.

Return type:

Literal[True]

decorate(fn)

Wrap a coroutine function to ensure it runs with the semaphore.

Example

Now you can rewrite this pattern:

``` semaphore = Semaphore(5)

async def limited():
async with semaphore:

return 1

```

like this:

``` semaphore = Semaphore(5)

@semaphore async def limited():

return 1

```

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

locked()

Returns True if semaphore cannot be acquired immediately.

Return type:

bool

release()

Release a semaphore, incrementing the internal counter by one.

When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.

property debug_logs_enabled: bool

Checks if debug logging is enabled for the logger.

Returns:

True if debug logging is enabled, False otherwise.

Return type:

bool

property logger: Logger

Returns a logger instance specific to the class using this mixin.

The logger ID is constructed from the module and class name, and optionally includes an instance name if available.

Returns:

A logger instance for the class.

Return type:

Logger

name: str | None

dank_mids.stats module

stats.py

This module provides functionality for logging and collecting statistics related to the Dank Mids library. It includes custom logging levels, a specialized logger, and classes for collecting and writing statistics.

Key components: - Custom logging levels (STATS, DEVHINT) - _StatsLogger: A specialized logger for Dank Mids statistics - _Collector: Handles collection and computation of stats-related data - _Writer: Converts collected stats into human-readable format - _SentryExporter: Pushes metrics to Sentry

This module is crucial for debugging, performance monitoring, and optimization of the Dank Mids library.

dank_mids.stats.devhint(msg, *args, **kwargs)
Return type:

None

dank_mids.stats.log(msg, *args, **kwargs)
Return type:

None

dank_mids.stats.log_duration(work_descriptor, start, *, level=13)

Log the duration of a specific operation.

Parameters:

  • work_descriptor (str) – Description of the work being timed.

  • start (float) – Start time of the operation.

  • level (int) – Logging level to use. Defaults to STATS.

Return type:

None

dank_mids.stats.log_errd_batch(batch)

Log information about a failed JSON-RPC batch.

Parameters:

batch (JSONRPCBatch) – The failed batch to log.

Return type:

None

dank_mids.stats.COLLECT_STATS: bool = False

Flag to enable or disable stats collection.

dank_mids.stats.DEVHINT = 15

Custom logging level for developer hints, between STATS and INFO.

dank_mids.stats.STATS = 13

Custom logging level for statistics, between DEBUG and INFO.

dank_mids.types module

class dank_mids.types.BlockId

A type representing the identifier for a specific block in the blockchain.

alias of str

class dank_mids.types.Error

Bases: DictStruct

Represents an error in a JSON-RPC response.

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:

attr (str) – The name of the attribute to access.

Raises:

KeyError – If the provided key is not a member of the struct.

Returns:

The value of the attribute.

Return type:

Any

__iter__()

Iterate through the keys of the Struct.

Yields:

Struct key.

Return type:

Iterator[str]

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

  • key (str) – The key to look up.

  • default (optional) – The value to return if the key is not present.

Returns:

The value associated with the key, or the default value.

Return type:

Any

items()

Returns an iterator over the struct’s field name and value pairs.

Returns:

An iterator over the field name and value pairs.

Return type:

Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Returns:

An iterator over the field names.

Return type:

Iterator[str]

values()

Returns an iterator over the struct’s field values.

Returns:

An iterator over the field values.

Return type:

Iterator[Any]

code: int

The error code.

data: Any | None

Additional error data, if any.

message: str

The error message.

class dank_mids.types.OverrideParams

Bases: dict

A typed dictionary representing override parameters.

__getitem__()

x.__getitem__(y) <==> x[y]

__init__(*args, **kwargs)
__iter__()

Implement iter(self).

clear() → None.  Remove all items from D.
copy() → a shallow copy of D
fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values
code: str
class dank_mids.types.PartialRequest

Bases: DictStruct

Represents a partial JSON-RPC request.

While technially part of a request, we can successfully make requests to many nodes without including the jsonrpc field.

This class leaves off the jsonrpc field reduce encoding burden and web traffic.

This works with many but not all nodes.

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:

attr (str) – The name of the attribute to access.

Raises:

KeyError – If the provided key is not a member of the struct.

Returns:

The value of the attribute.

Return type:

Any

__iter__()

Iterate through the keys of the Struct.

Yields:

Struct key.

Return type:

Iterator[str]

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

  • key (str) – The key to look up.

  • default (optional) – The value to return if the key is not present.

Returns:

The value associated with the key, or the default value.

Return type:

Any

items()

Returns an iterator over the struct’s field name and value pairs.

Returns:

An iterator over the field name and value pairs.

Return type:

Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Returns:

An iterator over the field names.

Return type:

Iterator[str]

values()

Returns an iterator over the struct’s field values.

Returns:

An iterator over the field values.

Return type:

Iterator[Any]

property data: bytes
id: str | int

The request identifier.

method: str

The RPC method to be called.

params: list | None

The parameters for the RPC method call.

class dank_mids.types.PartialResponse

Bases: DictStruct

Represents a partial JSON-RPC response.

We use these to more efficiently decode responses from the node.

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:

attr (str) – The name of the attribute to access.

Raises:

KeyError – If the provided key is not a member of the struct.

Returns:

The value of the attribute.

Return type:

Any

__iter__()

Iterate through the keys of the Struct.

Yields:

Struct key.

Return type:

Iterator[str]

decode_result(method=None, *, caller=None)
Parameters:

method (RPCEndpoint | None)

Return type:

HexBytes | Wei | uint | ChainId | BlockNumber | AttributeDict

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

  • key (str) – The key to look up.

  • default (optional) – The value to return if the key is not present.

Returns:

The value associated with the key, or the default value.

Return type:

Any

items()

Returns an iterator over the struct’s field name and value pairs.

Returns:

An iterator over the field name and value pairs.

Return type:

Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Returns:

An iterator over the field names.

Return type:

Iterator[str]

to_dict(method=None)

Returns a complete dictionary representation of this response Struct.

Parameters:

method (RPCEndpoint | None)

Return type:

RPCResponse

values()

Returns an iterator over the struct’s field values.

Returns:

An iterator over the field values.

Return type:

Iterator[Any]

error: Error | None

The error object, if the call failed.

property exception: Exception

If the rpc response contains an ‘error’ field, returns a specialized exception for the specified rpc error.

property payload_too_large: bool
result: Raw

The result of the RPC call, if successful.

class dank_mids.types.RawResponse

Bases: object

Wraps a Raw object that we know represents a Response with a decode helper method. A RawResponse is a properly shaped response for one rpc call, received back from a jsonrpc batch request. They represent either a successful or a failed response, stored as pre-decoded bytes.

__init__(raw)
Parameters:

raw (Raw)

Return type:

None

decode(partial: Literal[True]) → PartialResponse
decode(partial: Literal[False] = False) → Response

Decode the wrapped Raw object into a Response or a PartialResponse.

class dank_mids.types.Request

Bases: PartialRequest

Represents a complete JSON-RPC request.

Inherits from PartialRequest and adds the JSON-RPC version.

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:

attr (str) – The name of the attribute to access.

Raises:

KeyError – If the provided key is not a member of the struct.

Returns:

The value of the attribute.

Return type:

Any

__iter__()

Iterate through the keys of the Struct.

Yields:

Struct key.

Return type:

Iterator[str]

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

  • key (str) – The key to look up.

  • default (optional) – The value to return if the key is not present.

Returns:

The value associated with the key, or the default value.

Return type:

Any

items()

Returns an iterator over the struct’s field name and value pairs.

Returns:

An iterator over the field name and value pairs.

Return type:

Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Returns:

An iterator over the field names.

Return type:

Iterator[str]

values()

Returns an iterator over the struct’s field values.

Returns:

An iterator over the field values.

Return type:

Iterator[Any]

property data: bytes
id: str | int

The request identifier.

jsonrpc: Literal['2.0']

The JSON-RPC version, always set to “2.0”.

method: str

The RPC method to be called.

params: list | None

The parameters for the RPC method call.

class dank_mids.types.Response

Bases: PartialResponse

Represents a complete JSON-RPC response.

Inherits from PartialResponse and adds the response identifier and JSON-RPC version.

__getitem__(attr)

Lookup an attribute value via dictionary-style access.

Parameters:

attr (str) – The name of the attribute to access.

Raises:

KeyError – If the provided key is not a member of the struct.

Returns:

The value of the attribute.

Return type:

Any

__iter__()

Iterate through the keys of the Struct.

Yields:

Struct key.

Return type:

Iterator[str]

decode_result(method=None, *, caller=None)
Parameters:

method (RPCEndpoint | None)

Return type:

HexBytes | Wei | uint | ChainId | BlockNumber | AttributeDict

get(key, default=None)

Get the value associated with a key, or a default value if the key is not present.

Parameters:

  • key (str) – The key to look up.

  • default (optional) – The value to return if the key is not present.

Returns:

The value associated with the key, or the default value.

Return type:

Any

items()

Returns an iterator over the struct’s field name and value pairs.

Returns:

An iterator over the field name and value pairs.

Return type:

Iterator[Tuple[str, Any]]

keys()

Returns an iterator over the field names of the struct.

Returns:

An iterator over the field names.

Return type:

Iterator[str]

to_dict(method=None)

Returns a complete dictionary representation of this response Struct.

Parameters:

method (RPCEndpoint | None)

Return type:

RPCResponse

values()

Returns an iterator over the struct’s field values.

Returns:

An iterator over the field values.

Return type:

Iterator[Any]

error: Error | None

The error object, if the call failed.

property exception: Exception

If the rpc response contains an ‘error’ field, returns a specialized exception for the specified rpc error.

id: str | int | None

The response identifier, matching the request.

jsonrpc: Literal['2.0']

The JSON-RPC version, always set to “2.0”.

property payload_too_large: bool
result: Raw

The result of the RPC call, if successful.

class dank_mids.types.eth_callParams

Bases: dict

A typed dictionary representing the parameters for an eth_call.

See also

web3.eth.Eth.call(): Web3’s method to perform a call without creating a transaction.

__getitem__()

x.__getitem__(y) <==> x[y]

__init__(*args, **kwargs)
__iter__()

Implement iter(self).

clear() → None.  Remove all items from D.
copy() → a shallow copy of D
fromkeys(value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values
data: str
to: ChecksumAddress
dank_mids.types.better_decode(data, *, type=None, dec_hook=None, method=None)
Parameters:
Return type:

T

dank_mids.types.AsyncMiddleware

A type alias for asynchronous middleware functions.

alias of Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]

dank_mids.types.BatchId

A type representing the identifier for a batch of operations, which can be either an integer or a string.

alias of int | str

dank_mids.types.JsonrpcParams

A list of parameters for JSON-RPC calls, which should be eth_callParams, BlockId, and OverrideParams, in that order.

alias of List[eth_callParams | BlockId | OverrideParams]

dank_mids.types.Multicalls

A dictionary mapping BlockId to Multicall objects.

See also

dank_mids._requests.Multicall: The Multicall class used in this mapping.

alias of Dict[BlockId, Multicall]

Module contents

class dank_mids.BlockSemaphore

Bases: _AbstractPrioritySemaphore[str, _BlockSemaphoreContextManager]

A semaphore for managing concurrency based on block numbers.

This class extends _AbstractPrioritySemaphore to provide block-specific concurrency control.

Parameters:

  • value – The initial value of the semaphore.

  • name – An optional name for the semaphore.

See also

_BlockSemaphoreContextManager: The context manager used by this semaphore.

__call__(fn)

Convenient decorator method to wrap coroutine functions with the semaphore so you can rewrite this pattern:

``` semaphore = Semaphore(5)

async def limited():
async with semaphore:

return 1

```

like this:

``` semaphore = Semaphore(5)

@semaphore async def limited():

return 1

```

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

__getitem__(block)
Parameters:

block (int | str | Literal['latest', None])

Return type:

_BlockSemaphoreContextManager

__init__(value=1, *, name=None)

Initialize the semaphore with a given value and optional name for debugging.

Parameters:

  • value (int) – The initial value for the semaphore.

  • name (optional) – An optional name used only to provide useful context in debug logs.

Return type:

None

async acquire()

Acquire a semaphore.

If the internal counter is larger than zero on entry, decrement it by one and return True immediately. If it is zero on entry, block, waiting until some other coroutine has called release() to make it larger than 0, and then return True.

Return type:

Literal[True]

decorate(fn)

Wrap a coroutine function to ensure it runs with the semaphore.

Example

Now you can rewrite this pattern:

``` semaphore = Semaphore(5)

async def limited():
async with semaphore:

return 1

```

like this:

``` semaphore = Semaphore(5)

@semaphore async def limited():

return 1

```

Parameters:

fn (Callable[[~P], Awaitable[T]])

Return type:

Callable[[~P], Awaitable[T]]

locked()

Returns True if semaphore cannot be acquired immediately.

Return type:

bool

release()

Release a semaphore, incrementing the internal counter by one.

When it was zero on entry and another coroutine is waiting for it to become larger than zero again, wake up that coroutine.

property debug_logs_enabled: bool

Checks if debug logging is enabled for the logger.

Returns:

True if debug logging is enabled, False otherwise.

Return type:

bool

property logger: Logger

Returns a logger instance specific to the class using this mixin.

The logger ID is constructed from the module and class name, and optionally includes an instance name if available.

Returns:

A logger instance for the class.

Return type:

Logger

name: str | None
class dank_mids.Contract

Bases: Contract

An extended version of brownie.Contract with additional functionality for Dank Mids.

This class provides lazy initialization of contract methods and supports asynchronous operations through Dank Mids middleware.

__init__(*args, **kwargs)

Initialize the Contract instance.

This method sets up lazy initialization for contract methods.

balance()

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

Return type:

Wei

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)

Create a new Contract instance from an ABI.

This method allows for the creation of a Contract instance using a provided ABI, which is useful when working with contracts that are not in the project’s build files and not verified on a block explorer.

Parameters:

  • name (str) – The name of the contract.

  • address (str) – The address of the deployed contract.

  • abi (List[dict]) – The ABI (Application Binary Interface) of the contract.

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

Contract

classmethod from_ethpm(name, manifest_uri, address=None, owner=None, persist=True)

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:

Contract

classmethod from_explorer(address, as_proxy_for=None, owner=None, silent=False, persist=True)

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:

Contract

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)

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

info()

Display NatSpec documentation for this contract.

Return type:

None

classmethod remove_deployment(address=None, alias=None)

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)

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

Parameters:

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

  • persist (bool)

Return type:

None

property abi: List
property alias: str | None
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.

async dank_mids.dank_middleware(make_request, web3)

Create and return a DankMiddlewareController instance for an asynchronous Web3.

This function is used to create a middleware instance that can be added to an asynchronous Web3 object.

Parameters:

  • make_request (Callable[[RPCEndpoint, Any], Any]) – The next middleware in the chain or the actual request sender.

  • web3 (Web3) – The synchronous Web3 instance this middleware is attached to.

Returns:

An AsyncMiddleware function that can be used with synchronous Web3.

Return type:

Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]

See also

dank_mids.controller.DankMiddlewareController: The controller class that manages RPC requests sent thru the middleware.

dank_mids.patch_contract(contract, w3=None)

Patch a contract with async and call batching functionalities.

Parameters:

  • contract (Contract | Contract | str) – The contract to patch.

  • w3 (DankWeb3 | None) – Optional DankWeb3 instance.

Returns:

The patched contract.

Return type:

Contract | Contract

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