Coverage for src/quickmcp/factory/type_conversion.py: 72%

1""" 

2Safe type conversion system for MCP Factory. 

3""" 

4 

5import json 

6import logging 

7from typing import Any, Dict, Type, Union, get_origin, get_args, Optional, _SpecialForm 

8from datetime import datetime, date 

9from pathlib import Path 

10from decimal import Decimal 

11from enum import Enum 

12from dataclasses import is_dataclass, fields 

13 

14from .config import FactoryConfig, DEFAULT_CONFIG 

15from .errors import TypeConversionError 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class TypeConverter: 

21 """Safe type conversion with validation and error handling.""" 

22 

23 def __init__(self, config: Optional[FactoryConfig] = None): 

24 self.config = config or DEFAULT_CONFIG 

25 self._conversion_cache: Dict[tuple, Any] = {} 

26 

27 def convert_value(self, value: Any, target_type: Type, parameter_name: str = None) -> Any: 

28 """ 

29 Safely convert a value to the target type. 

30  

31 Args: 

32 value: The value to convert 

33 target_type: The target type to convert to 

34 parameter_name: Name of the parameter (for error messages) 

35  

36 Returns: 

37 The converted value 

38  

39 Raises: 

40 TypeConversionError: If conversion fails or is unsafe 

41 """ 

42 # If value is already the correct type, return as-is 

43 # Skip this check for generic types which can't be used with isinstance 

44 if not hasattr(target_type, '__origin__'): 

45 try: 

46 if isinstance(value, target_type): 

47 return value 

48 except TypeError: 

49 # Generic types can't be used with isinstance 

50 pass 

51 

52 # Handle None values 

53 if value is None: 

54 if self._is_optional_type(target_type): 

55 return None 

56 elif self.config.strict_type_conversion: 

57 raise TypeConversionError( 

58 f"Cannot convert None to non-optional type {target_type.__name__}", 

59 value, target_type, parameter_name 

60 ) 

61 else: 

62 return None 

63 

64 # Use cached conversion if available 

65 cache_key = (type(value), target_type, value if isinstance(value, (str, int, float, bool)) else None) 

66 if cache_key in self._conversion_cache: 

67 return self._conversion_cache[cache_key] 

68 

69 try: 

70 converted = self._convert_value_impl(value, target_type, parameter_name) 

71 

72 # Cache simple conversions 

73 if isinstance(value, (str, int, float, bool)): 

74 self._conversion_cache[cache_key] = converted 

75 

76 return converted 

77 

78 except Exception as e: 

79 if isinstance(e, TypeConversionError): 

80 raise 

81 else: 

82 raise TypeConversionError( 

83 f"Failed to convert {type(value).__name__} to {target_type.__name__}: {e}", 

84 value, target_type, parameter_name 

85 ) from e 

86 

87 def _convert_value_impl(self, value: Any, target_type: Type, parameter_name: str = None) -> Any: 

88 """Internal implementation of value conversion.""" 

89 

90 # Handle Any type - pass through unchanged 

91 if target_type is Any: 

92 return value 

93 

94 # Handle basic types 

95 if target_type in (str, int, float, bool): 

96 return self._convert_basic_type(value, target_type) 

97 

98 # Handle container types 

99 elif target_type in (list, tuple, set): 

100 return self._convert_container_type(value, target_type) 

101 

102 elif target_type == dict: 

103 return self._convert_dict_type(value) 

104 

105 # Handle special types 

106 elif target_type == bytes: 

107 return self._convert_bytes_type(value) 

108 

109 elif target_type == Path: 

110 return self._convert_path_type(value) 

111 

112 elif target_type in (datetime, date): 

113 return self._convert_datetime_type(value, target_type) 

114 

115 elif target_type == Decimal: 

116 return self._convert_decimal_type(value) 

117 

118 # Handle Enum types 

119 elif isinstance(target_type, type) and issubclass(target_type, Enum): 

120 return self._convert_enum_type(value, target_type) 

121 

122 # Handle dataclass types 

123 elif is_dataclass(target_type): 

124 return self._convert_dataclass_type(value, target_type) 

125 

126 # Handle Union types (including Optional) 

127 elif self._is_union_type(target_type): 

128 return self._convert_union_type(value, target_type, parameter_name) 

129 

130 # Handle generic types (List[int], Dict[str, str], etc.) 

131 elif hasattr(target_type, '__origin__'): 

132 return self._convert_generic_type(value, target_type, parameter_name) 

133 

134 # For other types, only allow conversion if explicitly enabled 

135 elif self.config.allow_type_coercion: 

136 try: 

137 return target_type(value) 

138 except Exception as e: 

139 raise TypeConversionError( 

140 f"Type coercion failed: {e}", 

141 value, target_type, parameter_name 

142 ) 

143 

144 else: 

145 raise TypeConversionError( 

146 f"No safe conversion available from {type(value).__name__} to {target_type.__name__}", 

147 value, target_type, parameter_name 

148 ) 

149 

150 def _convert_basic_type(self, value: Any, target_type: Type) -> Any: 

151 """Convert to basic types (str, int, float, bool).""" 

152 

153 if target_type == str: 

154 if isinstance(value, bytes): 

155 return value.decode('utf-8') 

156 return str(value) 

157 

158 elif target_type == int: 

159 if isinstance(value, (int, bool)): 

160 return int(value) 

161 elif isinstance(value, float): 

162 if self.config.strict_type_conversion and not self.config.allow_type_coercion: 

163 raise TypeConversionError(f"Float to int conversion not allowed in strict mode", value, target_type) 

164 if value.is_integer(): 

165 return int(value) 

166 elif self.config.strict_type_conversion: 

167 raise TypeConversionError(f"Float {value} is not an integer", value, target_type) 

168 else: 

169 return int(value) # Truncate 

170 elif isinstance(value, str): 

171 if self.config.strict_type_conversion and not self.config.allow_type_coercion: 

172 raise TypeConversionError(f"String to int conversion not allowed in strict mode", value, target_type) 

173 try: 

174 return int(value) 

175 except ValueError: 

176 # Try parsing as float first, then convert to int 

177 try: 

178 float_val = float(value) 

179 return int(float_val) 

180 except ValueError: 

181 raise TypeConversionError(f"Cannot parse '{value}' as integer", value, target_type) 

182 else: 

183 raise TypeConversionError(f"Cannot convert {type(value).__name__} to int", value, target_type) 

184 

185 elif target_type == float: 

186 if isinstance(value, (int, float, bool)): 

187 return float(value) 

188 elif isinstance(value, str): 

189 try: 

190 return float(value) 

191 except ValueError: 

192 raise TypeConversionError(f"Cannot parse '{value}' as float", value, target_type) 

193 else: 

194 raise TypeConversionError(f"Cannot convert {type(value).__name__} to float", value, target_type) 

195 

196 elif target_type == bool: 

197 if isinstance(value, bool): 

198 return value 

199 elif isinstance(value, (int, float)): 

200 return bool(value) 

201 elif isinstance(value, str): 

202 lower_val = value.lower().strip() 

203 if lower_val in ('true', '1', 'yes', 'on', 'enabled'): 

204 return True 

205 elif lower_val in ('false', '0', 'no', 'off', 'disabled', ''): 

206 return False 

207 else: 

208 raise TypeConversionError(f"Cannot parse '{value}' as boolean", value, target_type) 

209 else: 

210 return bool(value) 

211 

212 def _convert_container_type(self, value: Any, target_type: Type) -> Any: 

213 """Convert to container types (list, tuple, set).""" 

214 

215 if target_type == list: 

216 if isinstance(value, (list, tuple, set)): 

217 return list(value) 

218 elif isinstance(value, str): 

219 # Try to parse as JSON array 

220 try: 

221 parsed = json.loads(value) 

222 if isinstance(parsed, list): 

223 return parsed 

224 except (json.JSONDecodeError, TypeError): 

225 pass 

226 # Fall back to splitting by common delimiters 

227 for delimiter in [',', ';', '|', ' ']: 

228 if delimiter in value: 

229 return [item.strip() for item in value.split(delimiter) if item.strip()] 

230 return [value] # Single item list 

231 else: 

232 return [value] # Single item list 

233 

234 elif target_type == tuple: 

235 result = self._convert_container_type(value, list) 

236 return tuple(result) 

237 

238 elif target_type == set: 

239 result = self._convert_container_type(value, list) 

240 return set(result) 

241 

242 def _convert_dict_type(self, value: Any) -> dict: 

243 """Convert to dict type.""" 

244 if isinstance(value, dict): 

245 return value 

246 elif isinstance(value, str): 

247 try: 

248 parsed = json.loads(value) 

249 if isinstance(parsed, dict): 

250 return parsed 

251 except (json.JSONDecodeError, TypeError): 

252 pass 

253 # Try to parse as key=value pairs 

254 if '=' in value: 

255 result = {} 

256 for pair in value.split(','): 

257 if '=' in pair: 

258 key, val = pair.split('=', 1) 

259 result[key.strip()] = val.strip() 

260 return result 

261 

262 raise TypeConversionError(f"Cannot convert {type(value).__name__} to dict", value, dict) 

263 

264 def _convert_path_type(self, value: Any) -> Path: 

265 """Convert to Path type.""" 

266 if isinstance(value, Path): 

267 return value 

268 elif isinstance(value, str): 

269 return Path(value) 

270 else: 

271 return Path(str(value)) 

272 

273 def _convert_datetime_type(self, value: Any, target_type: Type) -> Union[datetime, date]: 

274 """Convert to datetime or date types.""" 

275 if isinstance(value, target_type): 

276 return value 

277 elif isinstance(value, str): 

278 # Try common datetime formats 

279 formats = [ 

280 '%Y-%m-%d %H:%M:%S', 

281 '%Y-%m-%d %H:%M:%S.%f', 

282 '%Y-%m-%dT%H:%M:%S', 

283 '%Y-%m-%dT%H:%M:%S.%f', 

284 '%Y-%m-%d' 

285 ] 

286 

287 for fmt in formats: 

288 try: 

289 parsed = datetime.strptime(value, fmt) 

290 if target_type == date: 

291 return parsed.date() 

292 return parsed 

293 except ValueError: 

294 continue 

295 

296 raise TypeConversionError(f"Cannot parse '{value}' as {target_type.__name__}", value, target_type) 

297 elif isinstance(value, (int, float)): 

298 # Treat as Unix timestamp 

299 dt = datetime.fromtimestamp(value) 

300 if target_type == date: 

301 return dt.date() 

302 return dt 

303 else: 

304 raise TypeConversionError(f"Cannot convert {type(value).__name__} to {target_type.__name__}", value, target_type) 

305 

306 def _convert_decimal_type(self, value: Any) -> Decimal: 

307 """Convert to Decimal type.""" 

308 if isinstance(value, Decimal): 

309 return value 

310 elif isinstance(value, (int, float)): 

311 return Decimal(str(value)) # Convert via string to avoid float precision issues 

312 elif isinstance(value, str): 

313 try: 

314 return Decimal(value) 

315 except Exception: 

316 raise TypeConversionError(f"Cannot parse '{value}' as Decimal", value, Decimal) 

317 else: 

318 raise TypeConversionError(f"Cannot convert {type(value).__name__} to Decimal", value, Decimal) 

319 

320 def _convert_enum_type(self, value: Any, enum_class: Type[Enum]) -> Enum: 

321 """Convert to Enum type.""" 

322 if isinstance(value, enum_class): 

323 return value 

324 elif isinstance(value, str): 

325 # Try exact match first 

326 try: 

327 return enum_class(value) 

328 except ValueError: 

329 pass 

330 

331 # Try case-insensitive match 

332 if not self.config.strict_type_conversion: 

333 for member in enum_class: 

334 if member.value.lower() == value.lower(): 

335 return member 

336 if member.name.lower() == value.lower(): 

337 return member 

338 

339 raise TypeConversionError( 

340 f"'{value}' is not a valid {enum_class.__name__}", 

341 value, enum_class 

342 ) 

343 else: 

344 # Try direct conversion 

345 try: 

346 return enum_class(value) 

347 except Exception: 

348 raise TypeConversionError( 

349 f"Cannot convert {type(value).__name__} to {enum_class.__name__}", 

350 value, enum_class 

351 ) 

352 

353 def _convert_dataclass_type(self, value: Any, dataclass_type: Type) -> Any: 

354 """Convert to dataclass type.""" 

355 if isinstance(value, dataclass_type): 

356 return value 

357 elif isinstance(value, dict): 

358 # Convert dict to dataclass 

359 field_values = {} 

360 for field in fields(dataclass_type): 

361 if field.name in value: 

362 # Recursively convert field values 

363 field_values[field.name] = self.convert_value( 

364 value[field.name], 

365 field.type, 

366 field.name 

367 ) 

368 elif field.default is not field.default_factory: 

369 # Use default value if available 

370 field_values[field.name] = field.default 

371 

372 return dataclass_type(**field_values) 

373 else: 

374 raise TypeConversionError( 

375 f"Cannot convert {type(value).__name__} to {dataclass_type.__name__}", 

376 value, dataclass_type 

377 ) 

378 

379 def _convert_bytes_type(self, value: Any) -> bytes: 

380 """Convert to bytes type.""" 

381 if isinstance(value, bytes): 

382 return value 

383 elif isinstance(value, str): 

384 return value.encode('utf-8') 

385 elif isinstance(value, (list, tuple)): 

386 # Assume list of integers 

387 return bytes(value) 

388 elif isinstance(value, bytearray): 

389 return bytes(value) 

390 else: 

391 raise TypeConversionError( 

392 f"Cannot convert {type(value).__name__} to bytes", 

393 value, bytes 

394 ) 

395 

396 def _is_union_type(self, target_type: Type) -> bool: 

397 """Check if the target type is a Union type.""" 

398 return get_origin(target_type) is Union 

399 

400 def _is_optional_type(self, target_type: Type) -> bool: 

401 """Check if the target type is Optional (Union[T, None]).""" 

402 if not self._is_union_type(target_type): 

403 return False 

404 

405 args = get_args(target_type) 

406 return len(args) == 2 and type(None) in args 

407 

408 def _convert_union_type(self, value: Any, target_type: Type, parameter_name: str = None) -> Any: 

409 """Convert to Union type by trying each possibility in order.""" 

410 args = get_args(target_type) 

411 

412 # Handle Optional[T] specially 

413 if self._is_optional_type(target_type): 

414 non_none_type = next(arg for arg in args if arg is not type(None)) 

415 return self.convert_value(value, non_none_type, parameter_name) 

416 

417 # Special handling: if value is a string that looks like a number and int comes before str, 

418 # prefer the int conversion 

419 value_type = type(value) 

420 if value_type == str and int in args and str in args: 

421 int_index = args.index(int) 

422 str_index = args.index(str) 

423 if int_index < str_index: 

424 # Try converting to int first 

425 try: 

426 return self.convert_value(value, int, parameter_name) 

427 except TypeConversionError: 

428 pass 

429 

430 # Check if value is already exactly one of the union types 

431 if value_type in args: 

432 return value 

433 

434 # Otherwise, try conversions in order 

435 last_error = None 

436 for union_type in args: 

437 try: 

438 result = self.convert_value(value, union_type, parameter_name) 

439 return result 

440 except TypeConversionError as e: 

441 last_error = e 

442 continue 

443 

444 # If none worked, raise the last error 

445 raise last_error or TypeConversionError( 

446 f"Cannot convert to any type in Union {target_type}", 

447 value, target_type, parameter_name 

448 ) 

449 

450 def _convert_generic_type(self, value: Any, target_type: Type, parameter_name: str = None) -> Any: 

451 """Convert to generic types like List[int], Dict[str, int], etc.""" 

452 origin = get_origin(target_type) 

453 args = get_args(target_type) 

454 

455 if origin == list and args: 

456 # Convert to List[T] 

457 item_type = args[0] 

458 list_value = self.convert_value(value, list, parameter_name) 

459 return [self.convert_value(item, item_type, f"{parameter_name}[{i}]") for i, item in enumerate(list_value)] 

460 

461 elif origin == dict and len(args) == 2: 

462 # Convert to Dict[K, V] 

463 key_type, value_type = args 

464 dict_value = self.convert_value(value, dict, parameter_name) 

465 return { 

466 self.convert_value(k, key_type, f"{parameter_name}[key]"): 

467 self.convert_value(v, value_type, f"{parameter_name}[{k}]") 

468 for k, v in dict_value.items() 

469 } 

470 

471 elif origin == set and args: 

472 # Convert to Set[T] 

473 item_type = args[0] 

474 set_value = self.convert_value(value, set, parameter_name) 

475 return {self.convert_value(item, item_type, f"{parameter_name}[item]") for item in set_value} 

476 

477 elif origin == tuple and args: 

478 # Convert to Tuple[T, ...] 

479 tuple_value = self.convert_value(value, tuple, parameter_name) 

480 if len(args) == len(tuple_value): 

481 return tuple( 

482 self.convert_value(item, arg_type, f"{parameter_name}[{i}]") 

483 for i, (item, arg_type) in enumerate(zip(tuple_value, args)) 

484 ) 

485 elif len(args) == 2 and args[1] == ...: 

486 # Variable length tuple Tuple[T, ...] 

487 item_type = args[0] 

488 return tuple( 

489 self.convert_value(item, item_type, f"{parameter_name}[{i}]") 

490 for i, item in enumerate(tuple_value) 

491 ) 

492 

493 # Fall back to converting to the origin type 

494 return self.convert_value(value, origin, parameter_name) 

495 

496 def clear_cache(self): 

497 """Clear the conversion cache.""" 

498 self._conversion_cache.clear() 

499 logger.debug("Cleared type conversion cache")