Coverage for src / bioimageio / spec / _internal / field_warning.py: 90%

1import dataclasses 

2import inspect 

3from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Union, get_args 

4 

5import pydantic.functional_validators 

6from annotated_types import BaseMetadata, GroupedMetadata 

7from loguru import logger 

8from pydantic import TypeAdapter 

9from pydantic_core import PydanticCustomError 

10from pydantic_core.core_schema import ( 

11 NoInfoValidatorFunction, 

12 ValidationInfo, 

13 WithInfoValidatorFunction, 

14) 

15from typing_extensions import Annotated, LiteralString 

16 

17from .utils import SLOTS 

18from .validation_context import get_validation_context 

19from .warning_levels import WARNING, WarningSeverity 

20 

21if TYPE_CHECKING: 

22 from pydantic.functional_validators import _V2Validator # type: ignore 

23 

24 

25ValidatorFunction = Union[NoInfoValidatorFunction, WithInfoValidatorFunction] 

26 

27AnnotationMetaData = Union[BaseMetadata, GroupedMetadata] 

28 

29 

30def warn( 

31 typ: Union[AnnotationMetaData, Any], 

32 msg: LiteralString, # warning message, e.g. "'{value}' incompatible with {typ} 

33 severity: WarningSeverity = WARNING, 

34): 

35 """treat a type or its annotation metadata as a warning condition""" 

36 if isinstance(typ, get_args(AnnotationMetaData)): 

37 typ = Annotated[Any, typ] 

38 

39 validator: TypeAdapter[Any] = TypeAdapter(typ) 

40 

41 return AfterWarner( 

42 validator.validate_python, severity=severity, msg=msg, context={"typ": typ} 

43 ) 

44 

45 

46def _call_validator_func( 

47 func: "_V2Validator", 

48 mode: Literal["after", "before", "plain", "wrap"], 

49 value: Any, 

50 info: ValidationInfo, 

51) -> Any: 

52 # determine if validator needs info arg 

53 # logic adapted from pydantic._internal._decorators.py v2.11.10 

54 sig = inspect.signature(func) 

55 parameters = list(sig.parameters.values()) 

56 n_positional = sum( 

57 1 

58 for param in parameters 

59 if param.kind 

60 in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) 

61 # First argument is the value being validated/serialized, and can have a default value 

62 # (e.g. `float`, which has signature `(x=0, /)`). We assume other parameters (the info arg 

63 # for instance) should be required, and thus without any default value. 

64 and (param.default is inspect.Parameter.empty or param is parameters[0]) 

65 ) 

66 needs_info = None 

67 if mode == "wrap": 

68 if n_positional == 3: 

69 needs_info = True 

70 elif n_positional == 2: 

71 needs_info = False 

72 else: 

73 assert mode in {"before", "after", "plain"}, ( 

74 f"invalid mode: {mode!r}, expected 'before', 'after' or 'plain" 

75 ) 

76 if n_positional == 2: 

77 needs_info = True 

78 elif n_positional == 1: 

79 needs_info = False 

80 

81 assert needs_info is not None, "could not determine if validator needs info arg" 

82 if needs_info: 

83 return func(value, info) # type: ignore 

84 else: 

85 return func(value) # type: ignore 

86 

87 

88def as_warning( 

89 func: "_V2Validator", 

90 *, 

91 mode: Literal["after", "before", "plain", "wrap"] = "after", 

92 severity: WarningSeverity = WARNING, 

93 msg: Optional[LiteralString] = None, 

94 msg_context: Optional[Dict[str, Any]] = None, 

95) -> ValidatorFunction: 

96 """turn validation function into a no-op, based on warning level""" 

97 

98 def wrapper(value: Any, info: ValidationInfo) -> Any: 

99 try: 

100 _call_validator_func(func, mode, value, info) 

101 except (AssertionError, ValueError) as e: 

102 issue_warning( 

103 msg or ",".join(e.args), 

104 field=info.field_name, 

105 log_depth=1, 

106 msg_context=msg_context, 

107 severity=severity, 

108 value=value, 

109 ) 

110 

111 return value 

112 

113 return wrapper 

114 

115 

116@dataclasses.dataclass(frozen=True, **SLOTS) 

117class AfterWarner(pydantic.functional_validators.AfterValidator): 

118 """Like AfterValidator, but wraps validation `func` `as_warning`""" 

119 

120 severity: WarningSeverity = WARNING 

121 msg: Optional[LiteralString] = None 

122 context: Optional[Dict[str, Any]] = None 

123 

124 def __post_init__(self): 

125 object.__setattr__( 

126 self, 

127 "func", 

128 as_warning( 

129 self.func, 

130 mode="after", 

131 severity=self.severity, 

132 msg=self.msg, 

133 msg_context=self.context, 

134 ), 

135 ) 

136 

137 

138@dataclasses.dataclass(frozen=True, **SLOTS) 

139class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

140 """Like BeforeValidator, but wraps validation `func` `as_warning`""" 

141 

142 severity: WarningSeverity = WARNING 

143 msg: Optional[LiteralString] = None 

144 context: Optional[Dict[str, Any]] = None 

145 

146 def __post_init__(self): 

147 object.__setattr__( 

148 self, 

149 "func", 

150 as_warning( 

151 self.func, 

152 mode="before", 

153 severity=self.severity, 

154 msg=self.msg, 

155 msg_context=self.context, 

156 ), 

157 ) 

158 

159 

160# TODO: add `loc: Loc` to `issue_warning()` 

161# and use a loguru handler to format warnings accordingly 

162def issue_warning( 

163 msg: LiteralString, 

164 *, 

165 value: Any, 

166 severity: WarningSeverity = WARNING, 

167 msg_context: Optional[Dict[str, Any]] = None, 

168 field: Optional[str] = None, 

169 log_depth: int = 1, 

170): 

171 msg_context = {"value": value, "severity": severity, **(msg_context or {})} 

172 

173 if severity >= (ctxt := get_validation_context()).warning_level: 

174 raise PydanticCustomError("warning", msg, msg_context) 

175 elif ctxt.log_warnings: 

176 log_msg = (field + ": " if field else "") + (msg.format(**msg_context)) 

177 logger.opt(depth=log_depth).log(severity, log_msg)