Add support for specifying nested structures as JSON

linodecli/api_request.py

@@ -257,11 +257,15 @@ def _build_request_body(ctx, operation, parsed_args) -> Optional[str]:
 
     # expand paths
     for k, v in vars(parsed_args).items():
+        if v is None:
+            continue
+
         cur = expanded_json
         for part in k.split(".")[:-1]:
             if part not in cur:
                 cur[part] = {}
             cur = cur[part]
+
         cur[k.split(".")[-1]] = v
 
     return json.dumps(_traverse_request_body(expanded_json))
@@ -415,4 +419,4 @@ def _check_retry(response):
 
 def _get_retry_after(headers):
     retry_str = headers.get("Retry-After", "")
-    return int(retry_str) if retry_str else 0
+    return int(retry_str) if retry_str else 0

linodecli/arg_helpers.py

@@ -341,7 +341,7 @@ def help_with_ops(ops, config):
     )
 
 
-def action_help(cli, command, action):
+def action_help(cli, command, action):  # pylint: disable=too-many-branches
     """
     Prints help relevant to the command and action
     """
@@ -407,6 +407,9 @@ def action_help(cli, command, action):
             if arg.nullable:
                 extensions.append("nullable")
 
+            if arg.is_parent:
+                extensions.append("conflicts with children")
+
             suffix = (
                 f" ({', '.join(extensions)})" if len(extensions) > 0 else ""
             )

linodecli/baked/operation.py

@@ -10,7 +10,7 @@
 import sys
 from getpass import getpass
 from os import environ, path
-from typing import List, Tuple
+from typing import Any, List, Tuple
 
 from openapi3.paths import Operation
 
@@ -427,14 +427,14 @@ def _add_args_post_put(self, parser) -> List[Tuple[str, str]]:
                     action=ArrayAction,
                     type=arg_type_handler,
                 )
-            elif arg.list_item:
+            elif arg.is_child:
                 parser.add_argument(
                     "--" + arg.path,
                     metavar=arg.name,
                     action=ListArgumentAction,
                     type=arg_type_handler,
                 )
-                list_items.append((arg.path, arg.list_parent))
+                list_items.append((arg.path, arg.parent))
             else:
                 if arg.datatype == "string" and arg.format == "password":
                     # special case - password input
@@ -463,10 +463,55 @@ def _add_args_post_put(self, parser) -> List[Tuple[str, str]]:
 
         return list_items
 
+    def _validate_parent_child_conflicts(self, parsed: argparse.Namespace):
+        """
+        This method validates that no child arguments (e.g. --interfaces.purpose) are
+        specified alongside their parent (e.g. --interfaces).
+        """
+        conflicts = {}
+
+        for arg in self.args:
+            parent = arg.parent
+            arg_value = getattr(parsed, arg.path, None)
+
+            if parent is None or arg_value is None:
+                print(arg.path, parent, arg_value)
+                continue
+
+            # Special case to ignore child arguments that are not specified
+            # but are implicitly populated by ListArgumentAction.
+            if isinstance(arg_value, list) and arg_value.count(None) == len(
+                arg_value
+            ):
+                continue
+
+            # If the parent isn't defined, we can
+            # skip this one
+            if getattr(parsed, parent) is None:
+                continue
+
+            if parent not in conflicts:
+                conflicts[parent] = []
+
+            # We found a conflict
+            conflicts[parent].append(arg)
+
+        # No conflicts found
+        if len(conflicts) < 1:
+            return
+
+        for parent, args in conflicts.items():
+            arg_format = ", ".join([f"--{v.path}" for v in args])
+            print(
+                f"Argument(s) {arg_format} cannot be specified when --{parent} is specified.",
+                file=sys.stderr,
+            )
+
+        sys.exit(2)
+
     @staticmethod
     def _handle_list_items(
-        list_items,
-        parsed,
+        list_items: List[Tuple[str, str]], parsed: Any
     ):  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
         lists = {}
 
@@ -563,4 +608,9 @@ def parse_args(self, args):
         elif self.method in ("post", "put"):
             list_items = self._add_args_post_put(parser)
 
-        return self._handle_list_items(list_items, parser.parse_args(args))
+        parsed = parser.parse_args(args)
+
+        if self.method in ("post", "put"):
+            self._validate_parent_child_conflicts(parsed)
+
+        return self._handle_list_items(list_items, parsed)

linodecli/baked/request.py

@@ -9,7 +9,13 @@ class OpenAPIRequestArg:
     """
 
     def __init__(
-        self, name, schema, required, prefix=None, list_parent=None
+        self,
+        name,
+        schema,
+        required,
+        prefix=None,
+        is_parent=False,
+        parent=None,
     ):  # pylint: disable=too-many-arguments
         """
         Parses a single Schema node into a argument the CLI can use when making
@@ -23,6 +29,10 @@ def __init__(
         :param prefix: The prefix for this arg's path, used in the actual argument
                        to the CLI to ensure unique arg names
         :type prefix: str
+        :param is_parent: Whether this argument is a parent to child fields.
+        :type is_parent: bool
+        :param parent: If applicable, the path to the parent list for this argument.
+        :type parent: Optional[str]
         """
         #: The name of this argument, mostly used for display and docs
         self.name = name
@@ -52,7 +62,7 @@ def __init__(
 
         # If this is a deeply nested array we should treat it as JSON.
         # This allows users to specify fields like --interfaces.ip_ranges.
-        if schema.type == "array" and list_parent is not None:
+        if is_parent or (schema.type == "array" and parent is not None):
             self.format = "json"
 
         #: The type accepted for this argument. This will ultimately determine what
@@ -64,13 +74,16 @@ def __init__(
         #: The type of item accepted in this list; if None, this is not a list
         self.item_type = None
 
-        #: Whether the argument is a field in a nested list.
-        self.list_item = list_parent is not None
+        #: Whether the argument is a parent to child fields.
+        self.is_parent = is_parent
+
+        #: Whether the argument is a nested field.
+        self.is_child = parent is not None
 
         #: The name of the list this argument falls under.
         #: This allows nested dictionaries to be specified in lists of objects.
         #: e.g. --interfaces.ipv4.nat_1_1
-        self.list_parent = list_parent
+        self.parent = parent
 
         #: The path of the path element in the schema.
         self.prefix = prefix
@@ -82,15 +95,8 @@ def __init__(
         if self.datatype == "array" and schema.items:
             self.item_type = schema.items.type
 
-        # make sure we're not doing something wrong
-        if self.item_type == "object":
-            raise ValueError(
-                "Invalid OpenAPIRequestArg creation; created arg for base object "
-                "instead of object's properties!  This is a programming error."
-            )
-
 
-def _parse_request_model(schema, prefix=None, list_parent=None):
+def _parse_request_model(schema, prefix=None, parent=None):
     """
     Parses a schema into a list of OpenAPIRequest objects
     :param schema: The schema to parse as a request model
@@ -112,8 +118,23 @@ def _parse_request_model(schema, prefix=None, list_parent=None):
             if v.type == "object" and not v.readOnly and v.properties:
                 # nested objects receive a prefix and are otherwise parsed normally
                 pref = prefix + "." + k if prefix else k
+
+                # Support specifying this object as JSON
+                args.append(
+                    OpenAPIRequestArg(
+                        k,
+                        v,
+                        False,
+                        prefix=prefix,
+                        is_parent=True,
+                        parent=parent,
+                    )
+                )
+
                 args += _parse_request_model(
-                    v, prefix=pref, list_parent=list_parent
+                    v,
+                    prefix=pref,
+                    parent=parent if parent is not None else pref,
                 )
             elif (
                 v.type == "array"
@@ -124,9 +145,20 @@ def _parse_request_model(schema, prefix=None, list_parent=None):
                 # handle lists of objects as a special case, where each property
                 # of the object in the list is its own argument
                 pref = prefix + "." + k if prefix else k
-                args += _parse_request_model(
-                    v.items, prefix=pref, list_parent=pref
+
+                # Support specifying this list as JSON
+                args.append(
+                    OpenAPIRequestArg(
+                        k,
+                        v.items,
+                        False,
+                        prefix=prefix,
+                        is_parent=True,
+                        parent=parent,
+                    )
                 )
+
+                args += _parse_request_model(v.items, prefix=pref, parent=pref)
             else:
                 # required fields are defined in the schema above the property, so
                 # we have to check here if required fields are defined/if this key
@@ -136,7 +168,7 @@ def _parse_request_model(schema, prefix=None, list_parent=None):
                     required = k in schema.required
                 args.append(
                     OpenAPIRequestArg(
-                        k, v, required, prefix=prefix, list_parent=list_parent
+                        k, v, required, prefix=prefix, parent=parent
                     )
                 )
 
@@ -193,4 +225,4 @@ def __init__(self, response_model):
             )
 
         # actually parse out what we can filter by
-        self.attrs = [c for c in response_model.attrs if c.filterable]
+        self.attrs = [c for c in response_model.attrs if c.filterable]

tests/integration/linodes/test_interfaces.py

@@ -1,5 +1,6 @@
 import json
 import time
+from typing import Any, Dict
 
 import pytest
 
@@ -65,9 +66,57 @@ def linode_with_vpc_interface():
     delete_target_id(target="vpcs", id=vpc_id)
 
 
-def test_with_vpc_interface(linode_with_vpc_interface):
-    linode_json, vpc_json = linode_with_vpc_interface
+@pytest.fixture
+def linode_with_vpc_interface_as_json():
+    vpc_json = create_vpc_w_subnet()
+
+    vpc_region = vpc_json["region"]
+    vpc_id = str(vpc_json["id"])
+    subnet_id = int(vpc_json["subnets"][0]["id"])
+
+    linode_json = json.loads(
+        exec_test_command(
+            BASE_CMD
+            + [
+                "create",
+                "--type",
+                "g6-nanode-1",
+                "--region",
+                vpc_region,
+                "--image",
+                DEFAULT_TEST_IMAGE,
+                "--root_pass",
+                DEFAULT_RANDOM_PASS,
+                "--interfaces",
+                json.dumps(
+                    [
+                        {
+                            "purpose": "vpc",
+                            "primary": True,
+                            "subnet_id": subnet_id,
+                            "ipv4": {"nat_1_1": "any", "vpc": "10.0.0.5"},
+                            "ip_ranges": ["10.0.0.6/32"],
+                        },
+                        {"purpose": "public"},
+                    ]
+                ),
+                "--json",
+                "--suppress-warnings",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )[0]
+
+    yield linode_json, vpc_json
 
+    delete_target_id(target="linodes", id=str(linode_json["id"]))
+    delete_target_id(target="vpcs", id=vpc_id)
+
+
+def assert_interface_configuration(
+    linode_json: Dict[str, Any], vpc_json: Dict[str, Any]
+):
     config_json = json.loads(
         exec_test_command(
             BASE_CMD
@@ -95,3 +144,11 @@ def test_with_vpc_interface(linode_with_vpc_interface):
 
     assert not public_interface["primary"]
     assert public_interface["purpose"] == "public"
+
+
+def test_with_vpc_interface(linode_with_vpc_interface):
+    assert_interface_configuration(*linode_with_vpc_interface)
+
+
+def test_with_vpc_interface_as_json(linode_with_vpc_interface_as_json):
+    assert_interface_configuration(*linode_with_vpc_interface_as_json)

tests/unit/test_arg_helpers.py

@@ -220,7 +220,7 @@ def test_action_help_post_method(self, capsys, mocker, mock_cli):
         assert "test description" in captured.out
         assert "test description 2" in captured.out
         assert "(required)" in captured.out
-        assert "(JSON, nullable)" in captured.out
+        assert "(JSON, nullable, conflicts with children)" in captured.out
         assert "filter results" not in captured.out
 
     def test_action_help_get_method(self, capsys, mocker, mock_cli):