Understanding ValueError in Production Systems
A ValueError in Python occurs when a function receives an argument with the correct type but an inappropriate value. In production environments, these errors can silently corrupt data pipelines, crash API endpoints, or cause cascading failures across microservices. Unlike syntax errors caught at compile time, ValueErrors emerge from runtime data conditions that developers often fail to anticipate during development.
The challenge with production ValueErrors is twofold: they typically surface under edge-case data inputs that weren't covered in test suites, and they can propagate through multiple service boundaries before manifesting, making root cause identification significantly harder than fixing the immediate exception.
What Exactly Triggers a ValueError?
The Python runtime raises a ValueError when an operation or function receives an argument with a valid type but an invalid value. Consider these common scenarios:
# Scenario 1: Invalid literal conversion
int("abc") # ValueError: invalid literal for int() with base 10
# Scenario 2: Mathematical domain error
import math
math.sqrt(-1) # ValueError: math domain error
# Scenario 3: Array operations with incompatible shapes
import numpy as np
np.array([1, 2]) + np.array([1, 2, 3]) # ValueError: operands could not be broadcast
# Scenario 4: Enum or categorical mismatch
from enum import Enum
class Status(Enum):
ACTIVE = "active"
INACTIVE = "inactive"
Status("deleted") # ValueError: 'deleted' is not a valid Status
# Scenario 5: Unpacking mismatch
a, b = [1, 2, 3] # ValueError: too many values to unpack (expected 2)In production, these triggers often hide behind layers of abstraction—a ValueError raised in a utility function three stack frames deep may only surface when a specific user submits malformed JSON to an API endpoint that nobody tested.
Why ValueErrors Are Particularly Dangerous in Production
Unlike TypeError or AttributeError, ValueErrors frequently indicate data integrity problems rather than coding mistakes. When your production system processes millions of records and a ValueError appears for 0.001% of them, the root cause could be:
- A silent schema change in an upstream data source
- Database corruption affecting a subset of rows
- A race condition producing invalid intermediate states
- Locale-specific formatting issues (dates, decimals) in user-submitted data
- Third-party API responses deviating from documented contracts
These errors demand a structured root cause analysis approach because the fix is rarely a simple try/except block—you need to understand how invalid data entered the system to prevent recurrence.
Root Cause Analysis Methodology for ValueErrors
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →Step 1: Capture the Complete Traceback with Context
Production systems must log not just the exception message but the full traceback and surrounding state. A bare except ValueError as e: print(e) discards the stack frames that reveal the data's origin. Use structured logging with traceback capture:
import traceback
import logging
import json
from datetime import datetime, timezone
logger = logging.getLogger("production_analyzer")
logger.setLevel(logging.ERROR)
# Configure JSON-structured logging for log aggregation systems
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)
def process_user_input(raw_input: str) -> int:
"""Convert user input to integer, logging full context on failure."""
try:
return int(raw_input)
except ValueError:
# Capture the full traceback as a string
tb_string = traceback.format_exc()
# Build a structured log entry with all available context
error_context = {
"event": "value_error_encountered",
"timestamp": datetime.now(timezone.utc).isoformat(),
"input_value": repr(raw_input),
"input_type": type(raw_input).__name__,
"input_length": len(raw_input) if raw_input else 0,
"traceback": tb_string,
"caller_function": "process_user_input",
"transaction_id": getattr(request_context, 'txn_id', None) # from middleware
}
logger.error(json.dumps(error_context, indent=2, default=str))
raise # Re-raise to let the global exception handler decide HTTP status
# Example of a global exception handler that enriches errors
def global_exception_handler(exc_type, exc_value, exc_tb):
"""Attach exception context to structured logs before crashing."""
tb_frames = traceback.extract_tb(exc_tb)
frame_info = []
for frame in tb_frames:
frame_info.append({
"filename": frame.filename,
"lineno": frame.lineno,
"function": frame.name,
"code_line": frame.line
})
crash_log = {
"event": "unhandled_exception",
"exception_type": exc_type.__name__,
"exception_message": str(exc_value),
"stack_frames": frame_info,
"process_id": os.getpid(),
"memory_mb": get_memory_usage() # custom helper
}
logger.critical(json.dumps(crash_log, indent=2, default=str))
# Install the global handler
import sys
sys.excepthook = global_exception_handlerThis approach ensures that when a ValueError occurs at 3 AM, your on-call engineer has immediate access to the exact input value, its type, its length, and the complete execution path—not just a cryptic "invalid literal" message.
Step 2: Reproduce the Error in an Isolated Environment
Reproduction is the cornerstone of root cause analysis. You need to reconstruct the exact conditions that triggered the ValueError without affecting production traffic. Build a minimal reproduction script that mirrors the production data path:
# reproduction_script.py
# Purpose: Isolate a ValueError from the payment processing pipeline
# Logged error: "ValueError: invalid literal for Decimal('')" in billing_service/charges.py:47
import sys
import json
from decimal import Decimal, InvalidOperation
from datetime import datetime
# Step 1: Extract the offending record from production logs
# Assume log aggregation identified this exact payload
offending_payload = {
"customer_id": "CUST-8912",
"amount": "", # <-- Empty string found in logs
"currency": "USD",
"timestamp": "2024-11-15T08:32:11Z",
"metadata": {
"source": "mobile_app_v3.2",
"session_id": "sess_9a3f1b"
}
}
# Step 2: Reconstruct the exact function that failed
def calculate_charge(amount_str: str, currency: str) -> Decimal:
"""Mirrors billing_service/charges.py line 47"""
# Original code assumed non-empty strings
amount = Decimal(amount_str) # ValueError here for empty string
exchange_rate = get_exchange_rate(currency, "USD")
return amount * exchange_rate
# Step 3: Test with the exact input
try:
result = calculate_charge(offending_payload["amount"], offending_payload["currency"])
print(f"Charge calculated: {result}")
except ValueError as e:
print(f"Reproduced! Error: {e}")
print(f"Input repr: {repr(offending_payload['amount'])}")
print(f"Input type: {type(offending_payload['amount'])}")
# Step 4: Check upstream data source
# Query the database for this transaction's raw data
print("\nInvestigating upstream source...")
# SELECT raw_amount FROM transactions WHERE session_id = 'sess_9a3f1b'
# Hypothetical result: raw_amount was NULL in DB, app serialized as empty stringWhen reproduction succeeds, you've confirmed the data pathway. When it fails—the same input doesn't trigger the error locally—suspect environmental differences: Python version, locale settings, library versions, or OS-level encoding defaults.
Step 3: Trace the Data Flow Backwards to the Source
The ValueError you see is the symptom. The root cause lives upstream where invalid data first entered the system. Perform a systematic backward trace:
# data_flow_tracer.py
# Trace the journey of data through service boundaries
import logging
from typing import Any, Optional, Dict
from datetime import datetime
logger = logging.getLogger("data_tracer")
class DataFlowTracer:
"""Wraps functions to record input/output at each pipeline stage."""
def __init__(self, trace_id: str):
self.trace_id = trace_id
self.stages: list[Dict[str, Any]] = []
def record_stage(self, stage_name: str, input_data: Any, output_data: Optional[Any] = None,
error: Optional[Exception] = None):
entry = {
"trace_id": self.trace_id,
"stage": stage_name,
"timestamp": datetime.now().isoformat(),
"input_repr": repr(input_data),
"input_type": type(input_data).__name__,
"output_repr": repr(output_data) if output_data else None,
"error": str(error) if error else None
}
self.stages.append(entry)
logger.info(json.dumps(entry, default=str))
return entry
def analyze(self):
"""After failure, reconstruct the data mutation chain."""
print(f"\n=== Data Flow Analysis for Trace: {self.trace_id} ===\n")
for i, stage in enumerate(self.stages):
print(f"Stage {i}: {stage['stage']}")
print(f" Input: {stage['input_repr']} (type: {stage['input_type']})")
if stage['output_repr']:
print(f" Output: {stage['output_repr']}")
if stage['error']:
print(f" ERROR: {stage['error']} <-- ValueError originated here")
print("\n=== End Trace ===\n")
# Usage example in a data pipeline
tracer = DataFlowTracer(trace_id="txn_CUST8912_20241115")
def api_gateway_parse(raw_body: bytes) -> Dict:
"""Stage 1: Parse JSON from HTTP request body."""
input_data = raw_body
try:
parsed = json.loads(raw_body.decode('utf-8'))
tracer.record_stage("api_gateway_parse", input_data, parsed)
return parsed
except json.JSONDecodeError as e:
tracer.record_stage("api_gateway_parse", input_data, error=e)
raise
def validation_layer_clean(parsed: Dict) -> Dict:
"""Stage 2: Validate and clean input fields."""
tracer.record_stage("validation_layer_clean", parsed)
# Bug: null amount becomes empty string instead of being rejected
if parsed.get("amount") is None:
parsed["amount"] = "" # <-- Root cause: null coercion to empty string
tracer.record_stage("validation_layer_clean_output", parsed, parsed)
return parsed
def billing_service_calculate(cleaned: Dict) -> Decimal:
"""Stage 3: Calculate final charge."""
tracer.record_stage("billing_service_calculate", cleaned)
try:
amount = Decimal(cleaned["amount"]) # ValueError: empty string
tracer.record_stage("billing_service_calculate_output", cleaned, amount)
return amount
except ValueError as e:
tracer.record_stage("billing_service_calculate", cleaned, error=e)
tracer.analyze() # Print the full data flow
raise
# Simulate the pipeline with a null amount
raw_request = b'{"customer_id": "CUST-8912", "amount": null, "currency": "USD"}'
try:
parsed_body = api_gateway_parse(raw_request)
cleaned_body = validation_layer_clean(parsed_body)
charge = billing_service_calculate(cleaned_body)
except ValueError:
print("Root cause: validation_layer_clean converted null to empty string")
print("Fix: Handle null explicitly - either reject or default to Decimal('0')")This tracer pattern reveals that the empty string didn't come from the user—it was manufactured by the validation layer when it encountered a JSON null. The real root cause is the null-to-empty-string coercion logic in validation_layer_clean, not the Decimal conversion.
Step 4: Classify the Root Cause and Implement the Fix
Once you've traced the data to its origin, classify the root cause to determine the appropriate fix strategy:
# root_cause_classification.py
# Different fix strategies based on root cause category
from decimal import Decimal, InvalidOperation
from typing import Optional, Union
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class RootCauseAnalysis:
category: str # One of: "schema_violation", "type_coercion_bug",
# "missing_validation", "environment_dependent",
# "race_condition", "data_corruption"
origin_stage: str # Which function introduced the invalid value
invalid_value: str # repr of the problematic value
recommendation: str # Concrete fix approach
# === Category 1: Schema Violation ===
# Root cause: upstream system changed data format without notice
def handle_schema_violation(raw_data: dict) -> Union[Decimal, None]:
"""
Fix: Implement schema validation at system boundaries.
Use Pydantic or JSON Schema to catch format deviations early.
"""
from pydantic import BaseModel, field_validator, ValidationError
class PaymentRequest(BaseModel):
amount: str
currency: str
@field_validator('amount')
@classmethod
def amount_must_be_parseable(cls, v: str) -> str:
if v is None or v.strip() == "":
raise ValueError("amount must be a non-empty numeric string")
try:
Decimal(v)
except InvalidOperation:
raise ValueError(f"amount '{v}' is not a valid decimal string")
return v
try:
validated = PaymentRequest(**raw_data)
return Decimal(validated.amount)
except ValidationError as e:
logger.error(f"Schema validation failed: {e}")
return None # Or raise HTTP 400
# === Category 2: Type Coercion Bug ===
# Root cause: null/None/empty values silently coerced to wrong type
def fix_null_coercion(raw_value: Optional[str]) -> Decimal:
"""
Fix: Explicit null handling with a clear default or rejection.
Never silently coerce None to empty string.
"""
if raw_value is None:
return Decimal("0") # Explicit default
if raw_value.strip() == "":
raise ValueError(f"Empty string not allowed for decimal conversion")
try:
return Decimal(raw_value)
except InvalidOperation:
raise ValueError(f"Cannot convert '{raw_value}' to Decimal")
# === Category 3: Missing Validation ===
# Root cause: no input validation before critical operation
def add_validation_gate(raw_value: str) -> Decimal:
"""
Fix: Insert validation gate before any destructive or type-casting operation.
Use a validation layer that rejects invalid data early with clear error messages.
"""
# Pre-validation: reject obviously invalid inputs
if not isinstance(raw_value, str):
raise TypeError(f"Expected string, got {type(raw_value).__name__}")
stripped = raw_value.strip()
if not stripped:
raise ValueError("Input string is empty or whitespace-only")
# Check for allowed characters before attempting conversion
allowed_chars = set("0123456789.-")
if not all(c in allowed_chars for c in stripped):
raise ValueError(f"Invalid characters in numeric string: {repr(raw_value)}")
# Now safe to convert
return Decimal(stripped)
# === Category 4: Environment-Dependent ValueError ===
# Root cause: locale-specific formatting (e.g., "1,234.56" vs "1.234,56")
def handle_locale_dependent_parsing(raw_value: str) -> Decimal:
"""
Fix: Normalize input to a canonical format before parsing.
Use locale-aware parsing or strip formatting characters.
"""
import re
from decimal import Decimal, InvalidOperation
# Detect and normalize common formats
# German locale: "1.234,56" -> remove thousands dots, replace comma decimal
if re.match(r'^[\d.]+,\d+$', raw_value):
normalized = raw_value.replace('.', '').replace(',', '.')
return Decimal(normalized)
# US/UK locale: "1,234.56" -> remove commas
if re.match(r'^[\d,]+\.\d+$', raw_value):
normalized = raw_value.replace(',', '')
return Decimal(normalized)
# Plain number
try:
return Decimal(raw_value)
except InvalidOperation:
raise ValueError(f"Cannot parse '{raw_value}' as decimal in any known locale")
# === Category 5: Race Condition ===
# Root cause: concurrent modification produces intermediate invalid state
def handle_race_condition_retry(shared_resource_key: str, max_retries: int = 3):
"""
Fix: Implement optimistic concurrency control with retry logic.
Detect stale data and re-fetch before operating.
"""
import time
for attempt in range(max_retries):
try:
# Fetch fresh value inside the retry loop
current_value = fetch_shared_resource(shared_resource_key)
result = process_value(current_value)
# Attempt atomic update
if atomic_compare_and_swap(shared_resource_key, current_value, result):
return result
except ValueError:
if attempt == max_retries - 1:
raise
time.sleep(0.1 * (2 ** attempt)) # Exponential backoff
raise ValueError(f"Failed to process {shared_resource_key} after {max_retries} attempts")Each root cause category demands a fundamentally different fix. Applying a generic try/except to the Decimal conversion site would mask schema violations and race conditions, allowing corrupt data to propagate silently through your system.
Building a Production-Ready ValueError Defense System
Defensive Input Validation Layer
The most effective way to prevent ValueErrors is to validate data at system boundaries before it enters your core logic. Build a centralized validation layer that rejects invalid data with actionable error responses:
# validation_layer.py
# Centralized input validation that catches ValueErrors at the boundary
from typing import Any, Callable, Dict, List, Optional, Union
from dataclasses import dataclass
from datetime import datetime
import re
import json
import logging
logger = logging.getLogger("validation_layer")
@dataclass
class ValidationResult:
is_valid: bool
normalized_value: Optional[Any] = None
error_message: Optional[str] = None
error_code: Optional[str] = None
class InputValidator:
"""Chainable validators that transform and validate input data."""
def __init__(self):
self.validators: List[Callable] = []
def add_validator(self, validator_fn: Callable[[Any], Any]):
"""Add a transformation/validation step to the chain."""
self.validators.append(validator_fn)
return self
def validate(self, raw_input: Any) -> ValidationResult:
"""Run all validators in sequence, short-circuit on first failure."""
current_value = raw_input
for validator in self.validators:
try:
current_value = validator(current_value)
except ValueError as e:
return ValidationResult(
is_valid=False,
error_message=str(e),
error_code="VALUE_ERROR"
)
except TypeError as e:
return ValidationResult(
is_valid=False,
error_message=str(e),
error_code="TYPE_ERROR"
)
return ValidationResult(
is_valid=True,
normalized_value=current_value
)
# Reusable validator functions
def validate_not_none(value: Any) -> Any:
"""Reject None values early."""
if value is None:
raise ValueError("Input cannot be None")
return value
def validate_not_empty(value: str) -> str:
"""Reject empty or whitespace-only strings."""
if isinstance(value, str) and value.strip() == "":
raise ValueError("Input string cannot be empty")
return value
def validate_numeric_string(value: str) -> str:
"""Validate that a string represents a parseable number."""
stripped = value.strip()
# Allow: integers, decimals, negative numbers
pattern = r'^-?\d+(\.\d+)?$'
if not re.match(pattern, stripped):
raise ValueError(f"'{value}' is not a valid numeric string")
return stripped
def validate_date_iso8601(value: str) -> datetime:
"""Parse ISO8601 date strings, rejecting invalid formats."""
from datetime import datetime
try:
return datetime.fromisoformat(value.replace('Z', '+00:00'))
except ValueError:
raise ValueError(f"'{value}' is not a valid ISO8601 date string")
# Compose validators for specific use cases
payment_amount_validator = (
InputValidator()
.add_validator(validate_not_none)
.add_validator(validate_not_empty)
.add_validator(validate_numeric_string)
)
date_validator = (
InputValidator()
.add_validator(validate_not_none)
.add_validator(validate_not_empty)
.add_validator(validate_date_iso8601)
)
# Usage in API endpoint
def process_payment(raw_request: Dict) -> Dict:
"""API endpoint that validates inputs before processing."""
amount_result = payment_amount_validator.validate(raw_request.get("amount"))
if not amount_result.is_valid:
return {
"status": "error",
"code": amount_result.error_code,
"message": f"Invalid amount: {amount_result.error_message}",
"field": "amount"
}
date_result = date_validator.validate(raw_request.get("effective_date"))
if not date_result.is_valid:
return {
"status": "error",
"code": date_result.error_code,
"message": f"Invalid date: {date_result.error_message}",
"field": "effective_date"
}
# At this point, all values are validated and normalized
amount_decimal = Decimal(amount_result.normalized_value)
effective_date = date_result.normalized_value
# Proceed with business logic safely
return process_validated_payment(amount_decimal, effective_date)This validation layer catches ValueErrors at the system boundary, converts them into structured error responses, and normalizes data before it reaches business logic. Users receive clear, field-specific error messages instead of cryptic 500 errors.
Monitoring and Alerting on ValueError Patterns
Even with robust validation, production ValueErrors will occur. Build monitoring that detects patterns and alerts on anomalies:
# monitoring_setup.py
# Production monitoring for ValueError patterns
import time
from collections import defaultdict, deque
from datetime import datetime, timedelta
from typing import Dict, List
import logging
logger = logging.getLogger("value_error_monitor")
class ValueErrorMonitor:
"""Tracks ValueError occurrences and triggers alerts on anomalies."""
def __init__(self, alert_threshold_per_minute: int = 10):
self.error_buffer: deque = deque(maxlen=10000) # Rolling window
self.threshold = alert_threshold_per_minute
self.error_counts_by_function: Dict[str, int] = defaultdict(int)
self.error_counts_by_input_pattern: Dict[str, int] = defaultdict(int)
self.last_alert_time: Dict[str, datetime] = {}
self.alert_cooldown = timedelta(minutes=5)
def record_value_error(self, function_name: str, input_value: str, traceback: str):
"""Record a ValueError occurrence with context."""
timestamp = datetime.now()
entry = {
"timestamp": timestamp,
"function": function_name,
"input_repr": repr(input_value),
"traceback_snippet": traceback[:500]
}
self.error_buffer.append(entry)
# Increment counters
self.error_counts_by_function[function_name] += 1
# Pattern detection: group by input characteristics
pattern_key = self._extract_pattern(input_value)
self.error_counts_by_input_pattern[pattern_key] += 1
# Check alert thresholds
self._check_alerts(function_name, pattern_key)
def _extract_pattern(self, value: str) -> str:
"""Extract a pattern signature from the input value."""
if value is None:
return "PATTERN:NULL"
if value == "":
return "PATTERN:EMPTY_STRING"
if value.strip() == "":
return "PATTERN:WHITESPACE_ONLY"
# Detect common problematic patterns
if isinstance(value, str):
if value.isdigit():
return "PATTERN:DIGITS_ONLY"
if all(c in "0123456789.,- " for c in value):
return "PATTERN:NUMERIC_LIKE"
if "\x00" in value:
return "PATTERN:NULL_BYTES"
return f"PATTERN:GENERIC_{type(value).__name__}"
def _check_alerts(self, function_name: str, pattern_key: str):
"""Check if alert thresholds are breached."""
now = datetime.now()
# Check function-specific spike
func_count = self.error_counts_by_function[function_name]
last_alert = self.last_alert_time.get(f"func_{function_name}")
if func_count >= self.threshold and (last_alert is None or now - last_alert > self.alert_cooldown):
self._trigger_alert(
f"ValueError spike in {function_name}",
f"Function {function_name} has generated {func_count} ValueErrors",
severity="WARNING"
)
self.last_alert_time[f"func_{function_name}"] = now
# Check pattern-specific spike
pattern_count = self.error_counts_by_input_pattern[pattern_key]
last_pattern_alert = self.last_alert_time.get(f"pattern_{pattern_key}")
if pattern_count >= self.threshold and (last_pattern_alert is None or now - last_pattern_alert > self.alert_cooldown):
self._trigger_alert(
f"ValueError pattern detected: {pattern_key}",
f"Input pattern {pattern_key} has caused {pattern_count} ValueErrors",
severity="CRITICAL"
)
self.last_alert_time[f"pattern_{pattern_key}"] = now
def _trigger_alert(self, title: str, description: str, severity: str):
"""Send alert to on-call system (PagerDuty, OpsGenie, Slack, etc.)."""
alert_payload = {
"title": title,
"description": description,
"severity": severity,
"timestamp": datetime.now().isoformat(),
"recent_errors": list(self.error_buffer)[-10:] # Last 10 for context
}
logger.critical(f"ALERT: {json.dumps(alert_payload, default=str)}")
# In production: send to alerting service API
# send_to_pagerduty(alert_payload)
# Global monitor instance
value_error_monitor = ValueErrorMonitor(alert_threshold_per_minute=10)
# Integration with existing error handlers
def monitored_function(raw_input: str) -> int:
"""Example function with ValueError monitoring."""
try:
return int(raw_input)
except ValueError as e:
value_error_monitor.record_value_error(
function_name="monitored_function",
input_value=raw_input,
traceback=traceback.format_exc()
)
raiseThis monitor transforms ValueErrors from silent log entries into actionable alerts. When a mobile app update starts sending empty strings for numeric fields, your on-call team knows within minutes, not after users complain.
Best Practices for Preventing ValueErrors in Production
1. Adopt Type Hints and Static Analysis
Type hints catch potential ValueError sources before deployment. Use mypy or pyright in CI/CD to enforce type correctness:
# Type hints reveal potential ValueError sources at analysis time
from typing import Optional, Union
from decimal import Decimal
# Before: unclear what types are acceptable
def calculate_discount(amount):
return amount * Decimal('0.1')
# After: explicit type contract catches mismatches in CI
def calculate_discount(amount: Decimal) -> Decimal:
"""Amount must be a pre-validated Decimal, not a raw string."""
if not isinstance(amount, Decimal):
raise TypeError(f"Expected Decimal, got {type(amount).__name__}")
return amount * Decimal('0.1')
# mypy will flag this call in CI before deployment:
# calculate_discount("100") # error: Argument 1 has incompatible type "str"2. Use Defensive Data Access Patterns
When reading from external sources (databases, APIs, message queues), never assume data validity. Implement read-time validation:
# defensive_data_access.py
# Validate data at read time, not just at write time
from decimal import Decimal, InvalidOperation
from typing import Optional, Dict, Any
import logging
logger = logging.getLogger(__name__)
class SafeDataReader:
"""Wraps database/API reads with validation to catch corrupt data early."""
@staticmethod
def safe_get_decimal(row: Dict[str, Any], field: str, default: Optional[Decimal] = None) -> Optional[Decimal]:
"""Safely extract a decimal value from a database row."""
raw_value = row.get(field)
if raw_value is None:
return default
if isinstance(raw_value, Decimal):
return raw_value
if isinstance(raw_value, (int, float)):
return Decimal(str(raw_value))
if isinstance(raw_value, str):
stripped = raw_value.strip()
if stripped == "":
logger.warning(f"Field '{field}' is empty string, returning default")
return default
try:
return Decimal(stripped)
except InvalidOperation:
logger.error(f"Field '{field}' has unparseable value: {repr(raw_value)}")
raise ValueError(f"Cannot convert field '{field}' value {repr(raw_value)} to Decimal")
raise TypeError(f"Field '{field}' has unexpected type {type(raw_value).__name__}")
@staticmethod
def safe_get_int(row: Dict[str, Any], field: str, default: Optional[int] = None) -> Optional[int]:
"""Safely extract an integer, handling common edge cases."""
raw_value = row.get(field)
if raw_value is None:
return default
if isinstance(raw_value, int) and not isinstance(raw_value, bool):
return raw_value
if isinstance(raw_value, float) and raw_value.is_integer():
return int(raw_value)
if isinstance(raw_value, str):
stripped = raw_value.strip()
if stripped == "":
return default
# Handle "1,234" formatted integers
cleaned = stripped.replace(',', '')
try:
return int(cleaned)
except ValueError:
logger.error(f"Field '{field}' has unparseable value: {repr(raw_value)}")
raise ValueError(f"Cannot convert field '{field}' value {repr(raw_value)} to int")
raise TypeError(f"Field '{field}' has unexpected type {type(raw_value).__name__}")
# Usage when reading from a database cursor
def process_transaction_row(row: Dict[str, Any]) -> None:
reader = SafeDataReader()
try:
amount = reader.safe_get_decimal(row, "amount")
quantity = reader.safe_get_int(row, "quantity", default=1)
# Business logic with validated values
total = amount * quantity
except (ValueError, TypeError) as e:
logger.error(f"Skipping corrupt row: {e}, row data: {row}")
# Move corrupt row to dead-letter queue for manual inspection
send_to_dead_letter_queue(row, str(e))3. Implement Circuit Breakers for Cascading ValueErrors
When a ValueError originates from an upstream dependency, prevent it from cascading through your entire system:
# circuit_breaker.py
# Prevent cascading failures from upstream ValueError sources
import time
import threading
from datetime import datetime, timedelta
from typing import Callable, Any
from functools import wraps
import logging
logger = logging.getLogger("circuit_breaker")
class CircuitBreaker:
"""Opens when upstream errors exceed threshold, preventing cascading failures."""
def __init__(self, name: str, failure_threshold: int = 5, recovery_timeout: int = 60):
self.name = name
self.failure_threshold = failure_threshold
self.recovery_timeout = timedelta(seconds=recovery_timeout)
self.failure_count = 0
self.last_failure_time: Optional[datetime] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
self.lock = threading.Lock()
def call(self, func: Callable, *args, **kwargs) -> Any:
"""Execute function with circuit breaker protection."""
with self.lock:
if self.state == "OPEN":
if self.last_failure_time and datetime.now() - self.last_failure_time > self.recovery_timeout:
self.state = "HALF_OPEN"
logger.info(f"Circuit {self.name} transitioning to HALF_OPEN")
else:
raise CircuitBreakerOpenError(
f"Circuit {self.name} is OPEN - upstream service returning invalid data"
)
try:
result = func(*args, **kwargs)
if self.state == "HALF_OPEN":
with self.lock:
self.state = "CLOSED"
self.failure_count = 0
logger.info(f"Circuit {self.name} closed successfully")
return result
except ValueError as e:
with self.lock:
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
logger.critical(
f"Circuit {self.name} OPENED after {self.failure_count} consecutive ValueErrors"
)
raise
class CircuitBreakerOpenError(Exception):
"""Raised when circuit breaker prevents execution."""
pass
# Usage with upstream data source
upstream_c