Coverage for youversion/utils.py: 100%

1"""Utility functions for YouVersion API client.""" 

2 

3import hashlib 

4import re 

5from typing import Any 

6 

7from pydantic import BaseModel, create_model 

8 

9 

10class DynamicPydanticFactory: 

11 """Factory for creating Pydantic models dynamically from API responses.""" 

12 

13 def __init__(self): 

14 """Initialize the factory with a cache for generated classes.""" 

15 self._class_cache: dict[str, type[BaseModel]] = {} 

16 

17 def _sanitize_name(self, name: str) -> str: 

18 """Sanitize a name to be a valid Python identifier. 

19 

20 Args: 

21 name: Original name 

22 

23 Returns: 

24 Sanitized name valid as Python identifier 

25 """ 

26 # Replace invalid characters with underscores 

27 name = re.sub(r"[^a-zA-Z0-9_]", "_", name) 

28 # Ensure it doesn't start with a number 

29 if name and name[0].isdigit(): 

30 name = "_" + name 

31 # Ensure it's not empty 

32 if not name: 

33 name = "_empty" 

34 return name 

35 

36 def _infer_type(self, value: Any, field_name: str = "") -> tuple[Any, Any]: 

37 """Infer Python type from a value. 

38 

39 Args: 

40 value: Value to infer type from 

41 field_name: Name of the field (for context) 

42 

43 Returns: 

44 Tuple of (type, default_value) 

45 """ 

46 if value is None: 

47 return (Any | None, None) 

48 elif isinstance(value, bool): 

49 return (bool, False) 

50 elif isinstance(value, int): 

51 return (int, 0) 

52 elif isinstance(value, float): 

53 return (float, 0.0) 

54 elif isinstance(value, str): 

55 # ponytail: API often sends null for optional string fields 

56 return (str | None, None) 

57 elif isinstance(value, list): 

58 if not value: 

59 # Empty list - can't infer element type 

60 # Use field_name to create model name 

61 # (e.g., "verses" -> "Verse") 

62 if field_name: 

63 # Singularize and PascalCase: "verses" -> "Verse" 

64 element_class_name = self._get_element_class_name(field_name) 

65 element_class = self.create_model(element_class_name, {}) 

66 return (list[element_class] | None, None) 

67 return (list[Any] | None, None) 

68 # Infer type from first element 

69 # Use field_name to create model name for dict elements 

70 if isinstance(value[0], dict) and field_name: 

71 element_class_name = self._get_element_class_name(field_name) 

72 element_type, _ = self._infer_type(value[0], element_class_name) 

73 else: 

74 element_type, _ = self._infer_type(value[0], field_name) 

75 return (list[element_type] | None, None) 

76 elif isinstance(value, dict): 

77 # Nested dict - create a nested Pydantic model 

78 nested_class_name = ( 

79 self._get_element_class_name(field_name) 

80 if field_name 

81 else self._sanitize_name(field_name) 

82 ) 

83 nested_class = self.create_model(nested_class_name, value) 

84 # Return as optional type with None default 

85 return (nested_class | None, None) 

86 else: 

87 return (Any, None) 

88 

89 def _get_element_class_name(self, field_name: str) -> str: 

90 """Get a class name for list elements based on field name. 

91 

92 Converts snake_case to PascalCase and handles plural forms. 

93 Examples: "download_urls" -> "DownloadUrl", "user_ids" -> "UserId" 

94 

95 Args: 

96 field_name: Field name (e.g., "verses", "download_urls", "results") 

97 

98 Returns: 

99 Singularized and PascalCase class name 

100 (e.g., "Verse", "DownloadUrl") 

101 """ 

102 sanitized = self._sanitize_name(field_name) 

103 if sanitized: 

104 # Handle plural forms: remove trailing 's' if present 

105 if sanitized.endswith("s") and len(sanitized) > 1: 

106 singular = sanitized[:-1] 

107 else: 

108 singular = sanitized 

109 

110 # Convert snake_case to PascalCase 

111 # Split by underscore, capitalize each word, then join 

112 parts = singular.split("_") 

113 # Check if already in PascalCase 

114 # (no underscores and has mixed case) 

115 if len(parts) == 1 and singular[0].isupper(): 

116 # Already PascalCase - preserve it 

117 pascal_case = singular 

118 else: 

119 # Convert to PascalCase: capitalize each word 

120 pascal_case = "".join(word.capitalize() for word in parts if word) 

121 

122 # Ensure it's not empty after processing 

123 if not pascal_case or pascal_case == "_empty": 

124 return "Item" 

125 return pascal_case 

126 return "Item" 

127 

128 def _value_signature(self, value: Any) -> str: 

129 """Build a stable shape signature for cache keys.""" 

130 if value is None: 

131 return "none" 

132 if isinstance(value, bool): 

133 return "bool" 

134 if isinstance(value, int): 

135 return "int" 

136 if isinstance(value, float): 

137 return "float" 

138 if isinstance(value, str): 

139 return "str" 

140 if isinstance(value, list): 

141 if not value: 

142 return "list" 

143 return f"list[{self._value_signature(value[0])}]" 

144 if isinstance(value, dict): 

145 parts = [ 

146 f"{k}:{self._value_signature(v)}" for k, v in sorted(value.items()) 

147 ] 

148 return "{" + ",".join(parts) + "}" 

149 return type(value).__name__ 

150 

151 def _cache_key(self, class_name: str, data: dict[str, Any]) -> str: 

152 """Cache key from model name and data shape (not object id). 

153 

154 ponytail: id(data) is reused after GC and caused stale schemas 

155 when paging moments (page-1 Extra model applied to page-2 data). 

156 """ 

157 digest = hashlib.sha256(self._value_signature(data).encode()).hexdigest()[:16] 

158 return f"{class_name}_{digest}" 

159 

160 def create_model(self, class_name: str, data: dict[str, Any]) -> type[BaseModel]: 

161 """Create a Pydantic model dynamically from a dictionary. 

162 

163 Args: 

164 class_name: Name for the generated class 

165 data: Dictionary containing the data structure 

166 

167 Returns: 

168 Generated Pydantic model class 

169 """ 

170 # Sanitize class name first 

171 sanitized = self._sanitize_name(class_name) 

172 # Check if already in PascalCase (multiple capital letters) 

173 if sanitized and sanitized[0].isupper(): 

174 # Check if it's already PascalCase 

175 # (has multiple words or is single word) 

176 # If it contains underscores, convert to PascalCase 

177 if "_" in sanitized: 

178 parts = sanitized.split("_") 

179 final_class_name = "".join(word.capitalize() for word in parts if word) 

180 else: 

181 # Already PascalCase or single word - preserve it 

182 final_class_name = sanitized 

183 elif sanitized: 

184 # Not uppercase, capitalize first letter only 

185 final_class_name = sanitized.capitalize() 

186 else: 

187 final_class_name = sanitized 

188 

189 # Check cache with sanitized class name and data shape 

190 cache_key = self._cache_key(final_class_name, data) 

191 if cache_key in self._class_cache: 

192 return self._class_cache[cache_key] 

193 

194 # Use final_class_name for model creation 

195 class_name = final_class_name 

196 

197 # Build field definitions for Pydantic 

198 # Make all fields optional to avoid field ordering issues 

199 field_definitions: dict[str, tuple[Any, Any]] = {} 

200 

201 for key, value in data.items(): 

202 field_name = self._sanitize_name(key) 

203 field_type, default_value = self._infer_type(value, field_name) 

204 

205 # Handle different default value types 

206 # Check if default_value is a Field instance by checking type name 

207 if ( 

208 default_value is not None 

209 and hasattr(default_value, "__class__") 

210 and default_value.__class__.__name__ == "FieldInfo" 

211 ): 

212 # Already a Pydantic Field 

213 field_definitions[field_name] = (field_type, default_value) 

214 elif default_value is None: 

215 # Optional field with None default 

216 type_str = str(field_type) 

217 if ( 

218 "Optional" in type_str 

219 or "Union" in type_str 

220 or "| None" in type_str 

221 ): 

222 optional_type = field_type 

223 else: 

224 optional_type = field_type | None 

225 field_definitions[field_name] = (optional_type, None) 

226 else: 

227 # Field with explicit default value 

228 field_definitions[field_name] = (field_type, default_value) 

229 

230 # Create the Pydantic model 

231 try: 

232 generated_class = create_model(class_name, **field_definitions) 

233 

234 # Cache the class 

235 self._class_cache[cache_key] = generated_class 

236 

237 return generated_class 

238 except (TypeError, ValueError): 

239 # Fallback to a simple model if creation fails 

240 fallback_fields = {self._sanitize_name(k): (Any, None) for k in data.keys()} 

241 return create_model(class_name, **fallback_fields) 

242 

243 def create_instance(self, class_name: str, data: dict[str, Any]) -> Any: 

244 """Create a Pydantic model instance from a dictionary. 

245 

246 Args: 

247 class_name: Name for the generated class 

248 data: Dictionary containing the data 

249 

250 Returns: 

251 Instance of the generated Pydantic model 

252 """ 

253 model_type = self.create_model(class_name, data) 

254 return self._create_instance_recursive(model_type, data) 

255 

256 def _create_instance_recursive( 

257 self, model_type: type[BaseModel], data: dict[str, Any] 

258 ) -> Any: 

259 """Recursively create Pydantic model instance. 

260 

261 Handles nested dictionaries and lists, creating model instances. 

262 

263 Args: 

264 model_type: The Pydantic model type to instantiate 

265 data: Dictionary containing the data 

266 

267 Returns: 

268 Instance of the Pydantic model 

269 """ 

270 processed_data = {} 

271 

272 for key, value in data.items(): 

273 field_name = self._sanitize_name(key) 

274 

275 # Handle nested dictionaries 

276 if isinstance(value, dict): 

277 # Check if the field type is a nested model 

278 field_info = model_type.model_fields.get(field_name) 

279 if field_info: 

280 annotation = field_info.annotation 

281 # Handle Optional types 

282 if hasattr(annotation, "__args__"): 

283 args = annotation.__args__ 

284 # Find the model type in Optional[ModelType] 

285 nested_class = None 

286 for arg in args: 

287 if hasattr(arg, "model_fields"): 

288 nested_class = arg 

289 break 

290 if nested_class: 

291 processed_data[field_name] = ( 

292 self._create_instance_recursive(nested_class, value) 

293 ) 

294 else: 

295 # No model type found, pass dict 

296 processed_data[field_name] = value 

297 elif hasattr(annotation, "model_fields"): 

298 # Direct model type 

299 processed_data[field_name] = self._create_instance_recursive( 

300 annotation, value 

301 ) 

302 else: 

303 # dict type, pass as-is 

304 processed_data[field_name] = value 

305 else: 

306 # No field info, pass as-is 

307 processed_data[field_name] = value 

308 # Handle lists 

309 elif isinstance(value, list): 

310 processed_list = [] 

311 for item in value: 

312 if isinstance(item, dict): 

313 # Check if list element type is a model 

314 field_info = model_type.model_fields.get(field_name) 

315 if field_info: 

316 annotation = field_info.annotation 

317 # Handle list[ModelType] or list[Any] 

318 if hasattr(annotation, "__args__"): 

319 element_type = annotation.__args__[0] 

320 if hasattr(element_type, "model_fields"): 

321 item_class = element_type 

322 processed_list.append( 

323 self._create_instance_recursive( 

324 item_class, item 

325 ) 

326 ) 

327 else: 

328 processed_list.append(item) 

329 else: 

330 processed_list.append(item) 

331 else: 

332 processed_list.append(item) 

333 else: 

334 processed_list.append(item) 

335 processed_data[field_name] = processed_list 

336 else: 

337 processed_data[field_name] = value 

338 

339 # Pydantic models handle extra fields and validation automatically 

340 # Use model_validate for better compatibility with Pydantic v2 

341 try: 

342 return model_type.model_validate(processed_data) 

343 except Exception: 

344 # If validation fails, try with only known fields 

345 model_fields = set(model_type.model_fields.keys()) 

346 filtered_data = { 

347 k: v for k, v in processed_data.items() if k in model_fields 

348 } 

349 try: 

350 return model_type.model_validate(filtered_data) 

351 except Exception: 

352 # Last resort: try direct instantiation 

353 return model_type(**filtered_data) 

354 

355 

356# Global factory instance 

357_factory = DynamicPydanticFactory() 

358 

359 

360def create_model_from_response( 

361 class_name: str, data: dict[str, Any] 

362) -> type[BaseModel]: 

363 """Create a Pydantic model dynamically from an API response. 

364 

365 Args: 

366 class_name: Name for the generated class 

367 data: Dictionary containing the API response data 

368 

369 Returns: 

370 Generated Pydantic model class 

371 

372 Example: 

373 >>> response = {"id": 1, "name": "Test"} 

374 >>> MyClass = create_model_from_response("MyResponse", response) 

375 >>> instance = MyClass(id=1, name="Test") 

376 """ 

377 return _factory.create_model(class_name, data) 

378 

379 

380def create_instance_from_response(class_name: str, data: dict[str, Any]) -> Any: 

381 """Create a Pydantic model instance from an API response. 

382 

383 Args: 

384 class_name: Name for the generated class 

385 data: Dictionary containing the API response data 

386 

387 Returns: 

388 Instance of the generated Pydantic model 

389 

390 Example: 

391 >>> response = {"id": 1, "name": "Test"} 

392 >>> instance = create_instance_from_response("MyResponse", response) 

393 >>> print(instance.id) # 1 

394 """ 

395 return _factory.create_instance(class_name, data) 

396 

397 

398# Backward compatibility aliases 

399create_dataclass_from_response = create_model_from_response