Source code for gemseo.core.grammars.json_schema

# Copyright 2021 IRT Saint Exupéry,
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
"""JSON schema handler."""
from __future__ import annotations

from collections import abc
from contextlib import contextmanager
from typing import Any
from typing import ClassVar
from typing import Iterable
from typing import Iterator
from typing import Mapping

from genson import SchemaBuilder
from genson import SchemaNode
from genson.schema.builder import _MetaSchemaBuilder
from genson.schema.strategies import Number
from genson.schema.strategies import Object
from numpy import float64
from numpy import int64

class _MergeRequiredStrategy(Object):
    """A genson Object with a modified required attribute handling.

    Genson Object does not merge the required attribute on purpose.
    See :ref:``.
    This class will merge the required attributes.

    By default, genson merges nodes, the update is triggered via a class attribute.

    # Do not merge the name and id properties.
    KEYWORDS = Object.KEYWORDS + ("name", "id")

    update: ClassVar[bool] = False
    """Whether to update or merge the schema."""

    def __handle_update(
        self, updated_names: Iterable[str], required: Iterable[str]
    ) -> None:
        """A context manager to handle the update vs merge.

            updated_names: All the names to update.
            required: The required names to add.
        # Pass the update switch to _SchemaNode.
        self.node_class.update = self.update

        if not self._required or not required:
            # Backup the current required before updating it with the new ones.
            _required = set(self._required)
            if self.update:
                # The elements we update from overrule the existing ones, the required
                # or not state for all the elements we update shall be reset.
                _required -= updated_names
            self._required = _required | set(required)

        # Reset to the merge behavior because _SchemaNode may be used by other instances
        # that should merge.
        self.node_class.update = False

    def add_schema(self, schema: Mapping[str, Any]) -> None:
        """Add a schema and merge the required attribute.

            schema: A schema to be added.
        with self.__handle_update(
            schema.get("properties", {}).keys(), schema.get("required")

    def add_object(self, obj: Mapping[str, Any]) -> None:
        """Add an object and merge the required attribute.

            obj: An object to be added.
        with self.__handle_update(obj.keys(), obj.keys()):

class _SchemaNode(SchemaNode):
    """Overload :meth:`.add_schema` and :meth:`.add_object` to allow updating.

    By default, genson merges nodes, the update is triggered via a class attribute.

    update: ClassVar[bool] = False
    """Whether to update or merge the schema."""

    def add_schema(self, schema) -> None:

    def add_object(self, obj) -> None:

    def __handle_update(self) -> None:
        """Handle the update or merge behavior.

        When updating, the already existing active strategies are removed such that only
        the last one added remains.
        if self.update and self._active_strategies:

class _MultipleMeta(type(abc.Mapping), _MetaSchemaBuilder):
    """Required meta class for inheriting from multiple classes with meta classes.

    Also fix the ``NODE_CLASS`` overloading because it does not use the ``NODE_CLASS``
    passed to a class derived from ``SchemaBuilder``.

    def __init__(self, name: str, bases: tuple(type), attrs: dict[str, Any]) -> None:
        super().__init__(name, bases, attrs)
        self.NODE_CLASS = type(
            "%sSchemaNode" % name, (_SchemaNode,), {"STRATEGIES": self.STRATEGIES}

class _Number(Number):
    """A number strategy that handles numpy data."""

    PYTHON_TYPES = Number.PYTHON_TYPES + (float64, int64)

[docs]class MutableMappingSchemaBuilder(abc.Mapping, SchemaBuilder, metaclass=_MultipleMeta): """A mutable genson SchemaBuilder with a dictionary-like interface. The :class:`SchemaBuilder` does not provide a way to mutate directly the properties of a schema (these are stored deeply). For ease of usage, this class brings the properties closer to the surface, and the mutability is only provided by the ability to delete a property. """ EXTRA_STRATEGIES = (_MergeRequiredStrategy, _Number) NODE_CLASS = _SchemaNode def __getitem__(self, key: str) -> dict[str, Any]: self.check_property_names(key) return[key] def __iter__(self) -> Iterator[str]: return iter( def __len__(self) -> int: return len( def __delitem__(self, key: str) -> None: del[key] if key in self.required: self.required.remove(key) @property def properties(self) -> dict[str, Any]: """Return the properties. Returns: The existing properties, otherwise an empty dictionary. """ try: return self._root_node._active_strategies[0]._properties except (AttributeError, IndexError): return {} @property def required(self) -> set[str]: """Return the required properties. Returns: The required properties, otherwise an empty set. """ try: required = self._root_node._active_strategies[0]._required except (AttributeError, IndexError): return set() if required is None: return set() return required
[docs] def check_property_names(self, *names: str) -> None: """Check that the names are existing properties. Args: *names: The names to be checked. Raises: KeyError: If a name is not an existing property. """ for name in names: if name not in raise KeyError(f"The name {name} is not in the grammar.")
[docs] def add_schema(self, schema, update: bool) -> None: """ Args: update: Whether to update or merge the schema. """ # noqa: D205 D212 D415 _MergeRequiredStrategy.update = update super().add_schema(schema)
[docs] def add_object(self, obj, update: bool) -> None: """ Args: update: Whether to update or merge the schema. """ # noqa: D205 D212 D415 _MergeRequiredStrategy.update = update super().add_object(obj)