Coverage for src/utils/validation.py: 18%

1# Copyright (c) CloudZero - ALL RIGHTS RESERVED - PROPRIETARY AND CONFIDENTIAL 

2# Unauthorized copying of this file and/or project, via any medium is strictly prohibited. 

3# Direct all questions to legal@cloudzero.com 

4 

5""" 

6Validation utilities for input validation and error handling. 

7Provides consistent validation patterns across the codebase. 

8""" 

9 

10from pathlib import Path 

11from typing import List, Dict, Any, Optional, Union 

12import polars as pl 

13from .console_helpers import print_error, print_warning 

14 

15 

16class ValidationError(Exception): 

17 """Custom exception for validation errors.""" 

18 pass 

19 

20 

21def validate_file_exists(file_path: Union[str, Path]) -> Path: 

22 """Validate that a file exists and return Path object.""" 

23 path = Path(file_path) 

24 if not path.exists(): 

25 raise ValidationError(f"File does not exist: {path}") 

26 if not path.is_file(): 

27 raise ValidationError(f"Path is not a file: {path}") 

28 return path 

29 

30 

31def validate_directory_exists(dir_path: Union[str, Path]) -> Path: 

32 """Validate that a directory exists and return Path object.""" 

33 path = Path(dir_path) 

34 if not path.exists(): 

35 raise ValidationError(f"Directory does not exist: {path}") 

36 if not path.is_dir(): 

37 raise ValidationError(f"Path is not a directory: {path}") 

38 return path 

39 

40 

41def validate_columns_exist(df: pl.DataFrame, columns: List[str]) -> List[str]: 

42 """Validate that columns exist in DataFrame and return available columns.""" 

43 available_columns = df.columns 

44 missing_columns = [col for col in columns if col not in available_columns] 

45 

46 if missing_columns: 

47 print_error(f"Columns not found: {', '.join(missing_columns)}") 

48 print_error(f"Available columns: {', '.join(available_columns)}") 

49 raise ValidationError(f"Missing columns: {missing_columns}") 

50 

51 return columns 

52 

53 

54def validate_single_column_exists(df: pl.DataFrame, column: str) -> str: 

55 """Validate that a single column exists in DataFrame.""" 

56 if column not in df.columns: 

57 print_error(f"Column '{column}' not found") 

58 print_error(f"Available columns: {', '.join(df.columns)}") 

59 raise ValidationError(f"Column '{column}' not found") 

60 

61 return column 

62 

63 

64def validate_numeric_columns(df: pl.DataFrame, columns: List[str]) -> List[str]: 

65 """Validate that columns are numeric types.""" 

66 numeric_types = {pl.Int8, pl.Int16, pl.Int32, pl.Int64, pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64, pl.Float32, pl.Float64} 

67 

68 invalid_columns = [] 

69 for col in columns: 

70 if col in df.columns: 

71 col_type = df[col].dtype 

72 if col_type not in numeric_types: 

73 invalid_columns.append(f"{col} ({col_type})") 

74 

75 if invalid_columns: 

76 print_error(f"Non-numeric columns: {', '.join(invalid_columns)}") 

77 raise ValidationError(f"Columns must be numeric: {invalid_columns}") 

78 

79 return columns 

80 

81 

82def validate_datetime_columns(df: pl.DataFrame, columns: List[str]) -> List[str]: 

83 """Validate that columns are datetime types.""" 

84 datetime_types = {pl.Datetime, pl.Date} 

85 

86 invalid_columns = [] 

87 for col in columns: 

88 if col in df.columns: 

89 col_type = df[col].dtype 

90 if col_type not in datetime_types: 

91 invalid_columns.append(f"{col} ({col_type})") 

92 

93 if invalid_columns: 

94 print_error(f"Non-datetime columns: {', '.join(invalid_columns)}") 

95 raise ValidationError(f"Columns must be datetime: {invalid_columns}") 

96 

97 return columns 

98 

99 

100def validate_non_empty_dataframe(df: pl.DataFrame, min_rows: int = 1) -> None: 

101 """Validate that DataFrame has minimum number of rows.""" 

102 if len(df) < min_rows: 

103 raise ValidationError(f"DataFrame must have at least {min_rows} rows, got {len(df)}") 

104 

105 

106def validate_config_value( 

107 config: Dict[str, Any], 

108 key: str, 

109 required: bool = False, 

110 valid_values: Optional[List[Any]] = None, 

111 value_type: Optional[type] = None 

112) -> Any: 

113 """Validate configuration value.""" 

114 value = config.get(key) 

115 

116 if value is None: 

117 if required: 

118 raise ValidationError(f"Required configuration key missing: {key}") 

119 return None 

120 

121 if value_type and not isinstance(value, value_type): 

122 raise ValidationError(f"Configuration key '{key}' must be of type {value_type.__name__}, got {type(value).__name__}") 

123 

124 if valid_values and value not in valid_values: 

125 raise ValidationError(f"Configuration key '{key}' must be one of {valid_values}, got {value}") 

126 

127 return value 

128 

129 

130def validate_aggregation_spec(agg_spec: str) -> Dict[str, str]: 

131 """Validate and parse aggregation specification.""" 

132 try: 

133 if ':' not in agg_spec: 

134 raise ValidationError(f"Invalid aggregation spec format: {agg_spec}") 

135 

136 parts = agg_spec.split(':') 

137 if len(parts) != 2: 

138 raise ValidationError(f"Invalid aggregation spec format: {agg_spec}") 

139 

140 group_cols_str, agg_str = parts 

141 group_cols = [col.strip() for col in group_cols_str.split(',')] 

142 

143 return { 

144 'group_columns': group_cols, 

145 'aggregation': agg_str.strip() 

146 } 

147 except Exception as e: 

148 raise ValidationError(f"Failed to parse aggregation spec '{agg_spec}': {e}") 

149 

150 

151def validate_sample_size(sample_size: int, total_rows: int) -> int: 

152 """Validate sample size parameters.""" 

153 if sample_size <= 0: 

154 raise ValidationError("Sample size must be positive") 

155 

156 if sample_size > total_rows: 

157 print_warning(f"Sample size ({sample_size}) larger than total rows ({total_rows})") 

158 return total_rows 

159 

160 return sample_size 

161 

162 

163def validate_cardinality_threshold(threshold: float) -> float: 

164 """Validate cardinality threshold value.""" 

165 if not 0 <= threshold <= 1: 

166 raise ValidationError("Cardinality threshold must be between 0 and 1") 

167 return threshold 

168 

169 

170def check_null_columns(df: pl.DataFrame, columns: List[str]) -> Dict[str, int]: 

171 """Check for null values in specified columns and return counts.""" 

172 null_counts = {} 

173 for col in columns: 

174 if col in df.columns: 

175 null_count = df[col].null_count() 

176 if null_count > 0: 

177 null_counts[col] = null_count 

178 return null_counts 

179 

180 

181def validate_telemetry_type(telemetry_type: str) -> str: 

182 """Validate telemetry type parameter.""" 

183 valid_types = ["allocation", "unit_cost"] 

184 if telemetry_type not in valid_types: 

185 raise ValidationError(f"Telemetry type must be one of {valid_types}, got {telemetry_type}") 

186 return telemetry_type 

187 

188 

189def validate_granularity(granularity: str) -> str: 

190 """Validate granularity parameter.""" 

191 valid_granularities = ["HOURLY", "DAILY"] 

192 granularity_upper = granularity.upper() 

193 if granularity_upper not in valid_granularities: 

194 raise ValidationError(f"Granularity must be one of {valid_granularities}, got {granularity}") 

195 return granularity_upper 

196 

197 

198def safe_convert_to_int(value: str, field_name: str = "value") -> int: 

199 """Safely convert string to integer with validation.""" 

200 try: 

201 return int(value) 

202 except ValueError: 

203 raise ValidationError(f"Invalid integer value for {field_name}: {value}") 

204 

205 

206def safe_convert_to_float(value: str, field_name: str = "value") -> float: 

207 """Safely convert string to float with validation.""" 

208 try: 

209 return float(value) 

210 except ValueError: 

211 raise ValidationError(f"Invalid float value for {field_name}: {value}") 

212 

213 

214def validate_choice_in_range(choice: int, min_val: int, max_val: int, field_name: str = "choice") -> int: 

215 """Validate that a choice is within valid range.""" 

216 if not min_val <= choice <= max_val: 

217 raise ValidationError(f"{field_name} must be between {min_val} and {max_val}, got {choice}") 

218 return choice