var/opt/nydus/ops/primordial/jsonutils.py000064400000006367147205316150014667 0ustar00# -*- coding: utf-8 -*- import datetime import json from json import JSONEncoder, JSONDecoder import logging import re from importlib import import_module from typing import Any, Callable, Dict from jsonschema import validate as json_validate from jsonschema import exceptions as json_exceptions from primordial.constants import CANONICAL_TIMESTRING_FORMAT DELAY_INCREASE_SECS = 1.0 MAX_RETRIES = 3 LOGGER = logging.getLogger(__name__) def validate_swagger(json_dict: Dict[Any, Any], schema_dict: Dict[Any, Any]) -> bool: """Test that json_dict dictionary conforms to the swagger schema provided. :param json_dict: A dictionary of values :param schema_dict: A dict representation of swagger schema :returns: True or False :raises AssertionError: if either is None """ assert json_dict is not None assert schema_dict is not None try: json_validate(json_dict, schema_dict) return True except json_exceptions.ValidationError: return False def pretty_json_str(json_dict: Dict[Any, Any]) -> str: """Return a pretty-formatted string containing the values in json_dict. :param json_dict: A dictionary or other object that is serializable as a json string :returns: a (unicode) string representation with a default format of sorted keys and indent of 4 spaces :raises TypeError: if the thing in question cannot be json serialized """ return json.dumps(json_dict, cls=DateTimeEncoder, sort_keys=True, separators=(',', ':')) def func_to_str(func: Callable) -> str: """Create a string for a python function in a module. Class and object methods not supported currently. :param func: a python function :returns: a string representing this function """ return "%s.%s" % (func.__module__, func.__name__) def str_to_func(func_str: str) -> Callable: """Use importlib to turn a string into a function. At the moment, this is limited to module functions and not classes. :param func_str: the string representation of a function, e.g. ``tests.test_workflow.ex_tst_func`` :returns: a python function :raises ImportError: if the function does not exist """ funcnotation = func_str.split('.') symbol = funcnotation.pop() module = import_module('.'.join(funcnotation)) return getattr(module, symbol) class DateTimeEncoder(JSONEncoder): """Uses our canonical format.""" def default(self, obj: Any) -> Any: # pylint: disable=E0202,W0221,arguments-renamed if isinstance(obj, datetime.datetime): ret = obj.strftime(CANONICAL_TIMESTRING_FORMAT) else: ret = JSONEncoder.default(self, obj) return ret # regex for CANONICAL_TIMESTRING_FORMAT DATETIME_REGEX = re.compile(r"^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}).(\d{6})Z$") class DateTimeDecoder(JSONDecoder): """Uses our canonical format.""" def __init__(self, *args, **kwargs): super().__init__(*args, object_hook=self.dict_to_object, **kwargs) def dict_to_object(self, d: Dict[Any, Any]) -> Any: # pylint: disable=C0103 for key, value in d.items(): if isinstance(value, str) and DATETIME_REGEX.match(value): d[key] = datetime.datetime.strptime(value, CANONICAL_TIMESTRING_FORMAT) return d