cdm_data_loaders.schemas.jsonschema ¶

def load_xsv_meta_schema(
    local_schema_path: Path | Traversable | None = None,
    remote_url: str = CUSTOM_META_SCHEMA_URI,
) -> frozendict[str, Any]:
    """Load the custom XSV metaschema and validate it against its declared metaschema.

    Reads the custom metaschema from ``schemas/xsv_config/0.1.0/schema.json`` if available,
    or from ``https://kbase.github.io/cdm-data-loaders/schemas/xsv_config/0.1.0/schema.json``
    otherwise, and validates the document against its own ``$schema`` metaschema.

    :param local_schema_path: optional local schema path or ``Traversable`` override
    :type local_schema_path: Path | Traversable | None
    :param remote_url: fallback URL to fetch the metaschema from when unavailable locally
    :type remote_url: str
    :return: validated custom metaschema as an immutable mapping
    :rtype: frozendict[str, Any]
    :raises TypeError: if the document is not a JSON object
    :raises ValueError: if the document omits ``$schema``, declares itself as its own metaschema,
        or uses an unsupported JSON Schema draft
    :raises jsonschema.exceptions.SchemaError: if the custom metaschema fails validation against its metaschema
    """
    raw_schema = _read_raw_xsv_meta_schema(local_schema_path=local_schema_path, remote_url=remote_url)
    if not isinstance(raw_schema, dict):
        err_msg = "Custom metaschema must be a JSON object"
        raise TypeError(err_msg)

    declared = raw_schema.get("$schema")
    if not isinstance(declared, str) or not declared:
        err_msg = "JSON Schema is missing the $schema keyword"
        raise ValueError(err_msg)

    if declared in {CUSTOM_META_SCHEMA_URI, remote_url}:
        err_msg = f"Custom metaschema cannot declare itself as its own $schema: {declared}"
        raise ValueError(err_msg)

    meta_validator = jsonschema.validators.validator_for(raw_schema, default=None)  # type: ignore[arg-type]
    if meta_validator is None:
        err_msg = f"Unsupported JSON Schema draft: {declared}"
        raise ValueError(err_msg)

    meta_validator.check_schema(raw_schema)
    return frozendict(raw_schema)

@cache
def get_xsv_meta_schema() -> frozendict[str, Any]:
    """Lazily load, validate, and cache the custom KBase XSV metaschema.

    :return: validated custom metaschema mapping
    :rtype: frozendict[str, Any]
    """
    return load_xsv_meta_schema()

def get_xsv_config_schema() -> frozendict[str, Any]:
    """Lazily extract the ``x-xsv-config`` extension schema from the custom XSV metaschema.

    :return: immutable ``x-xsv-config`` JSON Schema mapping
    :rtype: frozendict[str, Any]
    :raises ValueError: if the custom metaschema does not define ``properties['x-xsv-config']``
    :raises TypeError: if ``properties['x-xsv-config']`` is not a mapping
    """
    meta_schema = get_xsv_meta_schema()
    properties = meta_schema.get("properties")
    xsv_config_schema = properties.get(X_XSV_CONFIG_KEY) if isinstance(properties, Mapping) else None
    if xsv_config_schema is None:
        err_msg = f"Custom metaschema is missing '{X_XSV_CONFIG_KEY}' object schema under 'properties'"
        raise ValueError(err_msg)
    if not isinstance(xsv_config_schema, Mapping):
        err_msg = f"Custom metaschema '{X_XSV_CONFIG_KEY}' schema must be a mapping"
        raise TypeError(err_msg)
    return frozendict(xsv_config_schema)

def register_xsv_validator() -> None:
    """Lazily load, validate, and register the custom validator for the KBase XSV metaschema.

    :return: None
    :rtype: None
    """
    meta_schema = get_xsv_meta_schema()
    effective_meta_schema = (
        meta_schema
        if meta_schema.get("$id") == CUSTOM_META_SCHEMA_URI
        else frozendict({**meta_schema, "$id": CUSTOM_META_SCHEMA_URI})
    )
    existing = jsonschema.validators.validator_for({"$schema": CUSTOM_META_SCHEMA_URI}, default=None)  # type: ignore[arg-type]
    if existing is not None and getattr(existing, "META_SCHEMA", None) == effective_meta_schema:
        return
    base_validator = jsonschema.validators.validator_for({**meta_schema})
    xsv_validator = validators.extend(base_validator)
    xsv_validator.META_SCHEMA = effective_meta_schema
    validators.validates(CUSTOM_META_SCHEMA_URI)(xsv_validator)

def dereference_schema(
    schema: dict[str, Any],
    additional_resources: Mapping[str, dict[str, Any]] | None = None,
    *,
    max_ref_depth: int = 50,
    ref_sibling_mode: RefSiblingMode = "auto",
) -> dict[str, Any]:
    """Resolve every ``$ref`` in a schema using ``additional_resources`` where needed.

    Merges any ``allOf`` tags to return a fully self-contained schema.

    :param schema: the JSON Schema document to dereference
    :type schema: dict[str, Any]
    :param additional_resources: mapping of URI to schema dict, used to resolve external ``$ref``s
    :type additional_resources: Mapping[str, dict[str, Any]] | None
    :param max_ref_depth: maximum depth of a ``$ref`` chain before giving up
    :type max_ref_depth: int
    :param ref_sibling_mode: how to handle keywords declared alongside ``$ref``:
        ``'auto'`` (spec-compliant per detected draft), ``'always'`` (permissive), or
        ``'never'`` (strict pre-2019-09 semantics)
    :type ref_sibling_mode: RefSiblingMode
    :return: an equivalent, ``$ref``/``allOf``-free JSON Schema document
    :rtype: dict[str, Any]
    :raises DereferencingError: if a ``$ref`` is unresolvable, circular, or exceeds ``max_ref_depth``,
        or if a dialect cannot be determined for the root schema or an external resource
    """
    additional_resources = additional_resources or {}

    dialect_id = schema.get("$schema")
    spec = specification_with(dialect_id, default=DRAFT202012) if dialect_id else DRAFT202012

    try:
        root_resource = Resource.from_contents(schema, default_specification=spec)
    except CannotDetermineSpecification as exc:
        err_msg = f"Could not determine a JSON Schema dialect for the root schema: {exc}"
        raise DereferencingError(err_msg) from exc

    try:
        resources = [
            (uri, Resource.from_contents(doc, default_specification=spec)) for uri, doc in additional_resources.items()
        ]
    except CannotDetermineSpecification as exc:
        err_msg = f"Could not determine a JSON Schema dialect for an external resource in `additional_resources`: {exc}"
        raise DereferencingError(err_msg) from exc

    registry: Registry = Registry().with_resources(resources)
    resolver: Resolver[Any] = registry.resolver_with_root(root_resource)

    ctx = DereferenceContext(
        resolver=resolver,
        spec=spec,
        max_ref_depth=max_ref_depth,
        ref_sibling_mode=ref_sibling_mode,
    )
    return _dereference(schema, ctx)  # type: ignore[return-value]