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 

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 LiteralString, TypeForm 

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: TypeForm[Any], 

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

35 severity: WarningSeverity = WARNING, 

36): 

37 """treat a type validation as a warning""" 

38 validator = TypeAdapter(typ) 

39 

40 return AfterWarner( 

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

42 ) 

43 

44 

45def _call_validator_func( 

46 func: _V2Validator, 

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

48 value: Any, 

49 info: ValidationInfo, 

50) -> Any: 

51 # determine if validator needs info arg 

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

53 sig = inspect.signature(func) 

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

55 n_positional = sum( 

56 1 

57 for param in parameters 

58 if param.kind 

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

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

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

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

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

64 ) 

65 needs_info = None 

66 if mode == "wrap": 

67 if n_positional == 3: 

68 needs_info = True 

69 elif n_positional == 2: 

70 needs_info = False 

71 else: 

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

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

74 ) 

75 if n_positional == 2: 

76 needs_info = True 

77 elif n_positional == 1: 

78 needs_info = False 

79 

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

81 if needs_info: 

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

83 else: 

84 return func(value) # type: ignore 

85 

86 

87def as_warning( 

88 func: _V2Validator, 

89 *, 

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

91 severity: WarningSeverity = WARNING, 

92 msg: LiteralString | None = None, 

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

94) -> ValidatorFunction: 

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

96 

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

98 try: 

99 _call_validator_func(func, mode, value, info) 

100 except (AssertionError, ValueError) as e: 

101 issue_warning( 

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

103 field=info.field_name, 

104 log_depth=1, 

105 msg_context=msg_context, 

106 severity=severity, 

107 value=value, 

108 ) 

109 

110 return value 

111 

112 return wrapper 

113 

114 

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

116class AfterWarner(pydantic.functional_validators.AfterValidator): 

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

118 

119 severity: WarningSeverity = WARNING 

120 msg: LiteralString | None = None 

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

122 

123 def __post_init__(self): 

124 object.__setattr__( 

125 self, 

126 "func", 

127 as_warning( 

128 self.func, 

129 mode="after", 

130 severity=self.severity, 

131 msg=self.msg, 

132 msg_context=self.context, 

133 ), 

134 ) 

135 

136 

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

138class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

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

140 

141 severity: WarningSeverity = WARNING 

142 msg: LiteralString | None = None 

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

144 

145 def __post_init__(self): 

146 object.__setattr__( 

147 self, 

148 "func", 

149 as_warning( 

150 self.func, 

151 mode="before", 

152 severity=self.severity, 

153 msg=self.msg, 

154 msg_context=self.context, 

155 ), 

156 ) 

157 

158 

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

160# and use a loguru handler to format warnings accordingly 

161def issue_warning( 

162 msg: LiteralString, 

163 *, 

164 value: Any, 

165 severity: WarningSeverity = WARNING, 

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

167 field: str | None = None, 

168 log_depth: int = 1, 

169): 

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

171 

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

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

174 elif ctxt.log_warnings: 

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

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