docs: add CLI modularization analysis and common error handling utilities
Some checks failed
CLI Tests / test-cli (push) Failing after 12s
Cross-Node Transaction Testing / transaction-test (push) Successful in 4s
Multi-Node Stress Testing / stress-test (push) Has been cancelled
Deploy to Testnet / deploy-testnet (push) Has been cancelled
Node Failover Simulation / failover-test (push) Has been cancelled
Documentation Validation / validate-docs (push) Failing after 11s
Documentation Validation / validate-policies-strict (push) Successful in 4s
Security Scanning / security-scan (push) Successful in 31s
Some checks failed
CLI Tests / test-cli (push) Failing after 12s
Cross-Node Transaction Testing / transaction-test (push) Successful in 4s
Multi-Node Stress Testing / stress-test (push) Has been cancelled
Deploy to Testnet / deploy-testnet (push) Has been cancelled
Node Failover Simulation / failover-test (push) Has been cancelled
Documentation Validation / validate-docs (push) Failing after 11s
Documentation Validation / validate-policies-strict (push) Successful in 4s
Security Scanning / security-scan (push) Successful in 31s
- Analyze blockchain.py (1,388 lines) and agent.py (793 lines) structure - Correct size estimates from analysis (not 55k/26k lines) - Both files already well-organized with logical command groupings - Create common error handling utilities module for CLI - Add CLIError base class and specialized exceptions - Add handle_cli_error and handle_async_cli_error decorators - Add validation utilities for URLs, addresses, and required fields
This commit is contained in:
181
cli/utils/error_handling.py
Normal file
181
cli/utils/error_handling.py
Normal file
@@ -0,0 +1,181 @@
|
||||
"""
|
||||
Common error handling utilities for AITBC CLI
|
||||
Provides standardized error handling patterns and utilities for CLI commands
|
||||
"""
|
||||
|
||||
import sys
|
||||
from typing import Optional, Callable, Any
|
||||
from functools import wraps
|
||||
from . import error, warning, info
|
||||
|
||||
|
||||
class CLIError(Exception):
|
||||
"""Base exception for CLI errors"""
|
||||
def __init__(self, message: str, exit_code: int = 1):
|
||||
self.message = message
|
||||
self.exit_code = exit_code
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class NetworkError(CLIError):
|
||||
"""Network-related errors"""
|
||||
def __init__(self, message: str):
|
||||
super().__init__(f"Network error: {message}", exit_code=2)
|
||||
|
||||
|
||||
class ConfigurationError(CLIError):
|
||||
"""Configuration-related errors"""
|
||||
def __init__(self, message: str):
|
||||
super().__init__(f"Configuration error: {message}", exit_code=3)
|
||||
|
||||
|
||||
class ValidationError(CLIError):
|
||||
"""Validation errors for user input"""
|
||||
def __init__(self, message: str):
|
||||
super().__init__(f"Validation error: {message}", exit_code=4)
|
||||
|
||||
|
||||
class APIError(CLIError):
|
||||
"""API-related errors"""
|
||||
def __init__(self, message: str, status_code: Optional[int] = None):
|
||||
msg = f"API error: {message}"
|
||||
if status_code:
|
||||
msg += f" (HTTP {status_code})"
|
||||
super().__init__(msg, exit_code=5)
|
||||
|
||||
|
||||
def handle_cli_error(func: Callable) -> Callable:
|
||||
"""
|
||||
Decorator to standardize error handling in CLI commands.
|
||||
|
||||
Catches common exceptions and displays user-friendly error messages.
|
||||
|
||||
Args:
|
||||
func: Function to wrap with error handling
|
||||
|
||||
Returns:
|
||||
Wrapped function with standardized error handling
|
||||
"""
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except CLIError as e:
|
||||
error(e.message)
|
||||
sys.exit(e.exit_code)
|
||||
except KeyboardInterrupt:
|
||||
warning("\nOperation cancelled by user")
|
||||
sys.exit(130)
|
||||
except Exception as e:
|
||||
error(f"Unexpected error: {e}")
|
||||
sys.exit(1)
|
||||
return wrapper
|
||||
|
||||
|
||||
def handle_async_cli_error(func: Callable) -> Callable:
|
||||
"""
|
||||
Decorator to standardize error handling in async CLI commands.
|
||||
|
||||
Args:
|
||||
func: Async function to wrap with error handling
|
||||
|
||||
Returns:
|
||||
Wrapped async function with standardized error handling
|
||||
"""
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return await func(*args, **kwargs)
|
||||
except CLIError as e:
|
||||
error(e.message)
|
||||
sys.exit(e.exit_code)
|
||||
except KeyboardInterrupt:
|
||||
warning("\nOperation cancelled by user")
|
||||
sys.exit(130)
|
||||
except Exception as e:
|
||||
error(f"Unexpected error: {e}")
|
||||
sys.exit(1)
|
||||
return wrapper
|
||||
|
||||
|
||||
def safe_execute(
|
||||
operation: Callable,
|
||||
error_message: str = "Operation failed",
|
||||
default_return: Any = None,
|
||||
raise_on_error: bool = False
|
||||
) -> Any:
|
||||
"""
|
||||
Safely execute an operation with standardized error handling.
|
||||
|
||||
Args:
|
||||
operation: Function to execute
|
||||
error_message: Custom error message prefix
|
||||
default_return: Value to return on error (if not raising)
|
||||
raise_on_error: Whether to raise exception on error
|
||||
|
||||
Returns:
|
||||
Operation result or default_return on error
|
||||
|
||||
Raises:
|
||||
Exception: If raise_on_error is True and operation fails
|
||||
"""
|
||||
try:
|
||||
return operation()
|
||||
except Exception as e:
|
||||
if raise_on_error:
|
||||
raise
|
||||
error(f"{error_message}: {e}")
|
||||
return default_return
|
||||
|
||||
|
||||
def validate_required_fields(data: dict, required_fields: list) -> None:
|
||||
"""
|
||||
Validate that required fields are present in data dictionary.
|
||||
|
||||
Args:
|
||||
data: Dictionary to validate
|
||||
required_fields: List of required field names
|
||||
|
||||
Raises:
|
||||
ValidationError: If any required field is missing
|
||||
"""
|
||||
missing_fields = [field for field in required_fields if field not in data or data[field] is None]
|
||||
if missing_fields:
|
||||
raise ValidationError(f"Missing required fields: {', '.join(missing_fields)}")
|
||||
|
||||
|
||||
def validate_url(url: str) -> bool:
|
||||
"""
|
||||
Validate URL format.
|
||||
|
||||
Args:
|
||||
url: URL string to validate
|
||||
|
||||
Returns:
|
||||
True if valid, False otherwise
|
||||
"""
|
||||
import re
|
||||
url_pattern = re.compile(
|
||||
r'^https?://' # http:// or https://
|
||||
r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # domain...
|
||||
r'localhost|' # localhost...
|
||||
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
|
||||
r'(?::\d+)?' # optional port
|
||||
r'(?:/?|[/?]\S+)$', re.IGNORECASE)
|
||||
return bool(url_pattern.match(url))
|
||||
|
||||
|
||||
def validate_address(address: str) -> bool:
|
||||
"""
|
||||
Validate Ethereum address format.
|
||||
|
||||
Args:
|
||||
address: Ethereum address string
|
||||
|
||||
Returns:
|
||||
True if valid, False otherwise
|
||||
"""
|
||||
import re
|
||||
# Basic Ethereum address validation (0x followed by 40 hex characters)
|
||||
address_pattern = re.compile(r'^0x[a-fA-F0-9]{40}$')
|
||||
return bool(address_pattern.match(address))
|
||||
Reference in New Issue
Block a user