"""Defensive generator management with automatic fallbacks and error isolation."""
import logging
import traceback
from typing import Dict, List, Any, Optional, Callable, Type
from dataclasses import dataclass
from functools import wraps
import importlib.util
import sys
from .models import EntityType, ValidationResult
from .feature_flags import FeatureFlags
[docs]
@dataclass
class GeneratorResult:
"""Result from generator execution with error handling."""
data: List[Dict[str, Any]]
success: bool
generator_used: str # 'primary', 'fallback', 'simple'
errors: List[str]
warnings: List[str]
[docs]
@dataclass
class GeneratorInfo:
"""Information about a registered generator."""
generator_class: Type
fallback_class: Optional[Type]
simple_fallback: Callable
entity_type: EntityType
dependencies: List[str]
[docs]
class DefensiveGeneratorManager:
"""Manages generators with automatic fallbacks and error isolation."""
def __init__(self, feature_flags: Optional[FeatureFlags] = None):
self.feature_flags = feature_flags or FeatureFlags()
self.generators: Dict[str, GeneratorInfo] = {}
self.simple_fallbacks = self._load_basic_generators()
self.reference_data: Dict[str, List[Any]] = {}
self.logger = logging.getLogger(__name__)
# Error tracking for debugging
self.error_counts: Dict[str, int] = {}
self.performance_stats: Dict[str, List[float]] = {}
[docs]
def register_generator(self, entity_name: str, generator_class: Type,
fallback_class: Optional[Type] = None,
entity_type: EntityType = EntityType.UNKNOWN,
dependencies: List[str] = None) -> None:
"""Register a generator with optional fallback."""
simple_fallback = self._create_simple_fallback(entity_name, entity_type)
self.generators[entity_name] = GeneratorInfo(
generator_class=generator_class,
fallback_class=fallback_class,
simple_fallback=simple_fallback,
entity_type=entity_type,
dependencies=dependencies or []
)
[docs]
def generate_safely(self, entity_type: str, count: int, **kwargs) -> GeneratorResult:
"""Generate data with automatic fallback to simple versions on failure."""
start_time = self._get_time()
if entity_type not in self.generators:
return self._generate_unknown_entity(entity_type, count)
generator_info = self.generators[entity_type]
# Try primary generator first
if self.feature_flags.is_enabled('basic_generation'):
result = self._try_primary_generator(entity_type, generator_info, count, **kwargs)
if result.success:
self._record_performance(entity_type, 'primary', start_time)
return result
# Try fallback generator
if generator_info.fallback_class and self.feature_flags.is_enabled('simple_fallbacks'):
result = self._try_fallback_generator(entity_type, generator_info, count, **kwargs)
if result.success:
self._record_performance(entity_type, 'fallback', start_time)
return result
# Use simple fallback as last resort
result = self._use_simple_fallback(entity_type, generator_info, count)
self._record_performance(entity_type, 'simple', start_time)
return result
[docs]
def generate_with_dependencies(self, entity_type: str, count: int) -> GeneratorResult:
"""Generate data respecting dependency order."""
if entity_type not in self.generators:
return self._generate_unknown_entity(entity_type, count)
generator_info = self.generators[entity_type]
# Generate dependencies first
for dependency in generator_info.dependencies:
if dependency not in self.reference_data:
dep_result = self.generate_safely(dependency, count // 2) # Fewer dependencies
if dep_result.success:
self.reference_data[dependency] = [
record.get('id', f"dep_{i}") for i, record in enumerate(dep_result.data)
]
# Generate main entity with reference data
return self.generate_safely(entity_type, count, reference_data=self.reference_data)
[docs]
def batch_generate(self, entity_counts: Dict[str, int]) -> Dict[str, GeneratorResult]:
"""Generate multiple entities in dependency order."""
results = {}
# Sort by dependencies (simple topological sort)
sorted_entities = self._sort_by_dependencies(list(entity_counts.keys()))
for entity_type in sorted_entities:
count = entity_counts[entity_type]
result = self.generate_with_dependencies(entity_type, count)
results[entity_type] = result
# Store reference data for other entities
if result.success:
self.reference_data[entity_type] = [
record.get('id', f"{entity_type}_{i}") for i, record in enumerate(result.data)
]
return results
def _try_primary_generator(self, entity_type: str, generator_info: GeneratorInfo,
count: int, **kwargs) -> GeneratorResult:
"""Attempt to use primary generator with error isolation."""
try:
# Create generator instance
if 'reference_data' in kwargs:
generator = generator_info.generator_class(kwargs['reference_data'])
else:
generator = generator_info.generator_class()
# Validate generator has required methods
if not hasattr(generator, 'generate'):
raise AttributeError("Generator missing 'generate' method")
# Generate data with timeout protection
data = self._with_timeout(lambda: generator.generate(count), timeout_seconds=30)
# Basic validation
if not isinstance(data, list) or len(data) == 0:
raise ValueError("Generator returned invalid data")
return GeneratorResult(
data=data,
success=True,
generator_used='primary',
errors=[],
warnings=[]
)
except Exception as e:
error_msg = f"Primary generator failed for {entity_type}: {str(e)}"
self.logger.warning(error_msg)
self._increment_error_count(entity_type, 'primary')
return GeneratorResult(
data=[],
success=False,
generator_used='primary',
errors=[error_msg],
warnings=[]
)
def _try_fallback_generator(self, entity_type: str, generator_info: GeneratorInfo,
count: int, **kwargs) -> GeneratorResult:
"""Attempt to use fallback generator."""
try:
generator = generator_info.fallback_class()
data = generator.generate(count)
return GeneratorResult(
data=data,
success=True,
generator_used='fallback',
errors=[],
warnings=[f"Using fallback generator for {entity_type}"]
)
except Exception as e:
error_msg = f"Fallback generator failed for {entity_type}: {str(e)}"
self.logger.warning(error_msg)
self._increment_error_count(entity_type, 'fallback')
return GeneratorResult(
data=[],
success=False,
generator_used='fallback',
errors=[error_msg],
warnings=[]
)
def _use_simple_fallback(self, entity_type: str, generator_info: GeneratorInfo,
count: int) -> GeneratorResult:
"""Use simple fallback that always works."""
try:
data = generator_info.simple_fallback(count)
return GeneratorResult(
data=data,
success=True,
generator_used='simple',
errors=[],
warnings=[f"Using simple fallback for {entity_type}"]
)
except Exception as e:
# Even simple fallback failed - create minimal data
self.logger.error(f"Simple fallback failed for {entity_type}: {e}")
minimal_data = [{"id": i, "type": entity_type, "data": f"minimal_{i}"} for i in range(count)]
return GeneratorResult(
data=minimal_data,
success=True,
generator_used='minimal',
errors=[f"All generators failed for {entity_type}"],
warnings=["Using minimal data structure"]
)
def _generate_unknown_entity(self, entity_type: str, count: int) -> GeneratorResult:
"""Generate data for unregistered entity type."""
data = [
{
"id": i + 1,
"entity_type": entity_type,
"data": f"generated_{entity_type}_{i}",
"created_at": "2024-01-01T00:00:00Z"
}
for i in range(count)
]
return GeneratorResult(
data=data,
success=True,
generator_used='unknown_fallback',
errors=[],
warnings=[f"No generator registered for {entity_type}"]
)
def _create_simple_fallback(self, entity_name: str, entity_type: EntityType) -> Callable:
"""Create a simple fallback function for entity."""
def simple_fallback(count: int) -> List[Dict[str, Any]]:
base_data = {
EntityType.USER: lambda i: {
"id": i + 1,
"name": f"User_{i}",
"email": f"user_{i}@example.com"
},
EntityType.PRODUCT: lambda i: {
"id": i + 1,
"name": f"Product_{i}",
"price": 99.99
},
EntityType.ORDER: lambda i: {
"id": i + 1,
"total": 199.99,
"status": "pending"
}
}
generator_func = base_data.get(
entity_type,
lambda i: {"id": i + 1, "type": entity_name}
)
return [generator_func(i) for i in range(count)]
return simple_fallback
def _load_basic_generators(self) -> Dict[str, Callable]:
"""Load basic generators that always work."""
return {
'user': lambda count: [
{"id": i, "name": f"User_{i}", "email": f"user_{i}@test.com"}
for i in range(count)
],
'product': lambda count: [
{"id": i, "name": f"Product_{i}", "price": 10.0 + i}
for i in range(count)
],
'order': lambda count: [
{"id": i, "user_id": max(1, i % 10), "total": 50.0 + i}
for i in range(count)
]
}
def _sort_by_dependencies(self, entity_types: List[str]) -> List[str]:
"""Sort entity types by dependency order (with cycle detection)."""
sorted_types = []
visited = set()
in_stack = set()
def visit(entity_type: str):
if entity_type in visited or entity_type not in self.generators:
return
if entity_type in in_stack:
# Cycle detected - break it by skipping
self.logger.warning(f"Dependency cycle detected involving {entity_type}")
return
in_stack.add(entity_type)
# Visit dependencies first
for dep in self.generators[entity_type].dependencies:
visit(dep)
in_stack.discard(entity_type)
visited.add(entity_type)
sorted_types.append(entity_type)
for entity_type in entity_types:
visit(entity_type)
# Add unvisited types at the end
for entity_type in entity_types:
if entity_type not in sorted_types:
sorted_types.append(entity_type)
return sorted_types
def _with_timeout(self, func: Callable, timeout_seconds: int = 30) -> Any:
"""Execute function with timeout protection."""
# Simple timeout implementation - in production would use more sophisticated approach
try:
return func()
except Exception as e:
# For now, just re-raise - timeout handling would be added here
raise e
def _increment_error_count(self, entity_type: str, generator_type: str) -> None:
"""Track error counts for monitoring."""
key = f"{entity_type}_{generator_type}"
self.error_counts[key] = self.error_counts.get(key, 0) + 1
def _record_performance(self, entity_type: str, generator_type: str, start_time: float) -> None:
"""Record performance metrics."""
duration = self._get_time() - start_time
key = f"{entity_type}_{generator_type}"
if key not in self.performance_stats:
self.performance_stats[key] = []
self.performance_stats[key].append(duration)
# Keep only recent measurements
if len(self.performance_stats[key]) > 100:
self.performance_stats[key] = self.performance_stats[key][-50:]
def _get_time(self) -> float:
"""Get current time for performance measurement."""
import time
return time.time()
[docs]
def get_health_report(self) -> Dict[str, Any]:
"""Get health report showing error rates and performance."""
return {
'registered_generators': len(self.generators),
'error_counts': self.error_counts.copy(),
'performance_averages': {
key: sum(times) / len(times) if times else 0
for key, times in self.performance_stats.items()
},
'feature_flags': {
'enabled': list(self.feature_flags.enabled_features),
'disabled': list(self.feature_flags.disabled_features)
},
'reference_data_size': {
key: len(data) for key, data in self.reference_data.items()
}
}