compat

Documentation for eth_defi.compat Python module.

v6/v7 compatibility module

Functions

`add_middleware`(web3, middleware_func_or_name)	Add middleware with v6/v7 compatibility - FIXED VERSION
`check_if_retry_on_failure`(method)
`check_if_retry_on_failure_compat`(method)	v7 implementation of check_if_retry_on_failure
`check_if_retry_on_failure_v6`(method)	v6 implementation of check_if_retry_on_failure
`check_if_retry_on_failure_v7`(method)	v7 implementation of check_if_retry_on_failure
`clear_middleware`(web3_or_provider)	Clear all middleware with v6/v7 compatibility - handles both Web3 instances and providers
`construct_sign_and_send_raw_middleware`(...)	v7 wrapper for SignAndSendRawMiddlewareBuilder to maintain v6 compatibility
`create_http_provider`(args, *kwargs)	Web3 6/7 compatible HTTPProvider constructor.
`encode_abi_compat`(contract, fn_name, args)	Encode ABI with v6/v7 compatibility.
`encode_function_args`(func, args)	v7 implementation
`encode_function_args_v6`(func, args)	v6 implementation
`encode_function_args_v7`(func, args)	v7 implementation
`exception_retry_middleware`(make_request, ...)	v7 implementation of exception_retry_middleware - uses provider config when possible
`exception_retry_middleware_v6`(make_request, ...)	v6 implementation of exception_retry_middleware
`exception_retry_middleware_v7`(make_request, ...)	v7 implementation of exception_retry_middleware - uses provider config when possible
`get_function_info`(args, *kwargs)	v7 get_function_info equivalent - returns v6-compatible format
`get_function_info_v6`(args, *kwargs)	v6 get_function_info - handles multiple signatures
`get_function_info_v7`(args, *kwargs)	v7 get_function_info equivalent - returns v6-compatible format
`install_api_call_counter_middleware_compat`(web3)	Install API call counter middleware with v6/v7 compatibility
`install_poa_middleware`(web3[, layer])	Install POA middleware with v6/v7 compatibility
`install_retry_middleware_compat`(web3[, layer])	Install retry middleware with v6/v7 compatibility
`native_datetime_utc_fromtimestamp`(timestamp)	Convert timestamp to native UTC datetime object.
`native_datetime_utc_now`()	Get current UTC time as a native datetime object.

Classes

APICallCounterMiddleware

v7 API call counter middleware

class APICallCounterMiddleware

Bases: web3.middleware.base.Web3Middleware

v7 API call counter middleware

request_processor(method, params)

Process the request and count API calls

Parameters

method (web3.types.RPCEndpoint) –
params (Any) –

Return type

tuple[web3.types.RPCEndpoint, Any]

__init__(w3, counter)

Parameters: counter (collections.Counter) –

install_api_call_counter_middleware_compat(web3): Install API call counter middleware with v6/v7 compatibility

add_middleware(web3, middleware_func_or_name, layer=0)

Add middleware with v6/v7 compatibility - FIXED VERSION

Args:: web3: Web3 instance middleware_func_or_name: Either middleware function or string name layer: Layer to inject at (default 0)

check_if_retry_on_failure_v6(method): v6 implementation of check_if_retry_on_failure

check_if_retry_on_failure_v7(method): v7 implementation of check_if_retry_on_failure

exception_retry_middleware_v6(make_request, web3, retryable_exceptions, retryable_status_codes, retryable_rpc_error_codes, retries=10, sleep=5.0, backoff=1.6)

v6 implementation of exception_retry_middleware

Parameters

make_request (Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse]) –
web3 (web3.main.Web3) –
retries (int) –
sleep (float) –
backoff (float) –

Return type

Optional[Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse | None]]

exception_retry_middleware_v7(make_request, web3, retryable_exceptions, retryable_status_codes, retryable_rpc_error_codes, retries=10, sleep=5.0, backoff=1.6)

v7 implementation of exception_retry_middleware - uses provider config when possible

Parameters

make_request (Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse]) –
web3 (web3.main.Web3) –
retries (int) –
sleep (float) –
backoff (float) –

Return type

Optional[Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse | None]]

clear_middleware(web3_or_provider)

Clear all middleware with v6/v7 compatibility - handles both Web3 instances and providers

Parameters: web3_or_provider (web3.main.Web3 | web3.providers.rpc.rpc.HTTPProvider) –
Return type: None

install_poa_middleware(web3, layer=0): Install POA middleware with v6/v7 compatibility

install_retry_middleware_compat(web3, layer=0)

Install retry middleware with v6/v7 compatibility

Parameters

web3 (web3.providers.rpc.rpc.HTTPProvider) –
layer (int) –

encode_function_args_v6(func, args): v6 implementation

encode_function_args_v7(func, args): v7 implementation

get_function_info_v6(*args, **kwargs): v6 get_function_info - handles multiple signatures

get_function_info_v7(*args, **kwargs): v7 get_function_info equivalent - returns v6-compatible format

encode_abi_compat(contract, fn_name, args)

Encode ABI with v6/v7 compatibility.

In v6: contract.encodeABI(fn_name=”function_name”, args=[…]) In v7: contract.encode_abi(fn_name=”function_name”, args=[…])

Parameters

contract (web3.contract.contract.Contract) – Web3 contract instance
fn_name (str) – Function name to encode
args (list[Any]) – Arguments for the function

Returns

Encoded ABI string

Return type

eth_typing.encoding.HexStr

construct_sign_and_send_raw_middleware(private_key_or_account): v7 wrapper for SignAndSendRawMiddlewareBuilder to maintain v6 compatibility

encode_function_args(func, args): v7 implementation

get_function_info(*args, **kwargs): v7 get_function_info equivalent - returns v6-compatible format

exception_retry_middleware(make_request, web3, retryable_exceptions, retryable_status_codes, retryable_rpc_error_codes, retries=10, sleep=5.0, backoff=1.6)

v7 implementation of exception_retry_middleware - uses provider config when possible

Parameters

make_request (Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse]) –
web3 (web3.main.Web3) –
retries (int) –
sleep (float) –
backoff (float) –

Return type

Optional[Callable[[web3.types.RPCEndpoint, Any], web3.types.RPCResponse | None]]

check_if_retry_on_failure_compat(method): v7 implementation of check_if_retry_on_failure

native_datetime_utc_now()

Get current UTC time as a native datetime object.

Replacement for the deprecated datetime.datetime.utcnow(). Returns a native datetime object (no timezone info) representing UTC time.

This is optimized for blockchain contexts where: - All timestamps are assumed to be UTC - Timezone-aware objects add unnecessary overhead - native datetimes are sufficient and faster

Returns:: datetime.datetime: Native datetime object in UTC

Return type: datetime.datetime

native_datetime_utc_fromtimestamp(timestamp)

Convert timestamp to native UTC datetime object.

Replacement for the deprecated datetime.datetime.utcfromtimestamp(). Returns a native datetime object (no timezone info) representing UTC time.

This is optimized for blockchain contexts where: - All timestamps are assumed to be UTC - Timezone-aware objects add unnecessary overhead - native datetimes are sufficient and faster

Args:: timestamp (float): Unix timestamp (seconds since epoch)
Returns:: datetime.datetime: native datetime object in UTC

Parameters: timestamp (float) –
Return type: datetime.datetime

create_http_provider(*args, **kwargs)

Web3 6/7 compatible HTTPProvider constructor.

Example:

@pytest.fixture()
def provider_1(anvil):
    provider = create_http_provider(anvil.json_rpc_url, exception_retry_configuration=None)
    clear_middleware(provider)
    return provider

Return type: web3.providers.rpc.rpc.HTTPProvider