Coverage for bioimageio/spec/_internal/validation_context.py: 96%
75 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-02 14:21 +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 Dict, List, Literal, Optional, Union
8from urllib.parse import urlsplit, urlunsplit
9from zipfile import ZipFile
11from pydantic import ConfigDict, DirectoryPath
12from typing_extensions import Self
14from ._settings import settings
15from .io_basics import FileName, Sha256
16from .root_url import RootHttpUrl
17from .warning_levels import WarningLevel
20@dataclass(frozen=True)
21class ValidationContextBase:
22 file_name: Optional[FileName] = None
23 """File name of the bioimageio Yaml file."""
25 perform_io_checks: bool = settings.perform_io_checks
26 """Wether or not to perform validation that requires file io,
27 e.g. downloading a remote files.
29 Existence of local absolute file paths is still being checked."""
31 known_files: Dict[str, Optional[Sha256]] = field(default_factory=dict)
32 """Allows to bypass download and hashing of referenced files."""
34 update_hashes: bool = False
35 """Overwrite specified file hashes with values computed from the referenced file (instead of comparing them).
36 (Has no effect if `perform_io_checks=False`.)"""
39@dataclass(frozen=True)
40class ValidationContextSummary(ValidationContextBase):
41 """Summary of the validation context without internally used context fields."""
43 __pydantic_config__ = ConfigDict(extra="forbid")
44 """Pydantic config to include **ValdationContextSummary** in **ValidationDetail**."""
46 root: Union[RootHttpUrl, Path, Literal["in-memory"]] = Path()
49@dataclass(frozen=True)
50class ValidationContext(ValidationContextBase):
51 """A validation context used to control validation of bioimageio resources.
53 For example a relative file path in a bioimageio description requires the **root**
54 context to evaluate if the file is available and, if **perform_io_checks** is true,
55 if it matches its expected SHA256 hash value.
56 """
58 _context_tokens: "List[Token[Optional[ValidationContext]]]" = field(
59 init=False, default_factory=list
60 )
62 disable_cache: bool = False
63 """Disable caching downloads to `settings.cache_path`
64 and (re)download them to memory instead."""
66 root: Union[RootHttpUrl, DirectoryPath, ZipFile] = Path()
67 """Url/directory/archive serving as base to resolve any relative file paths."""
69 warning_level: WarningLevel = 50
70 """Treat warnings of severity `s` as validation errors if `s >= warning_level`."""
72 log_warnings: bool = settings.log_warnings
73 """If `True` warnings are logged to the terminal
75 Note: This setting does not affect warning entries
76 of a generated `bioimageio.spec.ValidationSummary`.
77 """
79 raise_errors: bool = False
80 """Directly raise any validation errors
81 instead of aggregating errors and returning a `bioimageio.spec.InvalidDescr`. (for debugging)"""
83 @property
84 def summary(self):
85 if isinstance(self.root, ZipFile):
86 if self.root.filename is None:
87 root = "in-memory"
88 else:
89 root = Path(self.root.filename)
90 else:
91 root = self.root
93 return ValidationContextSummary(
94 root=root,
95 file_name=self.file_name,
96 perform_io_checks=self.perform_io_checks,
97 known_files=copy(self.known_files),
98 update_hashes=self.update_hashes,
99 )
101 def __enter__(self):
102 self._context_tokens.append(_validation_context_var.set(self))
103 return self
105 def __exit__(self, type, value, traceback): # type: ignore
106 _validation_context_var.reset(self._context_tokens.pop(-1))
108 def replace( # TODO: probably use __replace__ when py>=3.13
109 self,
110 root: Optional[Union[RootHttpUrl, DirectoryPath, ZipFile]] = None,
111 warning_level: Optional[WarningLevel] = None,
112 log_warnings: Optional[bool] = None,
113 file_name: Optional[str] = None,
114 perform_io_checks: Optional[bool] = None,
115 known_files: Optional[Dict[str, Optional[Sha256]]] = None,
116 raise_errors: Optional[bool] = None,
117 update_hashes: Optional[bool] = None,
118 ) -> Self:
119 if known_files is None and root is not None and self.root != root:
120 # reset known files if root changes, but no new known_files are given
121 known_files = {}
123 return self.__class__(
124 root=self.root if root is None else root,
125 warning_level=(
126 self.warning_level if warning_level is None else warning_level
127 ),
128 log_warnings=self.log_warnings if log_warnings is None else log_warnings,
129 file_name=self.file_name if file_name is None else file_name,
130 perform_io_checks=(
131 self.perform_io_checks
132 if perform_io_checks is None
133 else perform_io_checks
134 ),
135 known_files=self.known_files if known_files is None else known_files,
136 raise_errors=self.raise_errors if raise_errors is None else raise_errors,
137 update_hashes=(
138 self.update_hashes if update_hashes is None else update_hashes
139 ),
140 )
142 @property
143 def source_name(self) -> str:
144 if self.file_name is None:
145 return "in-memory"
146 else:
147 try:
148 if isinstance(self.root, Path):
149 source = (self.root / self.file_name).absolute()
150 else:
151 parsed = urlsplit(str(self.root))
152 path = list(parsed.path.strip("/").split("/")) + [self.file_name]
153 source = urlunsplit(
154 (
155 parsed.scheme,
156 parsed.netloc,
157 "/".join(path),
158 parsed.query,
159 parsed.fragment,
160 )
161 )
162 except ValueError:
163 return self.file_name
164 else:
165 return str(source)
168_validation_context_var: ContextVar[Optional[ValidationContext]] = ContextVar(
169 "validation_context_var", default=None
170)
173def get_validation_context(
174 default: Optional[ValidationContext] = None,
175) -> ValidationContext:
176 """Get the currently active validation context (or a default)"""
177 return _validation_context_var.get() or default or ValidationContext()