bioimageio.spec.common

787class FileDescr(Node):
788    source: ImportantFileSource
789    """∈📦 file source"""
790
791    sha256: Optional[Sha256] = None
792    """SHA256 checksum of the source file"""
793
794    @model_validator(mode="after")
795    def _validate_sha256(self) -> Self:
796        if get_validation_context().perform_io_checks:
797            self.validate_sha256()
798
799        return self
800
801    def validate_sha256(self):
802        context = get_validation_context()
803        if (src_str := str(self.source)) in context.known_files:
804            actual_sha = context.known_files[src_str]
805        else:
806            local_source = download(self.source, sha256=self.sha256).path
807            actual_sha = get_sha256(local_source)
808            context.known_files[str(self.source)] = actual_sha
809
810        if self.sha256 == actual_sha:
811            pass
812        elif self.sha256 is None or context.update_hashes:
813            self.sha256 = actual_sha
814        elif self.sha256 != actual_sha:
815            raise ValueError(
816                f"Sha256 mismatch for {self.source}. Expected {self.sha256}, got "
817                + f"{actual_sha}. Update expected `sha256` or point to the matching "
818                + "file."
819            )
820
821        return
822
823    def download(self):
824
825        return download(self.source, sha256=self.sha256)

123class HttpUrl(RootHttpUrl):
124    """A URL with the HTTP or HTTPS scheme."""
125
126    root_model: ClassVar[Type[RootModel[Any]]] = RootModel[pydantic.HttpUrl]
127    _exists: Optional[bool] = None
128
129    def _after_validator(self):
130        self = super()._after_validator()
131        context = get_validation_context()
132        if (
133            context.perform_io_checks
134            and str(self._validated) not in context.known_files
135        ):
136            self._validated = _validate_url(self._validated)
137            self._exists = True
138
139        return self
140
141    def exists(self):
142        """True if URL is available"""
143        if self._exists is None:
144            try:
145                self._validated = _validate_url(self._validated)
146            except Exception as e:
147                logger.info(e)
148                self._exists = False
149            else:
150                self._exists = True
151
152        return self._exists

356class InvalidDescr(
357    ResourceDescrBase,
358    extra="allow",
359    title="An invalid resource description",
360):
361    """A representation of an invalid resource description"""
362
363    implemented_type: ClassVar[Literal["unknown"]] = "unknown"
364    if TYPE_CHECKING:  # see NodeWithExplicitlySetFields
365        type: Any = "unknown"
366    else:
367        type: Any
368
369    implemented_format_version: ClassVar[Literal["unknown"]] = "unknown"
370    if TYPE_CHECKING:  # see NodeWithExplicitlySetFields
371        format_version: Any = "unknown"
372    else:
373        format_version: Any

184class RelativeFilePath(
185    RelativePathBase[Union[AbsoluteFilePath, HttpUrl, ZipPath]], frozen=True
186):
187    """A path relative to the `rdf.yaml` file (also if the RDF source is a URL)."""
188
189    def model_post_init(self, __context: Any) -> None:
190        """add validation @private"""
191        if not self.root.parts:  # an empty path can only be a directory
192            raise ValueError(f"{self.root} is not a valid file path.")
193
194        super().model_post_init(__context)
195
196    def get_absolute(
197        self, root: "RootHttpUrl | Path | AnyUrl | ZipFile"
198    ) -> "AbsoluteFilePath | HttpUrl | ZipPath":
199        absolute = self._get_absolute_impl(root)
200        if (
201            isinstance(absolute, Path)
202            and (context := get_validation_context()).perform_io_checks
203            and str(self.root) not in context.known_files
204            and not absolute.is_file()
205        ):
206            raise ValueError(f"{absolute} does not point to an existing file")
207
208        return absolute

13class RootHttpUrl(ValidatedString):
14    """A 'URL folder', possibly an invalid HTTP URL"""
15
16    root_model: ClassVar[Type[RootModel[Any]]] = RootModel[pydantic.HttpUrl]
17    _validated: pydantic.HttpUrl
18
19    def absolute(self):
20        """analog to `absolute` method of pathlib."""
21        return self
22
23    @property
24    def scheme(self) -> str:
25        return self._validated.scheme
26
27    @property
28    def host(self) -> Optional[str]:
29        return self._validated.host
30
31    @property
32    def path(self) -> Optional[str]:
33        return self._validated.path
34
35    @property
36    def parent(self) -> RootHttpUrl:
37        parsed = urlsplit(str(self))
38        path = list(parsed.path.split("/"))
39        if (
40            parsed.netloc == "zenodo.org"
41            and parsed.path.startswith("/api/records/")
42            and parsed.path.endswith("/content")
43        ):
44            path[-2:-1] = []
45        else:
46            path = path[:-1]
47
48        return RootHttpUrl(
49            urlunsplit(
50                (
51                    parsed.scheme,
52                    parsed.netloc,
53                    "/".join(path),
54                    parsed.query,
55                    parsed.fragment,
56                )
57            )
58        )
59
60    @property
61    def parents(self) -> Iterable[RootHttpUrl]:
62        """iterate over all URL parents (max 100)"""
63        current = self
64        for _ in range(100):
65            current = current.parent
66            yield current

33class Sha256(ValidatedString):
34    """A SHA-256 hash value"""
35
36    root_model: ClassVar[Type[RootModel[Any]]] = RootModel[
37        Annotated[
38            str,
39            StringConstraints(
40                strip_whitespace=True, to_lower=True, min_length=64, max_length=64
41            ),
42        ]
43    ]