Coverage for smartmdao / validation.py: 100%

1import logging 

2from typing import Any, Dict, List, Optional, Protocol, Union, get_args, get_origin, runtime_checkable 

3 

4from .graph import map_producers 

5from .models import Step 

6 

7# Initialize module-level logger 

8logger = logging.getLogger(__name__) 

9 

10 

11class TypeMismatchError(TypeError): 

12 """Raised when a value, or a declared type, doesn't satisfy an expected type annotation.""" 

13 

14 

15@runtime_checkable 

16class TypeChecker(Protocol): 

17 """ 

18 Pluggable strategy deciding what "type compatible" means. 

19 

20 Implement this to customize how strict/lenient SmartMDAO's type 

21 enforcement is - e.g. to accept numpy scalars, or to allow `int` 

22 where a `float` is declared - without touching the core framework. 

23 """ 

24 

25 def check_value(self, value: Any, expected: type) -> bool: 

26 """Does `value` satisfy the `expected` annotation?""" 

27 ... 

28 

29 def check_types(self, produced: type, expected: type) -> bool: 

30 """Is everything satisfying `produced` guaranteed to also satisfy `expected`?""" 

31 ... 

32 

33 

34def _concrete_classes(annotation: Any) -> tuple: 

35 """ 

36 Reduces a (possibly Optional/Union) type annotation down to a flat 

37 tuple of concrete classes suitable for isinstance/issubclass checks. 

38 

39 Returns () for annotations we can't resolve to concrete classes 

40 (e.g. `typing.Any`, a bare `TypeVar`) - callers should treat that as 

41 "no constraint" rather than a failure. 

42 """ 

43 if annotation is Any: 

44 return () 

45 

46 origin = get_origin(annotation) 

47 if origin is Union: 

48 classes = [] 

49 for arg in get_args(annotation): 

50 classes.extend(_concrete_classes(arg)) 

51 return tuple(classes) 

52 

53 # Generic containers (e.g. list[float], dict[str, int]) are checked 

54 # structurally on their origin only - contents are not inspected. 

55 if origin is not None: 

56 return (origin,) 

57 

58 if isinstance(annotation, type): 

59 return (annotation,) 

60 

61 return () 

62 

63 

64def _format_type(tp: Any) -> str: 

65 return getattr(tp, "__name__", str(tp)) 

66 

67 

68class StandardTypeChecker: 

69 """ 

70 Strict, dependency-free type checker built on `isinstance`/`issubclass`. 

71 

72 Understands `Optional[X]`, `Union[X, Y]`, generic containers (checked 

73 structurally on their origin), and `typing.Any` (always satisfied). 

74 Does not consider `int` compatible with `float` - if you need that, 

75 write a `TypeChecker` that loosens `_concrete_classes` accordingly. 

76 """ 

77 

78 def check_value(self, value: Any, expected: type) -> bool: 

79 classes = _concrete_classes(expected) 

80 return not classes or isinstance(value, classes) 

81 

82 def check_types(self, produced: type, expected: type) -> bool: 

83 expected_classes = _concrete_classes(expected) 

84 if not expected_classes: 

85 return True 

86 

87 produced_classes = _concrete_classes(produced) 

88 if not produced_classes: 

89 return True 

90 

91 return all(issubclass(p, expected_classes) for p in produced_classes) 

92 

93 

94def validate_structure(steps: List[Step], checker: Optional[TypeChecker] = None) -> None: 

95 """ 

96 Statically validates every producer -> consumer edge in the pipeline: 

97 the type a step declares for an output must be compatible with the 

98 type every consumer of that variable declares for its input. 

99 

100 This is pure structural analysis - no step is ever executed - so it 

101 only needs to run once per pipeline shape, regardless of how many 

102 times the pipeline is later evaluated (e.g. inside an optimization loop). 

103 """ 

104 checker = checker or StandardTypeChecker() 

105 producers = map_producers(steps) 

106 

107 for consumer in steps: 

108 for param_name, expected_type in consumer.resolve_input_types().items(): 

109 producer = producers.get(param_name) 

110 if producer is None: 

111 continue # External input; validated per-call in validate_external_inputs. 

112 

113 produced_type = producer.resolve_output_types().get(param_name) 

114 if produced_type is None: 

115 continue # Producer didn't declare a type for this output. 

116 

117 if not checker.check_types(produced_type, expected_type): 

118 raise TypeMismatchError( 

119 f"Type mismatch on '{param_name}': step '{producer.name}' declares " 

120 f"{param_name} -> {_format_type(produced_type)}, but step " 

121 f"'{consumer.name}' expects {param_name}: {_format_type(expected_type)}." 

122 ) 

123 

124 logger.debug(f"Structural type validation passed for {len(steps)} steps.") 

125 

126 

127def validate_external_inputs( 

128 steps: List[Step], inputs: Dict[str, Any], checker: Optional[TypeChecker] = None 

129) -> None: 

130 """ 

131 Validates the concrete input values passed into `Pipeline.run(**inputs)` 

132 against the declared parameter types of the steps that consume them 

133 directly (i.e. variables that aren't produced internally by another step). 

134 """ 

135 checker = checker or StandardTypeChecker() 

136 producers = map_producers(steps) 

137 

138 for consumer in steps: 

139 for param_name, expected_type in consumer.resolve_input_types().items(): 

140 if param_name in producers or param_name not in inputs: 

141 continue 

142 

143 value = inputs[param_name] 

144 if not checker.check_value(value, expected_type): 

145 raise TypeMismatchError( 

146 f"Input '{param_name}'={value!r} ({type(value).__name__}) does not match " 

147 f"the type expected by step '{consumer.name}': {param_name}: {_format_type(expected_type)}." 

148 )