Coverage for src/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 original_source_name: Optional[str] = None 

29 """Original source of the bioimageio resource description, e.g. a URL or file path.""" 

30 

31 perform_io_checks: bool = settings.perform_io_checks 

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

33 e.g. downloading a remote files. 

34 

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

36 

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

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

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

40 ) 

41 ) 

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

43 

44 update_hashes: bool = False 

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

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

47 

48 

49@dataclass(frozen=True) 

50class ValidationContextSummary(ValidationContextBase): 

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

52 

53 __pydantic_config__ = ConfigDict(extra="forbid") 

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

55 

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

57 

58 

59@dataclass(frozen=True) 

60class ValidationContext(ValidationContextBase): 

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

62 

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

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

65 if it matches its expected SHA256 hash value. 

66 """ 

67 

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

69 init=False, 

70 default_factory=cast( 

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

72 ), 

73 ) 

74 

75 cache: Union[ 

76 DiskCache[RootHttpUrl], MemoryCache[RootHttpUrl], NoopCache[RootHttpUrl] 

77 ] = field(default=settings.disk_cache) 

78 disable_cache: bool = False 

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

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

81 

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

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

84 

85 warning_level: WarningLevel = 50 

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

87 

88 log_warnings: bool = settings.log_warnings 

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

90 

91 Note: This setting does not affect warning entries 

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

93 """ 

94 

95 progressbar: Union[None, bool, Callable[[], Progressbar]] = None 

96 """Control any progressbar. 

97 (Currently this is only used for file downloads.) 

98 

99 Can be: 

100 - `None`: use a default tqdm progressbar (if not settings.CI) 

101 - `True`: use a default tqdm progressbar 

102 - `False`: disable the progressbar 

103 - `callable`: A callable that returns a tqdm-like progressbar. 

104 """ 

105 

106 raise_errors: bool = False 

107 """Directly raise any validation errors 

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

109 

110 @property 

111 def summary(self): 

112 if isinstance(self.root, ZipFile): 

113 if self.root.filename is None: 

114 root = "in-memory" 

115 else: 

116 root = Path(self.root.filename) 

117 else: 

118 root = self.root 

119 

120 return ValidationContextSummary( 

121 root=root, 

122 file_name=self.file_name, 

123 perform_io_checks=self.perform_io_checks, 

124 known_files=copy(self.known_files), 

125 update_hashes=self.update_hashes, 

126 ) 

127 

128 def __enter__(self): 

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

130 return self 

131 

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

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

134 

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

136 self, 

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

138 warning_level: Optional[WarningLevel] = None, 

139 log_warnings: Optional[bool] = None, 

140 file_name: Optional[str] = None, 

141 perform_io_checks: Optional[bool] = None, 

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

143 raise_errors: Optional[bool] = None, 

144 update_hashes: Optional[bool] = None, 

145 original_source_name: Optional[str] = None, 

146 ) -> Self: 

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

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

149 known_files = {} 

150 

151 return self.__class__( 

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

153 warning_level=( 

154 self.warning_level if warning_level is None else warning_level 

155 ), 

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

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

158 perform_io_checks=( 

159 self.perform_io_checks 

160 if perform_io_checks is None 

161 else perform_io_checks 

162 ), 

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

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

165 update_hashes=( 

166 self.update_hashes if update_hashes is None else update_hashes 

167 ), 

168 original_source_name=( 

169 self.original_source_name 

170 if original_source_name is None 

171 else original_source_name 

172 ), 

173 ) 

174 

175 @property 

176 def source_name(self) -> str: 

177 if self.original_source_name is not None: 

178 return self.original_source_name 

179 elif self.file_name is None: 

180 return "in-memory" 

181 else: 

182 try: 

183 if isinstance(self.root, Path): 

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

185 else: 

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

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

188 source = urlunsplit( 

189 ( 

190 parsed.scheme, 

191 parsed.netloc, 

192 "/".join(path), 

193 parsed.query, 

194 parsed.fragment, 

195 ) 

196 ) 

197 except ValueError: 

198 return self.file_name 

199 else: 

200 return str(source) 

201 

202 

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

204 "validation_context_var", default=None 

205) 

206 

207 

208def get_validation_context( 

209 default: Optional[ValidationContext] = None, 

210) -> ValidationContext: 

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

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