from __future__ import annotations
import warnings
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, cast
import pystac
from pystac.serialization.identify import STACVersionID, identify_stac_object
from pystac.stac_object import STACObjectType
from pystac.utils import make_absolute_href
from pystac.validation.schema_uri_map import OldExtensionSchemaUriMap
from pystac.validation.stac_validator import GetSchemaError
if TYPE_CHECKING:
from pystac.stac_object import STACObject
# Import after above class definition
from pystac.validation.stac_validator import JsonSchemaSTACValidator, STACValidator
__all__ = [
"GetSchemaError",
"JsonSchemaSTACValidator",
"RegisteredValidator",
"validate",
"validate_all",
"validate_dict",
"validate_all_dict",
"set_validator",
]
[docs]
def validate(
stac_object: STACObject,
validator: STACValidator | None = None,
) -> list[Any]:
"""Validates a :class:`~pystac.STACObject`.
Args:
stac_object : The stac object to validate.
validator : A custom validator to use for validation of the STAC object.
If omitted, the default validator from
:class:`~pystac.validation.RegisteredValidator`
will be used instead.
Returns:
List[Any]: List of return values from the validation calls for the
core object and any extensions. Element type is specific to the
STACValidator implementation.
Raises:
STACValidationError
"""
return validate_dict(
stac_dict=stac_object.to_dict(),
stac_object_type=stac_object.STAC_OBJECT_TYPE,
stac_version=pystac.get_stac_version(),
extensions=stac_object.stac_extensions,
href=stac_object.get_self_href(),
validator=validator,
)
[docs]
def validate_dict(
stac_dict: dict[str, Any],
stac_object_type: STACObjectType | None = None,
stac_version: str | None = None,
extensions: list[str] | None = None,
href: str | None = None,
validator: STACValidator | None = None,
) -> list[Any]:
"""Validate a stac object serialized as JSON into a dict.
This method delegates to the call to
:meth:`~pystac.validation.stac_validator.STACValidator.validate` for the
:class:`~pystac.validation.stac_validator.STACValidator` registered
via :func:`~pystac.validation.set_validator` or
:class:`~pystac.validation.JsonSchemaSTACValidator` by default.
Args:
stac_dict : Dictionary that is the STAC json of the object.
stac_object_type : The stac object type of the object encoded in stac_dict.
One of :class:`~pystac.STACObjectType`. If not supplied, this will use
PySTAC's identification logic to identify the object type.
stac_version : The version of STAC to validate the object against. If not
supplied, this will use PySTAC's identification logic to identify the stac
version
extensions : Extension IDs for this stac object. If not supplied,
PySTAC's identification logic to identify the extensions.
href : Optional HREF of the STAC object being validated.
validator : A custom validator to use for validation of the STAC dictionary.
If omitted, the default validator from
:class:`~pystac.validation.RegisteredValidator`
will be used instead.
Returns:
List[Any]: List of return values from the validation calls for the
core object and any extensions. Element type is specific to the
STACValidator implementation.
Raises:
STACValidationError
"""
info = None
if stac_object_type is None:
info = identify_stac_object(stac_dict)
stac_object_type = info.object_type
if stac_version is None:
if info is None:
info = identify_stac_object(stac_dict)
stac_version = str(info.version_range.latest_valid_version())
if extensions is None:
if info is None:
info = identify_stac_object(stac_dict)
extensions = list(info.extensions)
stac_version_id = STACVersionID(stac_version)
# If the version is before 1.0.0-rc.2, substitute extension short IDs for
# their schemas.
if stac_version_id < "1.0.0-rc.2":
def _get_uri(ext: str) -> str | None:
return OldExtensionSchemaUriMap.get_extension_schema_uri(
ext,
stac_object_type,
stac_version_id,
)
extensions = [uri for uri in map(_get_uri, extensions) if uri is not None]
validator = validator or RegisteredValidator.get_validator()
return validator.validate(
stac_dict, stac_object_type, stac_version, extensions, href
)
[docs]
def validate_all(
stac_object: STACObject | dict[str, Any],
href: str | None = None,
stac_io: pystac.StacIO | None = None,
) -> None:
"""Validate a :class:`~pystac.STACObject`, or a STAC object serialized as
JSON into a dict.
If the STAC object represents a catalog or collection, this function will be
called recursively for each child link and all contained items.
Args:
stac_object : STAC object to validate
href : HREF of the STAC object being validated. Used for error
reporting and resolving relative links.
stac_io : Optional StacIO instance to use for reading hrefs. If None,
the StacIO.default() instance is used.
Raises:
STACValidationError or ValueError: STACValidationError is raised
if the STAC object or any contained catalog, collection,
or item has a validation error. ValueError is raised
if stac_object is a STACObject and href is not None, or if
stac_object is a dict and href is None.
"""
if stac_io is None:
stac_io = pystac.StacIO.default()
if isinstance(stac_object, dict):
warnings.warn(
"validating a STAC object as a dict is deprecated;"
" use validate_all_dict instead",
DeprecationWarning,
)
if href is None:
raise ValueError(
"href must be set if stac_object is a dict; it will be removed"
" once support for validating a STAC object as a dict is removed from"
" validate_all, due to the introduction of validate_all_dict"
)
validate_all_dict(stac_object, href, stac_io)
elif href is not None:
raise ValueError(
"href must be None if stac_object is a STACObject; it will be removed"
" once support for validating a STAC object as a dict is removed from"
" validate_all, due to the introduction of validate_all_dict"
)
else:
validate_all_dict(stac_object.to_dict(), stac_object.get_self_href(), stac_io)
[docs]
def validate_all_dict(
stac_dict: dict[str, Any],
href: str | None,
stac_io: pystac.StacIO | None = None,
) -> None:
"""Validate a stac object serialized as JSON into a dict.
If the STAC object represents a catalog or collection, this function will be
called recursively for each child link and all contained items.
Args:
stac_dict : Dictionary that is the STAC json of the object.
href : HREF of the STAC object being validated. Used for error
reporting and resolving relative links.
stac_io : Optional StacIO instance to use for reading hrefs. If None,
the StacIO.default() instance is used.
Raises:
STACValidationError: if the STAC object or any contained catalog, collection,
or item has a validation error
"""
if stac_io is None:
stac_io = pystac.StacIO.default()
info = identify_stac_object(stac_dict)
# Validate this object
validate_dict(
stac_dict,
stac_object_type=info.object_type,
stac_version=str(info.version_range.latest_valid_version()),
extensions=list(info.extensions),
href=href,
)
if info.object_type != pystac.STACObjectType.ITEM and "links" in stac_dict:
links = (
# Account for 0.6 links
stac_dict["links"].values()
if isinstance(stac_dict["links"], dict)
else stac_dict.get("links")
)
for link in cast(Iterable[Mapping[str, Any]], links):
if link.get("rel") in [pystac.RelType.ITEM, pystac.RelType.CHILD]:
link_href = make_absolute_href(
cast(str, link.get("href")), start_href=href
)
validate_all_dict(stac_io.read_json(link_href), link_href, stac_io)
[docs]
class RegisteredValidator:
_validator: STACValidator | None = None
[docs]
@classmethod
def get_validator(cls) -> STACValidator:
if cls._validator is None:
try:
import jsonschema as _ # noqa
except ImportError:
raise Exception(
"Cannot validate with default validator because package"
' "jsonschema" is not installed. Install pystac with the validation'
" optional requirements (e.g. pip install pystac[validation]) to"
" install jsonschema"
)
cls._validator = JsonSchemaSTACValidator()
return cls._validator
[docs]
@classmethod
def set_validator(cls, validator: STACValidator) -> None:
if not issubclass(type(validator), STACValidator):
raise Exception(f"Validator must be a subclass of {STACValidator}")
cls._validator = validator
[docs]
def set_validator(validator: STACValidator) -> None:
"""Sets the STACValidator to use in PySTAC.
Args:
validator : The STACValidator implementation to use for
validation.
"""
RegisteredValidator.set_validator(validator)