Coverage for postrfp / web / suxint / validate.py: 100%

1import inspect 

2import logging 

3 

4from pydantic import BaseModel 

5from webob.response import Response 

6 

7from .extractors import PathArg 

8from postrfp.shared.exceptions import RoutingError 

9from postrfp.shared.types import SuxType 

10 

11 

12log = logging.getLogger(__name__) 

13 

14 

15def validate_handlers( 

16 sux_instance: SuxType, strict: bool = True, check_return_types: bool = True 

17) -> None: 

18 """ 

19 Validate all handlers for missing adaptors, return types, and duplicate names. 

20 

21 Args: 

22 strict: If True, raise RoutingError if any validation error is found. 

23 check_return_types: If True, also validate handler return types. 

24 

25 Raises: 

26 RoutingError: If any errors are found and strict mode is enabled. 

27 """ 

28 from typing import get_origin, Annotated, get_args 

29 

30 def _unwrap_annotated(type_): 

31 if get_origin(type_) is Annotated: 

32 return get_args(type_)[0] 

33 return type_ 

34 

35 errors = [] 

36 warnings = [] 

37 seen_handler_names = set() 

38 

39 for handler in sux_instance.iter_handlers(): 

40 # Check for duplicate handler names 

41 if handler.name in seen_handler_names: 

42 errors.append(f"Duplicate handler name detected: '{handler.name}'.") 

43 else: 

44 seen_handler_names.add(handler.name) 

45 

46 # Validate required arguments 

47 for param_name in handler.required_arguments: 

48 adaptor = getattr(sux_instance.adaptor_module, param_name, None) 

49 if adaptor is None: 

50 adaptor_module_name = ( 

51 sux_instance.adaptor_module.__name__ 

52 if sux_instance.adaptor_module 

53 else "Unknown" 

54 ) 

55 errors.append( 

56 f"Adaptor for required parameter '{param_name}' not found in adaptor module {adaptor_module_name} " 

57 f"for handler '{handler.name}'." 

58 ) 

59 elif isinstance(adaptor, PathArg): 

60 if f"{{{adaptor.doc_name}}}" not in handler.path: 

61 errors.append( 

62 ( 

63 f"PathArg for '{param_name}' unresolvable in handler '{handler.name}'" 

64 f" - expected '_{adaptor.arg_name}' in function name." 

65 ) 

66 ) 

67 

68 # Validate optional arguments 

69 for param_name in handler.optional_arguments: 

70 adaptor = getattr(sux_instance.adaptor_module, param_name, None) 

71 if adaptor is None: 

72 warnings.append( 

73 f"Adaptor for optional parameter '{param_name}' " 

74 f"in handler '{handler.name}' not found in adaptor module {sux_instance.adaptor_module}." 

75 ) 

76 

77 # Validate return types 

78 if check_return_types: 

79 retval = handler.return_value 

80 if retval is not None: 

81 # First, unwrap Annotated to get the underlying type 

82 unwrapped_type = _unwrap_annotated(retval) 

83 # Then, get the origin (e.g., list from list[int]) or the type itself 

84 origin_or_base = get_origin(unwrapped_type) or unwrapped_type 

85 

86 # If the origin/base isn't a class (e.g. TypeVar, NewType, typing.Union args), flag it 

87 if not inspect.isclass(origin_or_base): 

88 errors.append( 

89 f"Handler '{handler.name}' return type '{retval}' resolves to non-class '{origin_or_base}'." 

90 ) 

91 # Perform the subclass check against allowed types 

92 elif not issubclass(origin_or_base, (BaseModel, dict, list, Response)): 

93 errors.append( 

94 f"Handler '{handler.name}' return type '{origin_or_base.__name__}' " 

95 f"is not an instance or subclass of BaseModel, dict, list or Response." 

96 ) 

97 

98 if warnings: 

99 log.warning(f"Validation Warnings: \n\n {'\n'.join(warnings)}") 

100 

101 if errors: 

102 errstring = "\n ".join(errors) 

103 log.error("Validation Errors:\n {errstring}") 

104 if strict: 

105 raise RoutingError( 

106 f"Validation Errors during initialisation: \n\n {errstring}" 

107 )