Coverage for src / bioimageio / spec / _internal / validation_context.py: 98%
85 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-08 13:04 +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 Progressbar
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."""
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`.)"""
49@dataclass(frozen=True)
50class ValidationContextSummary(ValidationContextBase):
51 """Summary of the validation context without internally used context fields."""
53 __pydantic_config__ = ConfigDict(extra="forbid")
54 """Pydantic config to include **ValdationContextSummary** in **ValidationDetail**."""
56 root: Union[RootHttpUrl, Path, Literal["in-memory"]] = Path()
59@dataclass(frozen=True)
60class ValidationContext(ValidationContextBase):
61 """A validation context used to control validation of bioimageio resources.
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 """
68 _context_tokens: "List[Token[Optional[ValidationContext]]]" = field(
69 init=False,
70 default_factory=cast(
71 "Callable[[], List[Token[Optional[ValidationContext]]]]", list
72 ),
73 )
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."""
82 root: Union[RootHttpUrl, DirectoryPath, ZipFile] = Path()
83 """Url/directory/archive serving as base to resolve any relative file paths."""
85 warning_level: WarningLevel = 50
86 """Treat warnings of severity `s` as validation errors if `s >= warning_level`."""
88 log_warnings: bool = settings.log_warnings
89 """If `True` warnings are logged to the terminal
91 Note:
92 This setting does not affect warning entries
93 of a generated `bioimageio.spec.ValidationSummary`.
94 """
96 progressbar: Union[None, bool, Callable[[], Progressbar]] = None
97 """Control any progressbar.
98 (Currently this is only used for file downloads.)
100 Can be:
101 - `None`: use a default tqdm progressbar (if not settings.CI)
102 - `True`: use a default tqdm progressbar
103 - `False`: disable the progressbar
104 - `callable`: A callable that returns a tqdm-like progressbar.
105 """
107 raise_errors: bool = False
108 """Directly raise any validation errors
109 instead of aggregating errors and returning a `bioimageio.spec.InvalidDescr`. (for debugging)"""
111 @property
112 def summary(self):
113 if isinstance(self.root, ZipFile):
114 if self.root.filename is None:
115 root = "in-memory"
116 else:
117 root = Path(self.root.filename)
118 else:
119 root = self.root
121 return ValidationContextSummary(
122 root=root,
123 file_name=self.file_name,
124 perform_io_checks=self.perform_io_checks,
125 known_files=copy(self.known_files),
126 update_hashes=self.update_hashes,
127 )
129 def __enter__(self):
130 self._context_tokens.append(_validation_context_var.set(self))
131 return self
133 def __exit__(self, type, value, traceback): # type: ignore
134 _validation_context_var.reset(self._context_tokens.pop(-1))
136 def replace( # TODO: probably use __replace__ when py>=3.13
137 self,
138 root: Optional[Union[RootHttpUrl, DirectoryPath, ZipFile]] = None,
139 warning_level: Optional[WarningLevel] = None,
140 log_warnings: Optional[bool] = None,
141 file_name: Optional[str] = None,
142 perform_io_checks: Optional[bool] = None,
143 known_files: Optional[Dict[str, Optional[Sha256]]] = None,
144 raise_errors: Optional[bool] = None,
145 update_hashes: Optional[bool] = None,
146 original_source_name: Optional[str] = None,
147 ) -> Self:
148 if known_files is None and root is not None and self.root != root:
149 # reset known files if root changes, but no new known_files are given
150 known_files = {}
152 return self.__class__(
153 root=self.root if root is None else root,
154 warning_level=(
155 self.warning_level if warning_level is None else warning_level
156 ),
157 log_warnings=self.log_warnings if log_warnings is None else log_warnings,
158 file_name=self.file_name if file_name is None else file_name,
159 perform_io_checks=(
160 self.perform_io_checks
161 if perform_io_checks is None
162 else perform_io_checks
163 ),
164 known_files=self.known_files if known_files is None else known_files,
165 raise_errors=self.raise_errors if raise_errors is None else raise_errors,
166 update_hashes=(
167 self.update_hashes if update_hashes is None else update_hashes
168 ),
169 original_source_name=(
170 self.original_source_name
171 if original_source_name is None
172 else original_source_name
173 ),
174 )
176 @property
177 def source_name(self) -> str:
178 if self.original_source_name is not None:
179 return self.original_source_name
180 elif self.file_name is None:
181 return "in-memory"
182 else:
183 try:
184 if isinstance(self.root, Path):
185 source = (self.root / self.file_name).absolute()
186 else:
187 parsed = urlsplit(str(self.root))
188 path = list(parsed.path.strip("/").split("/")) + [self.file_name]
189 source = urlunsplit(
190 (
191 parsed.scheme,
192 parsed.netloc,
193 "/".join(path),
194 parsed.query,
195 parsed.fragment,
196 )
197 )
198 except ValueError:
199 return self.file_name
200 else:
201 return str(source)
204_validation_context_var: ContextVar[Optional[ValidationContext]] = ContextVar(
205 "validation_context_var", default=None
206)
209def get_validation_context(
210 default: Optional[ValidationContext] = None,
211) -> ValidationContext:
212 """Get the currently active validation context (or a default)"""
213 return _validation_context_var.get() or default or ValidationContext()