Spaces:
Sleeping
Sleeping
File size: 4,687 Bytes
2929135 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# src/utils/validators.py
from typing import Dict, Any, List, Optional
from datetime import datetime
from .logger import setup_logger
from .error_handlers import ValidationError
logger = setup_logger(__name__)
class Validator:
@staticmethod
def validate_state(state: Dict[str, Any]) -> bool:
"""Validate the state structure and data types"""
required_keys = ["messages", "current_task", "metrics", "timestamp"]
try:
# Check required keys
for key in required_keys:
if key not in state:
raise ValidationError(
message=f"Missing required key: {key}",
error_code="INVALID_STATE_STRUCTURE"
)
# Validate timestamp
if not isinstance(state["timestamp"], datetime):
raise ValidationError(
message="Invalid timestamp format",
error_code="INVALID_TIMESTAMP"
)
return True
except Exception as e:
logger.error(f"State validation failed: {str(e)}")
raise
@staticmethod
def validate_metrics(metrics: Dict[str, Any]) -> bool:
"""Validate metrics data structure and values"""
required_categories = [
"patient_flow",
"resources",
"quality",
"staffing"
]
try:
# Check required categories
for category in required_categories:
if category not in metrics:
raise ValidationError(
message=f"Missing required metrics category: {category}",
error_code="INVALID_METRICS_STRUCTURE"
)
# Validate numeric values
Validator._validate_numeric_values(metrics)
return True
except Exception as e:
logger.error(f"Metrics validation failed: {str(e)}")
raise
@staticmethod
def validate_tool_input(
tool_name: str,
params: Dict[str, Any],
required_params: List[str]
) -> bool:
"""Validate input parameters for tools"""
try:
# Check required parameters
for param in required_params:
if param not in params:
raise ValidationError(
message=f"Missing required parameter: {param}",
error_code="MISSING_PARAMETER",
details={"tool": tool_name, "parameter": param}
)
return True
except Exception as e:
logger.error(f"Tool input validation failed: {str(e)}")
raise
@staticmethod
def validate_department_data(department_data: Dict[str, Any]) -> bool:
"""Validate department-specific data"""
required_fields = [
"capacity",
"current_occupancy",
"staff_count"
]
try:
# Check required fields
for field in required_fields:
if field not in department_data:
raise ValidationError(
message=f"Missing required field: {field}",
error_code="INVALID_DEPARTMENT_DATA"
)
# Validate capacity constraints
if department_data["current_occupancy"] > department_data["capacity"]:
raise ValidationError(
message="Current occupancy exceeds capacity",
error_code="INVALID_OCCUPANCY"
)
return True
except Exception as e:
logger.error(f"Department data validation failed: {str(e)}")
raise
@staticmethod
def _validate_numeric_values(data: Dict[str, Any], path: str = "") -> None:
"""Recursively validate numeric values in nested dictionary"""
for key, value in data.items():
current_path = f"{path}.{key}" if path else key
if isinstance(value, (int, float)):
if value < 0:
raise ValidationError(
message=f"Negative value not allowed: {current_path}",
error_code="INVALID_NUMERIC_VALUE"
)
elif isinstance(value, dict):
Validator._validate_numeric_values(value, current_path)# Input validation utilities implementation
|