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

1from __future__ import annotations 

2 

3import dataclasses 

4import inspect 

5from typing import TYPE_CHECKING, Any, Literal, Union, get_args 

6 

7import pydantic.functional_validators 

8from annotated_types import BaseMetadata, GroupedMetadata 

9from loguru import logger 

10from pydantic import TypeAdapter 

11from pydantic_core import PydanticCustomError 

12from pydantic_core.core_schema import ( 

13 NoInfoValidatorFunction, 

14 ValidationInfo, 

15 WithInfoValidatorFunction, 

16) 

17from typing_extensions import Annotated, LiteralString 

18 

19from .utils import SLOTS 

20from .validation_context import get_validation_context 

21from .warning_levels import WARNING, WarningSeverity 

22 

23if TYPE_CHECKING: 

24 from pydantic.functional_validators import _V2Validator # type: ignore 

25 

26 

27ValidatorFunction = Union[NoInfoValidatorFunction, WithInfoValidatorFunction] 

28 

29AnnotationMetaData = Union[BaseMetadata, GroupedMetadata] 

30 

31 

32def warn( 

33 typ: AnnotationMetaData | Any, 

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

35 severity: WarningSeverity = WARNING, 

36): 

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

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

39 typ = Annotated[Any, typ] 

40 

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

42 

43 return AfterWarner( 

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

45 ) 

46 

47 

48def _call_validator_func( 

49 func: _V2Validator, 

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

51 value: Any, 

52 info: ValidationInfo, 

53) -> Any: 

54 # determine if validator needs info arg 

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

56 sig = inspect.signature(func) 

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

58 n_positional = sum( 

59 1 

60 for param in parameters 

61 if param.kind 

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

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

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

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

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

67 ) 

68 needs_info = None 

69 if mode == "wrap": 

70 if n_positional == 3: 

71 needs_info = True 

72 elif n_positional == 2: 

73 needs_info = False 

74 else: 

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

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

77 ) 

78 if n_positional == 2: 

79 needs_info = True 

80 elif n_positional == 1: 

81 needs_info = False 

82 

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

84 if needs_info: 

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

86 else: 

87 return func(value) # type: ignore 

88 

89 

90def as_warning( 

91 func: _V2Validator, 

92 *, 

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

94 severity: WarningSeverity = WARNING, 

95 msg: LiteralString | None = None, 

96 msg_context: dict[str, Any] | None = None, 

97) -> ValidatorFunction: 

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

99 

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

101 try: 

102 _call_validator_func(func, mode, value, info) 

103 except (AssertionError, ValueError) as e: 

104 issue_warning( 

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

106 field=info.field_name, 

107 log_depth=1, 

108 msg_context=msg_context, 

109 severity=severity, 

110 value=value, 

111 ) 

112 

113 return value 

114 

115 return wrapper 

116 

117 

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

119class AfterWarner(pydantic.functional_validators.AfterValidator): 

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

121 

122 severity: WarningSeverity = WARNING 

123 msg: LiteralString | None = None 

124 context: dict[str, Any] | None = None 

125 

126 def __post_init__(self): 

127 object.__setattr__( 

128 self, 

129 "func", 

130 as_warning( 

131 self.func, 

132 mode="after", 

133 severity=self.severity, 

134 msg=self.msg, 

135 msg_context=self.context, 

136 ), 

137 ) 

138 

139 

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

141class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

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

143 

144 severity: WarningSeverity = WARNING 

145 msg: LiteralString | None = None 

146 context: dict[str, Any] | None = None 

147 

148 def __post_init__(self): 

149 object.__setattr__( 

150 self, 

151 "func", 

152 as_warning( 

153 self.func, 

154 mode="before", 

155 severity=self.severity, 

156 msg=self.msg, 

157 msg_context=self.context, 

158 ), 

159 ) 

160 

161 

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

163# and use a loguru handler to format warnings accordingly 

164def issue_warning( 

165 msg: LiteralString, 

166 *, 

167 value: Any, 

168 severity: WarningSeverity = WARNING, 

169 msg_context: dict[str, Any] | None = None, 

170 field: str | None = None, 

171 log_depth: int = 1, 

172): 

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

174 

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

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

177 elif ctxt.log_warnings: 

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

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