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 value=value, 

76 severity=severity, 

77 msg_context=msg_context, 

78 ) 

79 

80 return value 

81 

82 return wrapper 

83 

84 

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

86class AfterWarner(pydantic.functional_validators.AfterValidator): 

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

88 

89 severity: WarningSeverity = WARNING 

90 msg: Optional[LiteralString] = None 

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

92 

93 def __post_init__(self): 

94 object.__setattr__( 

95 self, 

96 "func", 

97 as_warning( 

98 self.func, 

99 mode="after", 

100 severity=self.severity, 

101 msg=self.msg, 

102 msg_context=self.context, 

103 ), 

104 ) 

105 

106 

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

108class BeforeWarner(pydantic.functional_validators.BeforeValidator): 

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

110 

111 severity: WarningSeverity = WARNING 

112 msg: Optional[LiteralString] = None 

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

114 

115 def __post_init__(self): 

116 object.__setattr__( 

117 self, 

118 "func", 

119 as_warning( 

120 self.func, 

121 mode="before", 

122 severity=self.severity, 

123 msg=self.msg, 

124 msg_context=self.context, 

125 ), 

126 ) 

127 

128 

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

130# and use a loguru handler to format warnings accordingly 

131def issue_warning( 

132 msg: LiteralString, 

133 *, 

134 value: Any, 

135 severity: WarningSeverity = WARNING, 

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

137 field: Optional[str] = None, 

138): 

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

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

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

142 elif ctxt.log_warnings: 

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

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