# This file is part of Cockpit.
#
# Copyright (C) 2023 Red Hat, Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from enum import Enum
from typing import Callable, Container, Dict, List, Mapping, Optional, Sequence, Type, TypeVar, Union
JsonLiteral = Union[str, float, bool, None]
# immutable
JsonValue = Union['JsonObject', Sequence['JsonValue'], JsonLiteral]
JsonObject = Mapping[str, JsonValue]
# mutable
JsonDocument = Union['JsonDict', 'JsonList', JsonLiteral]
JsonDict = Dict[str, JsonDocument]
JsonList = List[JsonDocument]
DT = TypeVar('DT')
T = TypeVar('T')
class JsonError(Exception):
value: object
def __init__(self, value: object, msg: str):
super().__init__(msg)
self.value = value
def typechecked(value: JsonValue, expected_type: Type[T]) -> T:
"""Ensure a JSON value has the expected type, returning it if so."""
if not isinstance(value, expected_type):
raise JsonError(value, f'must have type {expected_type.__name__}')
return value
# We can't use None as a sentinel because it's often the actual default value
# EllipsisType is difficult because it's not available before 3.10.
# See https://peps.python.org/pep-0484/#support-for-singleton-types-in-unions
class _Empty(Enum):
TOKEN = 0
_empty = _Empty.TOKEN
def _get(obj: JsonObject, cast: Callable[[JsonValue], T], key: str, default: Union[DT, _Empty]) -> Union[T, DT]:
try:
return cast(obj[key])
except KeyError:
if default is not _empty:
return default
raise JsonError(obj, f"attribute '{key}' required") from None
except JsonError as exc:
target = f"attribute '{key}'" + (' elements:' if exc.value is not obj[key] else ':')
raise JsonError(obj, f"{target} {exc!s}") from exc
def get_bool(obj: JsonObject, key: str, default: Union[DT, _Empty] = _empty) -> Union[DT, bool]:
return _get(obj, lambda v: typechecked(v, bool), key, default)
def get_int(obj: JsonObject, key: str, default: Union[DT, _Empty] = _empty) -> Union[DT, int]:
return _get(obj, lambda v: typechecked(v, int), key, default)
def get_str(obj: JsonObject, key: str, default: Union[DT, _Empty] = _empty) -> Union[DT, str]:
return _get(obj, lambda v: typechecked(v, str), key, default)
def get_str_or_none(obj: JsonObject, key: str, default: Optional[str]) -> Optional[str]:
return _get(obj, lambda v: None if v is None else typechecked(v, str), key, default)
def get_str_or_int(obj: JsonObject, key: str, default: Optional[Union[str, int]]) -> Optional[Union[str, int]]:
def as_str_or_int(value: JsonValue) -> Union[str, int]:
if not isinstance(value, (str, int)):
raise JsonError(value, 'must be a string or integer')
return value
return _get(obj, as_str_or_int, key, default)
def get_dict(obj: JsonObject, key: str, default: Union[DT, _Empty] = _empty) -> Union[DT, JsonObject]:
return _get(obj, lambda v: typechecked(v, dict), key, default)
def get_object(
obj: JsonObject,
key: str,
constructor: Callable[[JsonObject], T],
default: Union[DT, _Empty] = _empty
) -> Union[DT, T]:
return _get(obj, lambda v: constructor(typechecked(v, dict)), key, default)
def get_strv(obj: JsonObject, key: str, default: Union[DT, _Empty] = _empty) -> Union[DT, Sequence[str]]:
def as_strv(value: JsonValue) -> Sequence[str]:
return tuple(typechecked(item, str) for item in typechecked(value, list))
return _get(obj, as_strv, key, default)
def get_enum(
obj: JsonObject, key: str, choices: Container[str], default: Union[DT, _Empty] = _empty
) -> Union[DT, str]:
def as_choice(value: JsonValue) -> str:
# mypy can't do `__eq__()`-based type narrowing...
# https://github.com/python/mypy/issues/17101
if isinstance(value, str) and value in choices:
return value
raise JsonError(value, f'invalid value "{value}" not in {choices}')
return _get(obj, as_choice, key, default)
def get_objv(obj: JsonObject, key: str, constructor: Callable[[JsonObject], T]) -> Union[DT, Sequence[T]]:
def as_objv(value: JsonValue) -> Sequence[T]:
return tuple(constructor(typechecked(item, dict)) for item in typechecked(value, list))
return _get(obj, as_objv, key, ())
def create_object(message: 'JsonObject | None', kwargs: JsonObject) -> JsonObject:
"""Constructs a JSON object based on message and kwargs.
If only message is given, it is returned, unmodified. If message is None,
it is equivalent to an empty dictionary. A copy is always made.
If kwargs are present, then any underscore ('_') present in a key name is
rewritten to a dash ('-'). This is intended to bridge between the required
Python syntax when providing kwargs and idiomatic JSON (which uses '-' for
attributes). These values override values in message.
The idea is that `message` should be used for passing data along, and
kwargs used for data originating at a given call site, possibly including
modifications to an original message.
"""
result = dict(message or {})
for key, value in kwargs.items():
# rewrite '_' (necessary in Python syntax kwargs list) to '-' (idiomatic JSON)
json_key = key.replace('_', '-')
result[json_key] = value
return result
def json_merge_patch(current: JsonObject, patch: JsonObject) -> JsonObject:
"""Perform a JSON merge patch (RFC 7396) using 'current' and 'patch'.
Neither of the original dictionaries is modified — the result is returned.
"""
# Always take a copy ('result') — we never modify the input ('current')
result = dict(current)
for key, patch_value in patch.items():
if isinstance(patch_value, Mapping):
current_value = current.get(key, None)
if not isinstance(current_value, Mapping):
current_value = {}
result[key] = json_merge_patch(current_value, patch_value)
elif patch_value is not None:
result[key] = patch_value
else:
result.pop(key, None)
return result
def json_merge_and_filter_patch(current: JsonDict, patch: JsonDict) -> None:
"""Perform a JSON merge patch (RFC 7396) modifying 'current' with 'patch'.
Also modifies 'patch' to remove redundant operations.
"""
for key, patch_value in tuple(patch.items()):
current_value = current.get(key, None)
if isinstance(patch_value, dict):
if not isinstance(current_value, dict):
current[key] = current_value = {}
json_merge_and_filter_patch(current_value, patch_value)
else:
json_merge_and_filter_patch(current_value, patch_value)
if not patch_value:
del patch[key]
elif current_value == patch_value:
del patch[key]
elif patch_value is not None:
current[key] = patch_value
else:
del current[key]