Implement support for VPC Dual Stack

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(

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, *args, **kwargs):
+    def _serialize(self, is_put: bool = False):
         purpose_formats = {
             "public": {"purpose": "public", "primary": self.primary},
             "vlan": {
@@ -412,11 +489,8 @@ def _serialize(self, *args, **kwargs):
                 "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, *args, **kwargs):
                 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):
@@ -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]]):
         """

linode_api4/objects/networking.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import List, Optional
 
 from linode_api4.common import Price, RegionPrice
 from linode_api4.errors import UnexpectedResponseError
@@ -127,6 +127,11 @@ def delete(self):
         return True
 
 
+@dataclass
+class VPCIPAddressIPv6(JSONObject):
+    slaac_address: str = ""
+
+
 @dataclass
 class VPCIPAddress(JSONObject):
     """
@@ -152,6 +157,10 @@ class VPCIPAddress(JSONObject):
     address_range: Optional[str] = None
     nat_1_1: Optional[str] = None
 
+    ipv6_range: Optional[str] = None
+    ipv6_is_public: Optional[bool] = None
+    ipv6_addresses: Optional[List[VPCIPAddressIPv6]] = None
+
 
 class VLAN(Base):
     """

linode_api4/objects/vpc.py

@@ -1,11 +1,54 @@
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 from linode_api4.errors import UnexpectedResponseError
 from linode_api4.objects import Base, DerivedBase, Property, Region
+from linode_api4.objects.base import _flatten_request_body_recursive
 from linode_api4.objects.networking import VPCIPAddress
 from linode_api4.objects.serializable import JSONObject
 from linode_api4.paginated_list import PaginatedList
+from linode_api4.util import drop_null_keys
+
+
+@dataclass
+class VPCIPv6RangeOptions(JSONObject):
+    """
+    VPCIPv6RangeOptions is used to specify an IPv6 range when creating or updating a VPC.
+    """
+
+    range: str = ""
+    allocation_class: Optional[str] = None
+
+
+@dataclass
+class VPCIPv6Range(JSONObject):
+    """
+    VPCIPv6Range represents a single VPC IPv6 range.
+    """
+
+    put_class = VPCIPv6RangeOptions
+
+    range: str = ""
+
+
+@dataclass
+class VPCSubnetIPv6RangeOptions(JSONObject):
+    """
+    VPCSubnetIPv6RangeOptions is used to specify an IPv6 range when creating or updating a VPC subnet.
+    """
+
+    range: str = ""
+
+
+@dataclass
+class VPCSubnetIPv6Range(JSONObject):
+    """
+    VPCSubnetIPv6Range represents a single VPC subnet IPv6 range.
+    """
+
+    put_class = VPCSubnetIPv6RangeOptions
+
+    range: str = ""
 
 
 @dataclass
@@ -35,6 +78,7 @@ class VPCSubnet(DerivedBase):
         "id": Property(identifier=True),
         "label": Property(mutable=True),
         "ipv4": Property(),
+        "ipv6": Property(json_object=VPCSubnetIPv6Range, unordered=True),
         "linodes": Property(json_object=VPCSubnetLinode, unordered=True),
         "created": Property(is_datetime=True),
         "updated": Property(is_datetime=True),
@@ -55,6 +99,7 @@ class VPC(Base):
         "label": Property(mutable=True),
         "description": Property(mutable=True),
         "region": Property(slug_relationship=Region),
+        "ipv6": Property(json_object=VPCIPv6Range, unordered=True),
         "subnets": Property(derived_class=VPCSubnet),
         "created": Property(is_datetime=True),
         "updated": Property(is_datetime=True),
@@ -64,6 +109,9 @@ def subnet_create(
         self,
         label: str,
         ipv4: Optional[str] = None,
+        ipv6: Optional[
+            List[Union[VPCSubnetIPv6RangeOptions, Dict[str, Any]]]
+        ] = None,
         **kwargs,
     ) -> VPCSubnet:
         """
@@ -76,19 +124,16 @@ def subnet_create(
         :param ipv4: The IPv4 range of this subnet in CIDR format.
         :type ipv4: str
         :param ipv6: The IPv6 range of this subnet in CIDR format.
-        :type ipv6: str
+        :type ipv6: List[Union[VPCSubnetIPv6RangeOptions, Dict[str, Any]]]
         """
-        params = {
-            "label": label,
-        }
-
-        if ipv4 is not None:
-            params["ipv4"] = ipv4
+        params = {"label": label, "ipv4": ipv4, "ipv6": ipv6}
 
         params.update(kwargs)
 
         result = self._client.post(
-            "{}/subnets".format(VPC.api_endpoint), model=self, data=params
+            "{}/subnets".format(VPC.api_endpoint),
+            model=self,
+            data=drop_null_keys(_flatten_request_body_recursive(params)),
         )
         self.invalidate()