From 959283d57ba879f222875601da141bf6f91f6771 Mon Sep 17 00:00:00 2001 From: Alexander Rickardsson Date: Fri, 18 Aug 2023 11:46:10 +0200 Subject: [PATCH] client: handle references in user config schema Currently the client is not able to deal with references in the user config schema, but will throw key errors. This will improve the situation by traversing the original object until the actual definition is found --- aiven/client/cli.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/aiven/client/cli.py b/aiven/client/cli.py index bfa49f36..b1c7c431 100644 --- a/aiven/client/cli.py +++ b/aiven/client/cli.py @@ -177,11 +177,28 @@ def add_args(self, parser: ArgumentParser) -> None: help="Wait for up to N seconds for a response to a request (default: infinite)", ) + def _find_reference(self, obj_def: Mapping[str, Any], spec: Mapping[str, Any]) -> Optional[Mapping[str, Any]]: + ref_prefix = "#/definitions/" + + if "definitions" not in obj_def: + return None + ref = spec["$ref"][len(ref_prefix) :] + return obj_def["definitions"][ref] + def collect_user_config_options(self, obj_def: Mapping[str, Any], prefixes: list[str] | None = None) -> dict[str, Any]: opts = {} for prop, spec in sorted(obj_def.get("properties", {}).items()): full_prop = prefixes + [prop] if prefixes else [prop] full_name = ".".join(full_prop) + if "type" not in spec: + # It is possible this is a reference, and if so we need to find the actual entry + if "$ref" in spec: + spec = self._find_reference(obj_def, spec) + elif "allOf" in spec: + for entry in spec["allOf"]: + spec = self._find_reference(obj_def, entry) + if not spec: + continue types = spec["type"] if not isinstance(types, list): types = [types]