Coverage for bioimageio/spec/_internal/field_warning.py: 96%

1import dataclasses 

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

3 

4import pydantic.functional_validators 

5from annotated_types import BaseMetadata, GroupedMetadata 

6from loguru import logger 

7from pydantic import TypeAdapter 

8from pydantic._internal._decorators import inspect_validator 

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 info_arg = inspect_validator(func, mode) 

53 if info_arg: 

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

55 else: 

56 return func(value) # type: ignore 

57 

58 

59def as_warning( 

60 func: "_V2Validator", 

61 *, 

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

63 severity: WarningSeverity = WARNING, 

64 msg: Optional[LiteralString] = None, 

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

66) -> ValidatorFunction: 

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

68 

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

70 try: 

71 call_validator_func(func, mode, value, info) 

72 except (AssertionError, ValueError) as e: 

73 issue_warning( 

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

75 field=info.field_name, 

76 log_depth=2, 

77 msg_context=msg_context, 

78 severity=severity, 

79 value=value, 

80 ) 

81 

82 return value 

83 

84 return wrapper 

85 

86 

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

88class AfterWarner(pydantic.functional_validators.AfterValidator): 

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

90 

91 severity: WarningSeverity = WARNING 

92 msg: Optional[LiteralString] = None 

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

94 

95 def __post_init__(self): 

96 object.__setattr__( 

97 self, 

98 "func", 

99 as_warning( 

100 self.func, 

101 mode="after", 

102 severity=self.severity, 

103 msg=self.msg, 

104 msg_context=self.context, 

105 ), 

106 ) 

107 

108 

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

110class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

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

112 

113 severity: WarningSeverity = WARNING 

114 msg: Optional[LiteralString] = None 

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

116 

117 def __post_init__(self): 

118 object.__setattr__( 

119 self, 

120 "func", 

121 as_warning( 

122 self.func, 

123 mode="before", 

124 severity=self.severity, 

125 msg=self.msg, 

126 msg_context=self.context, 

127 ), 

128 ) 

129 

130 

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

132# and use a loguru handler to format warnings accordingly 

133def issue_warning( 

134 msg: LiteralString, 

135 *, 

136 value: Any, 

137 severity: WarningSeverity = WARNING, 

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

139 field: Optional[str] = None, 

140 log_depth: int = 1, 

141): 

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

143 

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

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

146 elif ctxt.log_warnings: 

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

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