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

1import dataclasses 

2import sys 

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._internal._decorators import inspect_validator 

10from pydantic_core import PydanticCustomError 

11from pydantic_core.core_schema import ( 

12 NoInfoValidatorFunction, 

13 ValidationInfo, 

14 WithInfoValidatorFunction, 

15) 

16from typing_extensions import Annotated, LiteralString 

17 

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 

24if sys.version_info < (3, 10): 

25 SLOTS: Dict[str, Any] = {} 

26else: 

27 SLOTS = {"slots": True} 

28 

29 

30ValidatorFunction = Union[NoInfoValidatorFunction, WithInfoValidatorFunction] 

31 

32AnnotationMetaData = Union[BaseMetadata, GroupedMetadata] 

33 

34 

35def warn( 

36 typ: Union[AnnotationMetaData, Any], 

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

38 severity: WarningSeverity = WARNING, 

39): 

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

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

42 typ = Annotated[Any, typ] 

43 

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

45 

46 return AfterWarner( 

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

48 ) 

49 

50 

51def call_validator_func( 

52 func: "_V2Validator", 

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

54 value: Any, 

55 info: ValidationInfo, 

56) -> Any: 

57 info_arg = inspect_validator(func, mode) 

58 if info_arg: 

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

60 else: 

61 return func(value) # type: ignore 

62 

63 

64def as_warning( 

65 func: "_V2Validator", 

66 *, 

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

68 severity: WarningSeverity = WARNING, 

69 msg: Optional[LiteralString] = None, 

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

71) -> ValidatorFunction: 

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

73 

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

75 try: 

76 call_validator_func(func, mode, value, info) 

77 except (AssertionError, ValueError) as e: 

78 issue_warning( 

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

80 value=value, 

81 severity=severity, 

82 msg_context=msg_context, 

83 ) 

84 

85 return value 

86 

87 return wrapper 

88 

89 

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

91class AfterWarner(pydantic.functional_validators.AfterValidator): 

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

93 

94 severity: WarningSeverity = WARNING 

95 msg: Optional[LiteralString] = None 

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

97 

98 def __post_init__(self): 

99 object.__setattr__( 

100 self, 

101 "func", 

102 as_warning( 

103 self.func, 

104 mode="after", 

105 severity=self.severity, 

106 msg=self.msg, 

107 msg_context=self.context, 

108 ), 

109 ) 

110 

111 

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

113class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

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

115 

116 severity: WarningSeverity = WARNING 

117 msg: Optional[LiteralString] = None 

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

119 

120 def __post_init__(self): 

121 object.__setattr__( 

122 self, 

123 "func", 

124 as_warning( 

125 self.func, 

126 mode="before", 

127 severity=self.severity, 

128 msg=self.msg, 

129 msg_context=self.context, 

130 ), 

131 ) 

132 

133 

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

135# and use a loguru handler to format warnings accordingly 

136def issue_warning( 

137 msg: LiteralString, 

138 *, 

139 value: Any, 

140 severity: WarningSeverity = WARNING, 

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

142 field: Optional[str] = None, 

143): 

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

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

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

147 elif ctxt.log_warnings: 

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

149 logger.opt(depth=1).log(severity, log_msg)