tessl/pypi-web3

A Python library for interacting with Ethereum blockchain

Overall
score

88%

Evaluation — 88%

↑ 1.01x

Agent success when using this tile

Overview

Eval results

Files

Middleware

Name: tessl/pypi-web3
Author: tessl

Request and response processing middleware for customizing Web3 behavior including validation, formatting, signing, filtering, and gas price strategies with composable middleware stack management.

Capabilities

Base Middleware Classes

Core middleware infrastructure and base classes.

class Web3Middleware:
    def __init__(self, w3: Union[Web3, AsyncWeb3]):
        """
        Initialize middleware with Web3 instance.
        
        Parameters:
        - w3: Web3 or AsyncWeb3 instance
        """

    def wrap_make_request(self, make_request: MakeRequestFn) -> MakeRequestFn:
        """
        Wrap request function with middleware logic.
        
        Parameters:
        - make_request: Original request function
        
        Returns:
        Wrapped request function
        """

    async def async_wrap_make_request(
        self, 
        make_request: AsyncMakeRequestFn
    ) -> AsyncMakeRequestFn:
        """
        Wrap async request function with middleware logic.
        
        Parameters:
        - make_request: Original async request function
        
        Returns:
        Wrapped async request function
        """

Middleware = Callable[[Union[Web3, AsyncWeb3]], Web3Middleware]

Built-in Middleware

Pre-built middleware for common functionality.

class AttributeDictMiddleware(Web3Middleware):
    """Convert response objects to attribute dictionaries."""

class BufferedGasEstimateMiddleware(Web3Middleware):
    """Buffer gas estimates to avoid repeated calls."""

class LocalFilterMiddleware(Web3Middleware):
    """Handle event filters locally instead of on node."""

class GasPriceStrategyMiddleware(Web3Middleware):
    """Apply gas price strategy to transactions."""

class ENSNameToAddressMiddleware(Web3Middleware):
    """Resolve ENS names to addresses in transactions."""

class ExtraDataToPOAMiddleware(Web3Middleware):
    """Handle Proof of Authority consensus extra data."""

class PythonicMiddleware(Web3Middleware):
    """Convert responses to Pythonic formats."""

class ValidationMiddleware(Web3Middleware):
    """Validate request parameters."""

Middleware Builders

Factory functions for creating configurable middleware.

def FormattingMiddlewareBuilder() -> Middleware:
    """
    Create response formatting middleware.
    
    Returns:
    Configured formatting middleware
    """

def SignAndSendRawMiddlewareBuilder(
    private_key_or_account: Union[PrivateKey, LocalAccount]
) -> Middleware:
    """
    Create transaction signing middleware.
    
    Parameters:
    - private_key_or_account: Private key or account for signing
    
    Returns:
    Configured signing middleware
    """

def StalecheckMiddlewareBuilder(
    allowable_delay: int = 60
) -> Middleware:
    """
    Create stale block check middleware.
    
    Parameters:
    - allowable_delay: Maximum block age in seconds
    
    Returns:
    Configured stalecheck middleware
    """

Middleware Management

Functions for combining and managing middleware stacks.

def combine_middleware(
    middleware: Sequence[Middleware],
    w3: Web3,
    provider_request_fn: MakeRequestFn
) -> Callable[..., RPCResponse]:
    """
    Combine middleware into request processing pipeline.
    
    Parameters:
    - middleware: List of middleware functions
    - w3: Web3 instance
    - provider_request_fn: Provider's request function
    
    Returns:
    Combined request processor
    """

async def async_combine_middleware(
    middleware: Sequence[Middleware],
    async_w3: AsyncWeb3,
    provider_request_fn: AsyncMakeRequestFn
) -> Callable[..., Coroutine[Any, Any, RPCResponse]]:
    """
    Combine async middleware into request processing pipeline.
    
    Parameters:
    - middleware: List of middleware functions
    - async_w3: AsyncWeb3 instance
    - provider_request_fn: Provider's async request function
    
    Returns:
    Combined async request processor
    """

Middleware Onion

Middleware stack management interface.

class MiddlewareOnion:
    def add(self, middleware: Middleware, name: Optional[str] = None) -> None:
        """
        Add middleware to the top of stack.
        
        Parameters:
        - middleware: Middleware function
        - name: Optional middleware name
        """

    def inject(
        self,
        middleware: Middleware,
        name: Optional[str] = None,
        layer: Optional[int] = None
    ) -> None:
        """
        Inject middleware at specific layer.
        
        Parameters:
        - middleware: Middleware function
        - name: Optional middleware name
        - layer: Stack position (0 = top)
        """

    def remove(self, name: str) -> None:
        """
        Remove middleware by name.
        
        Parameters:
        - name: Middleware name to remove
        """

    def replace(self, name: str, middleware: Middleware) -> None:
        """
        Replace existing middleware.
        
        Parameters:
        - name: Name of middleware to replace
        - middleware: New middleware function
        """

    def clear(self) -> None:
        """Remove all middleware from stack."""

Types

Middleware-related type definitions.

MakeRequestFn = Callable[[RPCEndpoint, Any], RPCResponse]
AsyncMakeRequestFn = Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]

class RPCResponse(TypedDict):
    id: int
    jsonrpc: str
    result: Any
    error: NotRequired[Dict[str, Any]]

RPCEndpoint = NewType('RPCEndpoint', str)

Usage Examples

Basic Middleware Usage

from web3 import Web3
from web3.middleware import geth_poa_middleware

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Add Proof of Authority middleware
w3.middleware_onion.inject(geth_poa_middleware, layer=0)

# Verify middleware is active
print(f"Middleware stack: {[name for name, _ in w3.middleware_onion]}")

Custom Middleware

from web3 import Web3
from web3.types import RPCEndpoint, RPCResponse

def request_logging_middleware(make_request, w3):
    """Log all RPC requests."""
    def middleware(method: RPCEndpoint, params):
        print(f"Making request: {method} with params: {params}")
        response = make_request(method, params)
        print(f"Response: {response}")
        return response
    return middleware

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
w3.middleware_onion.add(request_logging_middleware, 'logging')

# All requests will now be logged
balance = w3.eth.get_balance('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8')

Transaction Signing Middleware

from web3 import Web3
from web3.middleware import SignAndSendRawMiddlewareBuilder
from eth_account import Account

# Create account for signing
account = Account.create()
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Add signing middleware
signing_middleware = SignAndSendRawMiddlewareBuilder(account)
w3.middleware_onion.add(signing_middleware, 'signing')

# Transactions will be automatically signed
tx_hash = w3.eth.send_transaction({
    'from': account.address,
    'to': '0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8',
    'value': w3.to_wei(1, 'ether'),
    'gas': 21000
})

print(f"Signed transaction hash: {tx_hash.hex()}")

Gas Price Strategy Middleware

from web3 import Web3
from web3.gas_strategies.time_based import construct_time_based_gas_price_strategy

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Create time-based gas price strategy
gas_strategy = construct_time_based_gas_price_strategy(
    max_wait_seconds=60,
    sample_size=5
)

# Set gas price strategy
w3.eth.set_gas_price_strategy(gas_strategy)

# Gas prices will be determined by strategy
transaction = {
    'from': w3.eth.accounts[0],
    'to': '0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8',
    'value': w3.to_wei(1, 'ether'),
    'gas': 21000
    # gasPrice automatically set by strategy
}

tx_hash = w3.eth.send_transaction(transaction)

Stale Block Check Middleware

from web3 import Web3
from web3.middleware import StalecheckMiddlewareBuilder

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Add stale block check (fail if block is >60 seconds old)
stalecheck_middleware = StalecheckMiddlewareBuilder(allowable_delay=60)
w3.middleware_onion.add(stalecheck_middleware, 'stalecheck')

# Requests will fail if connected to stale node
try:
    balance = w3.eth.get_balance('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8')
    print(f"Balance: {balance}")
except Exception as e:
    print(f"Node appears to be stale: {e}")

Response Formatting Middleware

from web3 import Web3
from web3.middleware import FormattingMiddlewareBuilder

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# Add response formatting
formatting_middleware = FormattingMiddlewareBuilder()
w3.middleware_onion.add(formatting_middleware, 'formatting')

# Responses will be formatted for Python consumption
block = w3.eth.get_block('latest')
print(f"Block timestamp: {block.timestamp}")  # Properly formatted

Middleware Stack Management

from web3 import Web3

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))

# View current middleware stack
print("Current middleware:")
for name, middleware in w3.middleware_onion:
    print(f"  {name}: {middleware}")

# Remove specific middleware
w3.middleware_onion.remove('validation')

# Replace middleware
new_validation = lambda make_request, w3: make_request
w3.middleware_onion.replace('validation', new_validation)

# Clear all middleware (be careful!)
# w3.middleware_onion.clear()

Async Middleware

import asyncio
from web3 import AsyncWeb3

async def async_logging_middleware(make_request, async_w3):
    """Async logging middleware."""
    async def middleware(method, params):
        print(f"Async request: {method}")
        response = await make_request(method, params)
        print(f"Async response received")
        return response
    return middleware

async def main():
    w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider('http://localhost:8545'))
    w3.middleware_onion.add(async_logging_middleware, 'async_logging')
    
    # Middleware will process async requests
    balance = await w3.eth.get_balance('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8')
    print(f"Balance: {balance}")

asyncio.run(main())

Error Handling Middleware

from web3 import Web3
from web3.exceptions import Web3Exception

def error_handling_middleware(make_request, w3):
    """Add custom error handling."""
    def middleware(method, params):
        try:
            return make_request(method, params)
        except Exception as e:
            print(f"Request failed: {method} - {e}")
            # Could implement retry logic, fallback providers, etc.
            raise Web3Exception(f"Enhanced error info: {e}")
    return middleware

w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
w3.middleware_onion.add(error_handling_middleware, 'error_handling')

Install with Tessl CLI

npx tessl i tessl/pypi-web3