Implement support for VPC Dual Stack

Author: lgarber-akamai

Diff for: linode_api4/groups/vpc.py

@@ -2,8 +2,10 @@
 
 from linode_api4.errors import UnexpectedResponseError
 from linode_api4.groups import Group
-from linode_api4.objects import VPC, Region, VPCIPAddress
+from linode_api4.objects import VPC, Region, VPCIPAddress, VPCIPv6RangeOptions
+from linode_api4.objects.base import _flatten_request_body_recursive
 from linode_api4.paginated_list import PaginatedList
+from linode_api4.util import drop_null_keys
 
 
 class VPCGroup(Group):
@@ -33,6 +35,7 @@ def create(
         region: Union[Region, str],
         description: Optional[str] = None,
         subnets: Optional[List[Dict[str, Any]]] = None,
+        ipv6: Optional[List[Union[VPCIPv6RangeOptions, Dict[str, Any]]]] = None,
         **kwargs,
     ) -> VPC:
         """
@@ -48,30 +51,33 @@ def create(
         :type description: Optional[str]
         :param subnets: A list of subnets to create under this VPC.
         :type subnets: List[Dict[str, Any]]
+        :param ipv6: The IPv6 address ranges for this VPC.
+        :type ipv6: List[Union[VPCIPv6RangeOptions, Dict[str, Any]]]
 
         :returns: The new VPC object.
         :rtype: VPC
         """
         params = {
             "label": label,
             "region": region.id if isinstance(region, Region) else region,
+            "description": description,
+            "ipv6": ipv6,
+            "subnets": subnets,
         }
 
-        if description is not None:
-            params["description"] = description
-
         if subnets is not None and len(subnets) > 0:
             for subnet in subnets:
                 if not isinstance(subnet, dict):
                     raise ValueError(
                         f"Unsupported type for subnet: {type(subnet)}"
                     )
 
-            params["subnets"] = subnets
-
         params.update(kwargs)
 
-        result = self.client.post("/vpcs", data=params)
+        result = self.client.post(
+            "/vpcs",
+            data=drop_null_keys(_flatten_request_body_recursive(params)),
+        )
 
         if not "id" in result:
             raise UnexpectedResponseError(

Diff for: linode_api4/objects/account.py

@@ -601,7 +601,7 @@ def entity(self):
             )
         return self.cls(self._client, self.id)
 
-    def _serialize(self):
+    def _serialize(self, *args, **kwargs):
         """
         Returns this grant in as JSON the api will accept.  This is only relevant
         in the context of UserGrants.save
@@ -668,7 +668,7 @@ def _grants_dict(self):
 
         return grants
 
-    def _serialize(self):
+    def _serialize(self, *args, **kwargs):
         """
         Returns the user grants in as JSON the api will accept.
         This is only relevant in the context of UserGrants.save

Diff for: linode_api4/objects/base.py

@@ -114,6 +114,9 @@ def _flatten_base_subclass(obj: "Base") -> Optional[Dict[str, Any]]:
 
     @property
     def dict(self):
+        return self._serialize()
+
+    def _serialize(self, is_put: bool = False) -> Dict[str, Any]:
         result = vars(self).copy()
         cls = type(self)
 
@@ -123,7 +126,7 @@ def dict(self):
             elif isinstance(v, list):
                 result[k] = [
                     (
-                        item.dict
+                        item._serialize(is_put=is_put)
                         if isinstance(item, (cls, JSONObject))
                         else (
                             self._flatten_base_subclass(item)
@@ -136,7 +139,7 @@ def dict(self):
             elif isinstance(v, Base):
                 result[k] = self._flatten_base_subclass(v)
             elif isinstance(v, JSONObject):
-                result[k] = v.dict
+                result[k] = v._serialize(is_put=is_put)
 
         return result
 
@@ -278,9 +281,9 @@ def save(self, force=True) -> bool:
                     data[key] = None
 
             # Ensure we serialize any values that may not be already serialized
-            data = _flatten_request_body_recursive(data)
+            data = _flatten_request_body_recursive(data, is_put=True)
         else:
-            data = self._serialize()
+            data = self._serialize(is_put=True)
 
         resp = self._client.put(type(self).api_endpoint, model=self, data=data)
 
@@ -316,7 +319,7 @@ def invalidate(self):
 
         self._set("_populated", False)
 
-    def _serialize(self):
+    def _serialize(self, is_put: bool = False):
         """
         A helper method to build a dict of all mutable Properties of
         this object
@@ -345,7 +348,7 @@ def _serialize(self):
 
         # Resolve the underlying IDs of results
         for k, v in result.items():
-            result[k] = _flatten_request_body_recursive(v)
+            result[k] = _flatten_request_body_recursive(v, is_put=is_put)
 
         return result
 
@@ -503,7 +506,7 @@ def make_instance(cls, id, client, parent_id=None, json=None):
         return Base.make(id, client, cls, parent_id=parent_id, json=json)
 
 
-def _flatten_request_body_recursive(data: Any) -> Any:
+def _flatten_request_body_recursive(data: Any, is_put: bool = False) -> Any:
     """
     This is a helper recursively flatten the given data for use in an API request body.
 
@@ -515,15 +518,18 @@ def _flatten_request_body_recursive(data: Any) -> Any:
     """
 
     if isinstance(data, dict):
-        return {k: _flatten_request_body_recursive(v) for k, v in data.items()}
+        return {
+            k: _flatten_request_body_recursive(v, is_put=is_put)
+            for k, v in data.items()
+        }
 
     if isinstance(data, list):
-        return [_flatten_request_body_recursive(v) for v in data]
+        return [_flatten_request_body_recursive(v, is_put=is_put) for v in data]
 
     if isinstance(data, Base):
         return data.id
 
     if isinstance(data, MappedObject) or issubclass(type(data), JSONObject):
-        return data.dict
+        return data._serialize(is_put=is_put)
 
     return data

Diff for: linode_api4/objects/linode.py

@@ -1,6 +1,6 @@
 import string
 import sys
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 from os import urandom
@@ -291,10 +291,83 @@ def _populate(self, json):
 
 @dataclass
 class ConfigInterfaceIPv4(JSONObject):
+    """
+    ConfigInterfaceIPv4 represents the IPv4 configuration of a VPC interface.
+    """
+
     vpc: str = ""
     nat_1_1: str = ""
 
 
+@dataclass
+class ConfigInterfaceIPv6SLAACOptions(JSONObject):
+    """
+    ConfigInterfaceIPv6SLAACOptions is used to set a single IPv6 SLAAC configuration of a VPC interface.
+    """
+
+    range: str = ""
+
+
+@dataclass
+class ConfigInterfaceIPv6RangeOptions(JSONObject):
+    """
+    ConfigInterfaceIPv6RangeOptions is used to set a single IPv6 range configuration of a VPC interface.
+    """
+
+    range: str = ""
+
+
+@dataclass
+class ConfigInterfaceIPv6Options(JSONObject):
+    """
+    ConfigInterfaceIPv6Options is used to set the IPv6 configuration of a VPC interface.
+    """
+
+    slaac: List[ConfigInterfaceIPv6SLAACOptions] = field(
+        default_factory=lambda: []
+    )
+    ranges: List[ConfigInterfaceIPv6RangeOptions] = field(
+        default_factory=lambda: []
+    )
+    is_public: bool = False
+
+
+@dataclass
+class ConfigInterfaceIPv6SLAAC(JSONObject):
+    """
+    ConfigInterfaceIPv6SLAAC represents a single SLAAC address under a VPC interface's IPv6 configuration.
+    """
+
+    put_class = ConfigInterfaceIPv6SLAACOptions
+
+    range: str = ""
+    address: str = ""
+
+
+@dataclass
+class ConfigInterfaceIPv6Range(JSONObject):
+    """
+    ConfigInterfaceIPv6Range represents a single IPv6 address under a VPC interface's IPv6 configuration.
+    """
+
+    put_class = ConfigInterfaceIPv6RangeOptions
+
+    range: str = ""
+
+
+@dataclass
+class ConfigInterfaceIPv6(JSONObject):
+    """
+    ConfigInterfaceIPv6 represents the IPv6 configuration of a VPC interface.
+    """
+
+    put_class = ConfigInterfaceIPv6Options
+
+    slaac: List[ConfigInterfaceIPv6SLAAC] = field(default_factory=lambda: [])
+    ranges: List[ConfigInterfaceIPv6Range] = field(default_factory=lambda: [])
+    is_public: bool = False
+
+
 class NetworkInterface(DerivedBase):
     """
     This class represents a Configuration Profile's network interface object.
@@ -320,6 +393,7 @@ class NetworkInterface(DerivedBase):
         "vpc_id": Property(id_relationship=VPC),
         "subnet_id": Property(),
         "ipv4": Property(mutable=True, json_object=ConfigInterfaceIPv4),
+        "ipv6": Property(mutable=True, json_object=ConfigInterfaceIPv6),
         "ip_ranges": Property(mutable=True),
     }
 
@@ -391,7 +465,10 @@ class ConfigInterface(JSONObject):
     # VPC-specific
     vpc_id: Optional[int] = None
     subnet_id: Optional[int] = None
+
     ipv4: Optional[Union[ConfigInterfaceIPv4, Dict[str, Any]]] = None
+    ipv6: Optional[Union[ConfigInterfaceIPv6, Dict[str, Any]]] = None
+
     ip_ranges: Optional[List[str]] = None
 
     # Computed
@@ -400,7 +477,7 @@ class ConfigInterface(JSONObject):
     def __repr__(self):
         return f"Interface: {self.purpose}"
 
-    def _serialize(self):
+    def _serialize(self, is_put: bool = False):
         purpose_formats = {
             "public": {"purpose": "public", "primary": self.primary},
             "vlan": {
@@ -412,11 +489,8 @@ def _serialize(self):
                 "purpose": "vpc",
                 "primary": self.primary,
                 "subnet_id": self.subnet_id,
-                "ipv4": (
-                    self.ipv4.dict
-                    if isinstance(self.ipv4, ConfigInterfaceIPv4)
-                    else self.ipv4
-                ),
+                "ipv4": self.ipv4,
+                "ipv6": self.ipv6,
                 "ip_ranges": self.ip_ranges,
             },
         }
@@ -426,11 +500,14 @@ def _serialize(self):
                 f"Unknown interface purpose: {self.purpose}",
             )
 
-        return {
-            k: v
-            for k, v in purpose_formats[self.purpose].items()
-            if v is not None
-        }
+        return _flatten_request_body_recursive(
+            {
+                k: v
+                for k, v in purpose_formats[self.purpose].items()
+                if v is not None
+            },
+            is_put=is_put,
+        )
 
 
 class Config(DerivedBase):
@@ -510,16 +587,16 @@ def _populate(self, json):
 
         self._set("devices", MappedObject(**devices))
 
-    def _serialize(self):
+    def _serialize(self, *args, **kwargs):
         """
         Overrides _serialize to transform interfaces into json
         """
-        partial = DerivedBase._serialize(self)
+        partial = DerivedBase._serialize(self, *args, **kwargs)
         interfaces = []
 
         for c in self.interfaces:
             if isinstance(c, ConfigInterface):
-                interfaces.append(c._serialize())
+                interfaces.append(c._serialize(*args, **kwargs))
             else:
                 interfaces.append(c)
 
@@ -571,6 +648,7 @@ def interface_create_vpc(
         subnet: Union[int, VPCSubnet],
         primary=False,
         ipv4: Union[Dict[str, Any], ConfigInterfaceIPv4] = None,
+        ipv6: Union[Dict[str, Any], ConfigInterfaceIPv6Options] = None,
         ip_ranges: Optional[List[str]] = None,
     ) -> NetworkInterface:
         """
@@ -584,6 +662,8 @@ def interface_create_vpc(
         :type primary: bool
         :param ipv4: The IPv4 configuration of the interface for the associated subnet.
         :type ipv4: Dict or ConfigInterfaceIPv4
+        :param ipv6: The IPv6 configuration of the interface for the associated subnet.
+        :type ipv6: Dict or ConfigInterfaceIPv6Options
         :param ip_ranges: A list of IPs or IP ranges in the VPC subnet.
                           Packets to these CIDRs are routed through the
                           VPC network interface.
@@ -594,19 +674,16 @@ def interface_create_vpc(
         """
         params = {
             "purpose": "vpc",
-            "subnet_id": subnet.id if isinstance(subnet, VPCSubnet) else subnet,
+            "subnet_id": subnet,
             "primary": primary,
+            "ipv4": ipv4,
+            "ipv6": ipv6,
+            "ip_ranges": ip_ranges,
         }
 
-        if ipv4 is not None:
-            params["ipv4"] = (
-                ipv4.dict if isinstance(ipv4, ConfigInterfaceIPv4) else ipv4
-            )
-
-        if ip_ranges is not None:
-            params["ip_ranges"] = ip_ranges
-
-        return self._interface_create(params)
+        return self._interface_create(
+            drop_null_keys(_flatten_request_body_recursive(params))
+        )
 
     def interface_reorder(self, interfaces: List[Union[int, NetworkInterface]]):
         """
@@ -1927,8 +2004,8 @@ def _populate(self, json):
         ndist = [Image(self._client, d) for d in self.images]
         self._set("images", ndist)
 
-    def _serialize(self):
-        dct = Base._serialize(self)
+    def _serialize(self, *args, **kwargs):
+        dct = Base._serialize(self, *args, **kwargs)
         dct["images"] = [d.id for d in self.images]
         return dct