Common utility functions for python code that interacts with Ethereum
—
Enhanced logging capabilities with DEBUG2 level support and environment introspection tools for development and troubleshooting.
Get loggers with extended debugging capabilities.
def get_logger(name: str, logger_class=None) -> Logger:
"""
Get standard or custom logger instance.
Args:
name (str): Logger name (typically __name__)
logger_class: Optional logger class (defaults to standard Logger)
Returns:
Logger: Configured logger instance
"""
def get_extended_debug_logger(name: str) -> ExtendedDebugLogger:
"""
Get logger with DEBUG2 level support.
Args:
name (str): Logger name
Returns:
ExtendedDebugLogger: Logger with debug2() method
"""
def setup_DEBUG2_logging():
"""Configure DEBUG2 logging level in logging system."""Enhanced logger classes and mixins for automatic logger setup.
class ExtendedDebugLogger(logging.Logger):
"""Logger class with DEBUG2 level support."""
def debug2(self, message, *args, **kwargs):
"""Log at DEBUG2 level (more verbose than DEBUG)."""
class HasLogger:
"""Mixin class providing logger property."""
@property
def logger(self) -> Logger:
"""Get logger for this class."""
class HasExtendedDebugLogger:
"""Mixin class providing extended debug logger property."""
@property
def logger(self) -> ExtendedDebugLogger:
"""Get extended debug logger for this class."""
class HasLoggerMeta(type):
"""Metaclass for automatic logger setup."""
class HasExtendedDebugLoggerMeta(type):
"""Metaclass for automatic extended logger setup."""Environment introspection tools for troubleshooting.
def get_environment_summary() -> str:
"""
Get comprehensive environment information.
Returns:
str: Detailed environment summary including Python version,
platform info, installed packages, and system details
"""
def python_version() -> str:
"""Get Python version information."""
def platform_info() -> str:
"""Get detailed platform information."""
def pip_freeze() -> str:
"""Get pip freeze output for dependency tracking."""DEBUG2_LEVEL_NUM = 8 # Numeric value for DEBUG2 logging levelfrom eth_utils import get_logger, get_extended_debug_logger
# Standard logger
logger = get_logger(__name__)
logger.info("Application started")
logger.debug("Debug information")
# Extended debug logger
debug_logger = get_extended_debug_logger(__name__)
debug_logger.info("Application started")
debug_logger.debug("Standard debug info")
debug_logger.debug2("Very detailed debug info") # More verbose than debugfrom eth_utils import HasLogger, HasExtendedDebugLogger
class EthereumClient(HasLogger):
"""Ethereum client with automatic logging."""
def connect(self, url):
self.logger.info(f"Connecting to {url}")
try:
# Connection logic here
self.logger.info("Connected successfully")
except Exception as e:
self.logger.error(f"Connection failed: {e}")
class DetailedEthereumClient(HasExtendedDebugLogger):
"""Ethereum client with detailed debugging."""
def send_transaction(self, tx):
self.logger.info("Sending transaction")
self.logger.debug(f"Transaction data: {tx}")
self.logger.debug2(f"Raw transaction bytes: {tx.raw_data}")
# Send logic here
self.logger.debug2("Transaction sent, waiting for receipt")
# Usage
client = EthereumClient()
client.connect("https://mainnet.infura.io/v3/...")
detailed_client = DetailedEthereumClient()
detailed_client.send_transaction(some_transaction)from eth_utils import setup_DEBUG2_logging, get_extended_debug_logger
import logging
# Setup DEBUG2 level
setup_DEBUG2_logging()
# Configure logging level
logging.basicConfig(level=8) # DEBUG2_LEVEL_NUM
# Use extended logger
logger = get_extended_debug_logger(__name__)
logger.info("Info level message") # Always shown
logger.debug("Debug level message") # Shown if level <= DEBUG
logger.debug2("Debug2 level message") # Shown if level <= DEBUG2 (8)from eth_utils import get_environment_summary, python_version, platform_info, pip_freeze
def diagnose_environment():
"""Generate comprehensive environment diagnostic."""
print("=== Environment Diagnostic ===")
# Complete environment summary
print("Full Environment Summary:")
print(get_environment_summary())
print()
# Specific components
print(f"Python Version: {python_version()}")
print(f"Platform: {platform_info()}")
print("\nInstalled Packages:")
print(pip_freeze())
# Run diagnostic
diagnose_environment()from eth_utils import get_extended_debug_logger
import json
class TransactionProcessor(HasExtendedDebugLogger):
"""Transaction processor with comprehensive logging."""
def process_transaction(self, tx_hash):
self.logger.info(f"Processing transaction {tx_hash}")
try:
# Fetch transaction
self.logger.debug(f"Fetching transaction data for {tx_hash}")
tx_data = self.fetch_transaction(tx_hash)
self.logger.debug2(f"Raw transaction data: {json.dumps(tx_data, indent=2)}")
# Validate transaction
self.logger.debug("Validating transaction")
if not self.validate_transaction(tx_data):
self.logger.warning(f"Transaction validation failed: {tx_hash}")
return False
# Process transaction
self.logger.debug("Processing validated transaction")
result = self.execute_transaction(tx_data)
self.logger.debug2(f"Transaction execution result: {result}")
self.logger.info(f"Successfully processed transaction {tx_hash}")
return True
except Exception as e:
self.logger.error(f"Error processing transaction {tx_hash}: {e}", exc_info=True)
return False
def fetch_transaction(self, tx_hash):
"""Fetch transaction with detailed logging."""
self.logger.debug2(f"Making RPC call for transaction {tx_hash}")
# RPC call logic here
return {"hash": tx_hash, "value": "0x1"}
def validate_transaction(self, tx_data):
"""Validate transaction with debug logging."""
self.logger.debug2(f"Validating fields: {list(tx_data.keys())}")
# Validation logic here
return True
def execute_transaction(self, tx_data):
"""Execute transaction with detailed tracing."""
self.logger.debug2("Starting transaction execution")
# Execution logic here
return {"status": "success"}
# Usage
processor = TransactionProcessor()
processor.process_transaction("0x123...")from eth_utils import get_logger, get_extended_debug_logger, setup_DEBUG2_logging
import logging
def setup_application_logging(log_level="INFO", enable_debug2=False):
"""Setup application-wide logging configuration."""
# Configure basic logging
logging.basicConfig(
level=getattr(logging, log_level.upper()),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('application.log')
]
)
# Enable DEBUG2 if requested
if enable_debug2:
setup_DEBUG2_logging()
if log_level.upper() == "DEBUG2":
logging.getLogger().setLevel(8) # DEBUG2_LEVEL_NUM
# Create application loggers
app_logger = get_logger("application")
if enable_debug2:
debug_logger = get_extended_debug_logger("application.debug")
return app_logger, debug_logger
else:
return app_logger, None
# Setup logging
app_logger, debug_logger = setup_application_logging(log_level="DEBUG", enable_debug2=True)
app_logger.info("Application logging configured")
if debug_logger:
debug_logger.debug2("Extended debugging enabled")from eth_utils import get_logger, get_environment_summary
import traceback
class ErrorReporter(HasLogger):
"""Error reporting with environment context."""
def report_error(self, error, context=None):
"""Report error with full context."""
error_id = f"ERR_{hash(str(error)) % 10000:04d}"
self.logger.error(f"Error {error_id}: {error}")
if context:
self.logger.error(f"Error context: {context}")
# Log full traceback
self.logger.error(f"Traceback:\n{traceback.format_exc()}")
# Log environment for debugging
self.logger.debug(f"Environment at error time:\n{get_environment_summary()}")
return error_id
def report_transaction_error(self, tx_hash, error):
"""Report transaction-specific error."""
context = {
"transaction_hash": tx_hash,
"error_type": type(error).__name__,
"timestamp": "2024-01-01T00:00:00Z" # Would use actual timestamp
}
return self.report_error(error, context)
# Usage
reporter = ErrorReporter()
try:
# Some operation that might fail
process_transaction("0x123...")
except Exception as e:
error_id = reporter.report_transaction_error("0x123...", e)
print(f"Error reported with ID: {error_id}")from eth_utils import get_extended_debug_logger
import time
from functools import wraps
def log_performance(logger_name=None):
"""Decorator to log function performance."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
logger = get_extended_debug_logger(logger_name or func.__module__)
start_time = time.time()
logger.debug2(f"Starting {func.__name__} with args={args}, kwargs={kwargs}")
try:
result = func(*args, **kwargs)
execution_time = time.time() - start_time
logger.debug(f"{func.__name__} completed in {execution_time:.4f}s")
logger.debug2(f"{func.__name__} result: {result}")
return result
except Exception as e:
execution_time = time.time() - start_time
logger.error(f"{func.__name__} failed after {execution_time:.4f}s: {e}")
raise
return wrapper
return decorator
# Usage
class BlockchainAnalyzer(HasExtendedDebugLogger):
@log_performance("blockchain.analyzer")
def analyze_transaction(self, tx_hash):
"""Analyze transaction with performance logging."""
self.logger.info(f"Analyzing transaction {tx_hash}")
# Simulate analysis work
time.sleep(0.1)
return {"status": "analyzed", "gas_used": 21000}
# Usage
analyzer = BlockchainAnalyzer()
result = analyzer.analyze_transaction("0x123...")