# Copyright 2018 ACSONE SA/NV # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html). import inspect import logging from collections import OrderedDict _logger = logging.getLogger(__name__) # Decorator attribute added on a route function (cfr Odoo's route) ROUTING_DECORATOR_ATTR = "original_routing" SUPPORTED_META = ["title", "description", "example", "examples"] def cerberus_to_json(schema): """Convert a Cerberus schema to a JSON schema""" result = OrderedDict() required = [] properties = OrderedDict() result["type"] = "object" result["required"] = required result["properties"] = properties for field, spec in list(schema.items()): props = _get_field_props(spec) properties[field] = props if spec.get("required"): required.append(field) # sort required to get the same order on each run and easy comparison into # the tests required.sort() return result def _get_field_props(spec): # noqa: C901 resp = OrderedDict() # dictionary of tuple (json type, json fromat) by cerberus type for # cerberus types requiring a specific mapping to a json type/format type_map = { "dict": ("object",), "list": ("array",), "objectid": ("string", "objectid"), "datetime": ("string", "date-time"), "float": ("number", "float"), } _type = spec.get("type") if _type is None: return resp if "description" in spec: resp["description"] = spec["description"] if "meta" in spec: for key in SUPPORTED_META: value = spec["meta"].get(key) if value: resp[key] = value if "allowed" in spec: resp["enum"] = spec["allowed"] if "default" in spec: resp["default"] = spec["default"] if "minlength" in spec: if _type == "string": resp["minLength"] = spec["minlength"] elif _type == "list": resp["minItems"] = spec["minlength"] if "maxlength" in spec: if _type == "string": resp["maxLength"] = spec["maxlength"] elif _type == "list": resp["maxItems"] = spec["maxlength"] if "min" in spec: if _type in ["number", "integer", "float"]: resp["minimum"] = spec["min"] if "max" in spec: if _type in ["number", "integer", "float"]: resp["maximum"] = spec["max"] if "readonly" in spec: resp["readOnly"] = spec["readonly"] if "regex" in spec: resp["pattern"] = spec["regex"] if "nullable" in spec: resp["nullable"] = spec["nullable"] if "allowed" in spec: resp["enum"] = spec["allowed"] json_type = type_map.get(_type, (_type,)) resp["type"] = json_type[0] if json_type[0] == "object": if "schema" in spec: resp.update(cerberus_to_json(spec["schema"])) additional_properties = {} if "keysrules" in spec: rule_value_type = spec["keysrules"].get("type", "string") if rule_value_type != "string": _logger.debug( "Openapi only support key/value mapping definition where" " the keys are strings. Received %s", rule_value_type, ) if "valuesrules" in spec: values_rules = spec["valuesrules"] rule_value_type = values_rules.get("type", "string") additional_properties["type"] = rule_value_type if "schema" in values_rules: additional_properties.update(cerberus_to_json(values_rules["schema"])) if additional_properties: resp["additionalProperties"] = additional_properties elif json_type[0] == "array": if "schema" in spec: resp["items"] = _get_field_props(spec["schema"]) else: resp["items"] = {"type": "string"} else: try: resp["format"] = json_type[1] # pylint:disable=except-pass except IndexError: pass return resp def _inspect_methods(cls): """Return all methods of a given class as (name, value) pairs sorted by name. inspect.getmembers was initially used. Unfortunately, instance's properties was accessed into the loop and could raise some exception since we are into the startup process and all the resources are not yet initialized. """ results = [] for attribute in inspect.classify_class_attrs(cls): if attribute.kind != "method": continue name = attribute.name method = getattr(cls, name) results.append((name, method)) results.sort(key=lambda pair: pair[0]) return results