Allow dict passthrough for config_create 'devices' field

Author: lgarber-akamai

linode_api4/objects/linode.py

@@ -1243,15 +1243,20 @@ def _func(value):
 
         return password
 
+    # Type aliases for config_create parameters
+    ConfigCreateDevice = Union["Disk", "Volume", Dict[str, Any]]
+    ConfigCreateDisk = Union["Disk", int]
+    ConfigCreateVolume = Union["Volume", int]
+
     # create derived objects
     def config_create(
         self,
         kernel=None,
         label=None,
-        devices=[],
-        disks=[],
-        volumes=[],
-        interfaces=[],
+        devices=None,
+        disks=None,
+        volumes=None,
+        interfaces=None,
         **kwargs,
     ):
         """
@@ -1265,15 +1270,20 @@ def config_create(
         :param volumes: The volumes, starting after the last disk, to map to this
             config
         :param devices: A list of devices to assign to this config, in device
-            index order.  Values must be of type Disk or Volume. If this is
-            given, you may not include disks or volumes.
+            index order, or a raw device mapping dict to pass directly to the
+            API. List values must be of type Disk or Volume. If this is given,
+            you may not include disks or volumes.
         :param **kwargs: Any other arguments accepted by the api.
 
         :returns: A new Linode Config
         """
         # needed here to avoid circular imports
         from .volume import Volume  # pylint: disable=import-outside-toplevel
 
+        disks = [] if disks is None else disks
+        volumes = [] if volumes is None else volumes
+        interfaces = [] if interfaces is None else interfaces
+
         hypervisor_prefix = "sd" if self.hypervisor == "kvm" else "xvd"
 
         device_limit = int(
@@ -1288,52 +1298,58 @@ def config_create(
             for suffix in generate_device_suffixes(device_limit)
         ]
 
-        device_map = {
-            device_names[i]: None for i in range(0, len(device_names))
-        }
+        def _flatten_device(device: Any):
+            if device is None:
+                return None
+            elif isinstance(device, Disk):
+                return {"disk_id": device.id}
+            elif isinstance(device, Volume):
+                return {"volume_id": device.id}
+            else:
+                raise TypeError("Disk or Volume expected!")
+
+        def _build_devices():
+            # Devices is a dict, pass through
+            if isinstance(devices, dict):
+                return devices
+
+            if devices is not None:
+                device_list = (
+                    devices if isinstance(devices, list) else [devices]
+                )
+
+                return {
+                    device_names[i]: _flatten_device(device_list[i])
+                    for i in range(len(devices))
+                }
+
+            # Devices were given through named args
+            return [
+                (
+                    {"volume_id": volume.id}
+                    if isinstance(volume, int)
+                    else _flatten_device(volume)
+                )
+                for volume in volumes
+            ] + [
+                (
+                    {"disk_id": disk.id}
+                    if isinstance(disk, int)
+                    else _flatten_device(disk)
+                )
+                for disk in disks
+            ]
 
         if devices and (disks or volumes):
             raise ValueError(
                 'You may not call config_create with "devices" and '
                 'either of "disks" or "volumes" specified!'
             )
 
-        if not devices:
-            if not isinstance(disks, list):
-                disks = [disks]
-            if not isinstance(volumes, list):
-                volumes = [volumes]
-
-            devices = []
-
-            for d in disks:
-                if d is None:
-                    devices.append(None)
-                elif isinstance(d, Disk):
-                    devices.append(d)
-                else:
-                    devices.append(Disk(self._client, int(d), self.id))
-
-            for v in volumes:
-                if v is None:
-                    devices.append(None)
-                elif isinstance(v, Volume):
-                    devices.append(v)
-                else:
-                    devices.append(Volume(self._client, int(v)))
-
-        if not devices:
-            raise ValueError("Must include at least one disk or volume!")
+        device_map = _build_devices()
 
-        for i, d in enumerate(devices):
-            if d is None:
-                pass
-            elif isinstance(d, Disk):
-                device_map[device_names[i]] = {"disk_id": d.id}
-            elif isinstance(d, Volume):
-                device_map[device_names[i]] = {"volume_id": d.id}
-            else:
-                raise TypeError("Disk or Volume expected!")
+        if len(device_map) < 1:
+            raise ValueError("Must include at least one disk or volume!")
 
         param_interfaces = []
         for interface in interfaces:
@@ -1845,8 +1861,8 @@ def clone(
         to_linode=None,
         region=None,
         instance_type=None,
-        configs=[],
-        disks=[],
+        configs=None,
+        disks=None,
         label=None,
         group=None,
         with_backups=None,
@@ -1902,6 +1918,9 @@ def clone(
                 'You may only specify one of "to_linode" and "region"'
             )
 
+        configs = [] if configs is None else configs
+        disks = [] if disks is None else disks
+
         if region and not type:
             raise ValueError('Specifying a region requires a "service" as well')
 

test/unit/objects/linode_test.py

@@ -459,6 +459,34 @@ def test_create_disk(self):
         assert disk.id == 12345
         assert disk.disk_encryption == InstanceDiskEncryptionType.disabled
 
+    def test_create_config_with_device_map(self):
+        """
+        Tests that config_create passes through a raw device map unchanged.
+        """
+        linode = Instance(self.client, 123)
+        devices = {
+            "sda": {"disk_id": 111},
+            "sdb": {"volume_id": 222},
+            "sdc": None,
+        }
+
+        with self.mock_post(
+            {"id": 456, "devices": devices, "interfaces": []}
+        ) as m:
+            config = linode.config_create(label="test-config", devices=devices)
+
+            self.assertEqual(m.call_url, "/linode/instances/123/configs")
+            self.assertEqual(
+                m.call_data,
+                {
+                    "label": "test-config",
+                    "devices": devices,
+                    "interfaces": [],
+                },
+            )
+
+        self.assertEqual(config.id, 456)
+
     def test_get_placement_group(self):
         """
         Tests that you can get the placement group for a Linode