Coverage for bioimageio/spec/_internal/validation_context.py: 98%

1from __future__ import annotations 

2 

3from contextvars import ContextVar, Token 

4from copy import copy 

5from dataclasses import dataclass, field 

6from pathlib import Path 

7from typing import Callable, Dict, List, Literal, Optional, Union, cast 

8from urllib.parse import urlsplit, urlunsplit 

9from zipfile import ZipFile 

10 

11from genericache import DiskCache, MemoryCache, NoopCache 

12from pydantic import ConfigDict, DirectoryPath 

13from typing_extensions import Self 

14 

15from ._settings import settings 

16from .io_basics import FileName, Sha256 

17from .progress import Progressbar 

18from .root_url import RootHttpUrl 

19from .utils import SLOTS 

20from .warning_levels import WarningLevel 

21 

22 

23@dataclass(frozen=True, **SLOTS) 

24class ValidationContextBase: 

25 file_name: Optional[FileName] = None 

26 """File name of the bioimageio Yaml file.""" 

27 

28 perform_io_checks: bool = settings.perform_io_checks 

29 """Wether or not to perform validation that requires file io, 

30 e.g. downloading a remote files. 

31 

32 Existence of local absolute file paths is still being checked.""" 

33 

34 known_files: Dict[str, Optional[Sha256]] = field( 

35 default_factory=cast( # TODO: (py>3.8) use dict[str, Optional[Sha256]] 

36 Callable[[], Dict[str, Optional[Sha256]]], dict 

37 ) 

38 ) 

39 """Allows to bypass download and hashing of referenced files.""" 

40 

41 update_hashes: bool = False 

42 """Overwrite specified file hashes with values computed from the referenced file (instead of comparing them). 

43 (Has no effect if `perform_io_checks=False`.)""" 

44 

45 

46@dataclass(frozen=True) 

47class ValidationContextSummary(ValidationContextBase): 

48 """Summary of the validation context without internally used context fields.""" 

49 

50 __pydantic_config__ = ConfigDict(extra="forbid") 

51 """Pydantic config to include **ValdationContextSummary** in **ValidationDetail**.""" 

52 

53 root: Union[RootHttpUrl, Path, Literal["in-memory"]] = Path() 

54 

55 

56@dataclass(frozen=True) 

57class ValidationContext(ValidationContextBase): 

58 """A validation context used to control validation of bioimageio resources. 

59 

60 For example a relative file path in a bioimageio description requires the **root** 

61 context to evaluate if the file is available and, if **perform_io_checks** is true, 

62 if it matches its expected SHA256 hash value. 

63 """ 

64 

65 _context_tokens: "List[Token[Optional[ValidationContext]]]" = field( 

66 init=False, 

67 default_factory=cast( 

68 "Callable[[], List[Token[Optional[ValidationContext]]]]", list 

69 ), 

70 ) 

71 

72 cache: Union[ 

73 DiskCache[RootHttpUrl], MemoryCache[RootHttpUrl], NoopCache[RootHttpUrl] 

74 ] = field(default=settings.disk_cache) 

75 disable_cache: bool = False 

76 """Disable caching downloads to `settings.cache_path` 

77 and (re)download them to memory instead.""" 

78 

79 root: Union[RootHttpUrl, DirectoryPath, ZipFile] = Path() 

80 """Url/directory/archive serving as base to resolve any relative file paths.""" 

81 

82 warning_level: WarningLevel = 50 

83 """Treat warnings of severity `s` as validation errors if `s >= warning_level`.""" 

84 

85 log_warnings: bool = settings.log_warnings 

86 """If `True` warnings are logged to the terminal 

87 

88 Note: This setting does not affect warning entries 

89 of a generated `bioimageio.spec.ValidationSummary`. 

90 """ 

91 

92 progressbar_factory: Optional[Callable[[], Progressbar]] = None 

93 """Callable to return a tqdm-like progressbar. 

94 

95 Currently this is only used for file downloads.""" 

96 

97 raise_errors: bool = False 

98 """Directly raise any validation errors 

99 instead of aggregating errors and returning a `bioimageio.spec.InvalidDescr`. (for debugging)""" 

100 

101 @property 

102 def summary(self): 

103 if isinstance(self.root, ZipFile): 

104 if self.root.filename is None: 

105 root = "in-memory" 

106 else: 

107 root = Path(self.root.filename) 

108 else: 

109 root = self.root 

110 

111 return ValidationContextSummary( 

112 root=root, 

113 file_name=self.file_name, 

114 perform_io_checks=self.perform_io_checks, 

115 known_files=copy(self.known_files), 

116 update_hashes=self.update_hashes, 

117 ) 

118 

119 def __enter__(self): 

120 self._context_tokens.append(_validation_context_var.set(self)) 

121 return self 

122 

123 def __exit__(self, type, value, traceback): # type: ignore 

124 _validation_context_var.reset(self._context_tokens.pop(-1)) 

125 

126 def replace( # TODO: probably use __replace__ when py>=3.13 

127 self, 

128 root: Optional[Union[RootHttpUrl, DirectoryPath, ZipFile]] = None, 

129 warning_level: Optional[WarningLevel] = None, 

130 log_warnings: Optional[bool] = None, 

131 file_name: Optional[str] = None, 

132 perform_io_checks: Optional[bool] = None, 

133 known_files: Optional[Dict[str, Optional[Sha256]]] = None, 

134 raise_errors: Optional[bool] = None, 

135 update_hashes: Optional[bool] = None, 

136 ) -> Self: 

137 if known_files is None and root is not None and self.root != root: 

138 # reset known files if root changes, but no new known_files are given 

139 known_files = {} 

140 

141 return self.__class__( 

142 root=self.root if root is None else root, 

143 warning_level=( 

144 self.warning_level if warning_level is None else warning_level 

145 ), 

146 log_warnings=self.log_warnings if log_warnings is None else log_warnings, 

147 file_name=self.file_name if file_name is None else file_name, 

148 perform_io_checks=( 

149 self.perform_io_checks 

150 if perform_io_checks is None 

151 else perform_io_checks 

152 ), 

153 known_files=self.known_files if known_files is None else known_files, 

154 raise_errors=self.raise_errors if raise_errors is None else raise_errors, 

155 update_hashes=( 

156 self.update_hashes if update_hashes is None else update_hashes 

157 ), 

158 ) 

159 

160 @property 

161 def source_name(self) -> str: 

162 if self.file_name is None: 

163 return "in-memory" 

164 else: 

165 try: 

166 if isinstance(self.root, Path): 

167 source = (self.root / self.file_name).absolute() 

168 else: 

169 parsed = urlsplit(str(self.root)) 

170 path = list(parsed.path.strip("/").split("/")) + [self.file_name] 

171 source = urlunsplit( 

172 ( 

173 parsed.scheme, 

174 parsed.netloc, 

175 "/".join(path), 

176 parsed.query, 

177 parsed.fragment, 

178 ) 

179 ) 

180 except ValueError: 

181 return self.file_name 

182 else: 

183 return str(source) 

184 

185 

186_validation_context_var: ContextVar[Optional[ValidationContext]] = ContextVar( 

187 "validation_context_var", default=None 

188) 

189 

190 

191def get_validation_context( 

192 default: Optional[ValidationContext] = None, 

193) -> ValidationContext: 

194 """Get the currently active validation context (or a default)""" 

195 return _validation_context_var.get() or default or ValidationContext()