Coverage for src / bioimageio / spec / _internal / validation_context.py: 98%
85 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 08:44 +0000
1from __future__ import annotations
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
11from genericache import DiskCache, MemoryCache, NoopCache
12from pydantic import ConfigDict, DirectoryPath
13from typing_extensions import Self
15from ._settings import settings
16from .io_basics import FileName, Sha256
17from .progress import ProgressbarLike
18from .root_url import RootHttpUrl
19from .utils import SLOTS
20from .warning_levels import WarningLevel
23@dataclass(frozen=True, **SLOTS)
24class ValidationContextBase:
25 file_name: Optional[FileName] = None
26 """File name of the bioimageio YAML file."""
28 original_source_name: Optional[str] = None
29 """Original source of the bioimageio resource description, e.g. a URL or file path."""
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.
35 Existence of local absolute file paths is still being checked."""
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 (even if perform_io_checks is True).
45 Keys should be file paths or URL strings as they appear in the
46 bioimageio.yaml file.
48 Values are Sha256 values compared to hash values in the description.
49 For `None` values no hash value comparison is performed.
51 If `perfrom_io_checks` is True, checked files will be added to
52 this dictionary with their SHA-256 value.
54 If `perform_io_checks` is False and `known_files` is not empty,
55 missing, 'unknown' file references are considered invalid.
56 """
58 update_hashes: bool = False
59 """Overwrite specified file hashes with values computed from the referenced file (instead of comparing them).
60 (Has no effect if `perform_io_checks=False`.)"""
63@dataclass(frozen=True)
64class ValidationContextSummary(ValidationContextBase):
65 """Summary of the validation context without internally used context fields."""
67 __pydantic_config__ = ConfigDict(extra="forbid")
68 """Pydantic config to include **ValdationContextSummary** in **ValidationDetail**."""
70 root: Union[RootHttpUrl, Path, Literal["in-memory"]] = Path()
73@dataclass(frozen=True)
74class ValidationContext(ValidationContextBase):
75 """A validation context used to control validation of bioimageio resources.
77 For example a relative file path in a bioimageio description requires the **root**
78 context to evaluate if the file is available and, if **perform_io_checks** is true,
79 if it matches its expected SHA256 hash value.
80 """
82 _context_tokens: "List[Token[Optional[ValidationContext]]]" = field(
83 init=False,
84 default_factory=cast(
85 "Callable[[], List[Token[Optional[ValidationContext]]]]", list
86 ),
87 )
89 cache: Union[
90 DiskCache[RootHttpUrl], MemoryCache[RootHttpUrl], NoopCache[RootHttpUrl]
91 ] = field(default=settings.disk_cache)
92 disable_cache: bool = False
93 """Disable caching downloads to `settings.cache_path`
94 and (re)download them to memory instead."""
96 root: Union[RootHttpUrl, DirectoryPath, ZipFile] = Path()
97 """Url/directory/archive serving as base to resolve any relative file paths."""
99 warning_level: WarningLevel = 50
100 """Treat warnings of severity `s` as validation errors if `s >= warning_level`."""
102 log_warnings: bool = settings.log_warnings
103 """If `True` warnings are logged to the terminal
105 Note:
106 This setting does not affect warning entries
107 of a generated `bioimageio.spec.ValidationSummary`.
108 """
110 progressbar: Union[None, bool, Callable[[], ProgressbarLike]] = None
111 """Control any progressbar.
112 (Currently this is only used for file downloads.)
114 Can be:
115 - `None`: use a default tqdm progressbar (if not settings.CI)
116 - `True`: use a default tqdm progressbar
117 - `False`: disable the progressbar
118 - `callable`: A callable that returns a tqdm-like progressbar.
119 """
121 raise_errors: bool = False
122 """Directly raise any validation errors
123 instead of aggregating errors and returning a `bioimageio.spec.InvalidDescr`. (for debugging)"""
125 @property
126 def summary(self):
127 if isinstance(self.root, ZipFile):
128 if self.root.filename is None:
129 root = "in-memory"
130 else:
131 root = Path(self.root.filename)
132 else:
133 root = self.root
135 return ValidationContextSummary(
136 root=root,
137 file_name=self.file_name,
138 perform_io_checks=self.perform_io_checks,
139 known_files=copy(self.known_files),
140 update_hashes=self.update_hashes,
141 )
143 def __enter__(self):
144 self._context_tokens.append(_validation_context_var.set(self))
145 return self
147 def __exit__(self, type, value, traceback): # type: ignore
148 _validation_context_var.reset(self._context_tokens.pop(-1))
150 def replace( # TODO: probably use __replace__ when py>=3.13
151 self,
152 root: Optional[Union[RootHttpUrl, DirectoryPath, ZipFile]] = None,
153 warning_level: Optional[WarningLevel] = None,
154 log_warnings: Optional[bool] = None,
155 file_name: Optional[str] = None,
156 perform_io_checks: Optional[bool] = None,
157 known_files: Optional[Dict[str, Optional[Sha256]]] = None,
158 raise_errors: Optional[bool] = None,
159 update_hashes: Optional[bool] = None,
160 original_source_name: Optional[str] = None,
161 ) -> Self:
162 if known_files is None and root is not None and self.root != root:
163 # reset known files if root changes, but no new known_files are given
164 known_files = {}
166 return self.__class__(
167 root=self.root if root is None else root,
168 warning_level=(
169 self.warning_level if warning_level is None else warning_level
170 ),
171 log_warnings=self.log_warnings if log_warnings is None else log_warnings,
172 file_name=self.file_name if file_name is None else file_name,
173 perform_io_checks=(
174 self.perform_io_checks
175 if perform_io_checks is None
176 else perform_io_checks
177 ),
178 known_files=self.known_files if known_files is None else known_files,
179 raise_errors=self.raise_errors if raise_errors is None else raise_errors,
180 update_hashes=(
181 self.update_hashes if update_hashes is None else update_hashes
182 ),
183 original_source_name=(
184 self.original_source_name
185 if original_source_name is None
186 else original_source_name
187 ),
188 )
190 @property
191 def source_name(self) -> str:
192 if self.original_source_name is not None:
193 return self.original_source_name
194 elif self.file_name is None:
195 return "in-memory"
196 else:
197 try:
198 if isinstance(self.root, Path):
199 source = (self.root / self.file_name).absolute()
200 else:
201 parsed = urlsplit(str(self.root))
202 path = list(parsed.path.strip("/").split("/")) + [self.file_name]
203 source = urlunsplit(
204 (
205 parsed.scheme,
206 parsed.netloc,
207 "/".join(path),
208 parsed.query,
209 parsed.fragment,
210 )
211 )
212 except ValueError:
213 return self.file_name
214 else:
215 return str(source)
218_validation_context_var: ContextVar[Optional[ValidationContext]] = ContextVar(
219 "validation_context_var", default=None
220)
223def get_validation_context(
224 default: Optional[ValidationContext] = None,
225) -> ValidationContext:
226 """Get the currently active validation context (or a default)"""
227 return _validation_context_var.get() or default or ValidationContext()