feat(vpc): set acl calls to public visibility (#913)

Co-authored-by: Jonathan R. <jremy@scaleway.com>

Author: scaleway-bot

scaleway-async/scaleway_async/vpc/v2/__init__.py

@@ -1,11 +1,14 @@
 # This file was automatically generated. DO NOT EDIT.
 # If you have any remark or suggestion do not hesitate to open an issue.
+from .types import AclRuleProtocol
+from .types import Action
 from .types import ListPrivateNetworksRequestOrderBy
 from .types import ListSubnetsRequestOrderBy
 from .types import ListVPCsRequestOrderBy
 from .types import Subnet
 from .types import PrivateNetwork
 from .types import Route
+from .types import AclRule
 from .types import VPC
 from .types import AddSubnetsRequest
 from .types import AddSubnetsResponse
@@ -19,6 +22,8 @@
 from .types import DeleteVPCRequest
 from .types import EnableDHCPRequest
 from .types import EnableRoutingRequest
+from .types import GetAclRequest
+from .types import GetAclResponse
 from .types import GetPrivateNetworkRequest
 from .types import GetRouteRequest
 from .types import GetVPCRequest
@@ -28,6 +33,8 @@
 from .types import ListSubnetsResponse
 from .types import ListVPCsRequest
 from .types import ListVPCsResponse
+from .types import SetAclRequest
+from .types import SetAclResponse
 from .types import SetSubnetsRequest
 from .types import SetSubnetsResponse
 from .types import UpdatePrivateNetworkRequest
@@ -36,12 +43,15 @@
 from .api import VpcV2API
 
 __all__ = [
+    "AclRuleProtocol",
+    "Action",
     "ListPrivateNetworksRequestOrderBy",
     "ListSubnetsRequestOrderBy",
     "ListVPCsRequestOrderBy",
     "Subnet",
     "PrivateNetwork",
     "Route",
+    "AclRule",
     "VPC",
     "AddSubnetsRequest",
     "AddSubnetsResponse",
@@ -55,6 +65,8 @@
     "DeleteVPCRequest",
     "EnableDHCPRequest",
     "EnableRoutingRequest",
+    "GetAclRequest",
+    "GetAclResponse",
     "GetPrivateNetworkRequest",
     "GetRouteRequest",
     "GetVPCRequest",
@@ -64,6 +76,8 @@
     "ListSubnetsResponse",
     "ListVPCsRequest",
     "ListVPCsResponse",
+    "SetAclRequest",
+    "SetAclResponse",
     "SetSubnetsRequest",
     "SetSubnetsResponse",
     "UpdatePrivateNetworkRequest",

scaleway-async/scaleway_async/vpc/v2/api.py

@@ -13,21 +13,26 @@
     fetch_all_pages_async,
 )
 from .types import (
+    Action,
     ListPrivateNetworksRequestOrderBy,
     ListSubnetsRequestOrderBy,
     ListVPCsRequestOrderBy,
+    AclRule,
     AddSubnetsRequest,
     AddSubnetsResponse,
     CreatePrivateNetworkRequest,
     CreateRouteRequest,
     CreateVPCRequest,
     DeleteSubnetsRequest,
     DeleteSubnetsResponse,
+    GetAclResponse,
     ListPrivateNetworksResponse,
     ListSubnetsResponse,
     ListVPCsResponse,
     PrivateNetwork,
     Route,
+    SetAclRequest,
+    SetAclResponse,
     SetSubnetsRequest,
     SetSubnetsResponse,
     Subnet,
@@ -42,15 +47,18 @@
     unmarshal_VPC,
     unmarshal_AddSubnetsResponse,
     unmarshal_DeleteSubnetsResponse,
+    unmarshal_GetAclResponse,
     unmarshal_ListPrivateNetworksResponse,
     unmarshal_ListSubnetsResponse,
     unmarshal_ListVPCsResponse,
+    unmarshal_SetAclResponse,
     unmarshal_SetSubnetsResponse,
     marshal_AddSubnetsRequest,
     marshal_CreatePrivateNetworkRequest,
     marshal_CreateRouteRequest,
     marshal_CreateVPCRequest,
     marshal_DeleteSubnetsRequest,
+    marshal_SetAclRequest,
     marshal_SetSubnetsRequest,
     marshal_UpdatePrivateNetworkRequest,
     marshal_UpdateRouteRequest,
@@ -1112,3 +1120,96 @@ async def delete_route(
         )
 
         self._throw_on_error(res)
+
+    async def get_acl(
+        self,
+        *,
+        vpc_id: str,
+        is_ipv6: bool,
+        region: Optional[ScwRegion] = None,
+    ) -> GetAclResponse:
+        """
+        Get Acl Rules for VPC.
+        Retrieve a list of ACL rules for a VPC, specified by its VPC ID.
+        :param vpc_id: ID of the Network ACL's VPC.
+        :param is_ipv6: Defines whether this set of ACL rules is for IPv6 (false = IPv4). Each Network ACL can have rules for only one IP type.
+        :param region: Region to target. If none is passed will use default region from the config.
+        :return: :class:`GetAclResponse <GetAclResponse>`
+
+        Usage:
+        ::
+
+            result = await api.get_acl(
+                vpc_id="example",
+                is_ipv6=False,
+            )
+        """
+
+        param_region = validate_path_param(
+            "region", region or self.client.default_region
+        )
+        param_vpc_id = validate_path_param("vpc_id", vpc_id)
+
+        res = self._request(
+            "GET",
+            f"/vpc/v2/regions/{param_region}/vpc/{param_vpc_id}/acl-rules",
+            params={
+                "is_ipv6": is_ipv6,
+            },
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_GetAclResponse(res.json())
+
+    async def set_acl(
+        self,
+        *,
+        vpc_id: str,
+        rules: List[AclRule],
+        is_ipv6: bool,
+        default_policy: Action,
+        region: Optional[ScwRegion] = None,
+    ) -> SetAclResponse:
+        """
+        Set VPC ACL rules.
+        Set the list of ACL rules and the default routing policy for a VPC.
+        :param vpc_id: ID of the Network ACL's VPC.
+        :param rules: List of Network ACL rules.
+        :param is_ipv6: Defines whether this set of ACL rules is for IPv6 (false = IPv4). Each Network ACL can have rules for only one IP type.
+        :param default_policy: Action to take for packets which do not match any rules.
+        :param region: Region to target. If none is passed will use default region from the config.
+        :return: :class:`SetAclResponse <SetAclResponse>`
+
+        Usage:
+        ::
+
+            result = await api.set_acl(
+                vpc_id="example",
+                rules=[],
+                is_ipv6=False,
+                default_policy=Action.unknown_action,
+            )
+        """
+
+        param_region = validate_path_param(
+            "region", region or self.client.default_region
+        )
+        param_vpc_id = validate_path_param("vpc_id", vpc_id)
+
+        res = self._request(
+            "PUT",
+            f"/vpc/v2/regions/{param_region}/vpc/{param_vpc_id}/acl-rules",
+            body=marshal_SetAclRequest(
+                SetAclRequest(
+                    vpc_id=vpc_id,
+                    rules=rules,
+                    is_ipv6=is_ipv6,
+                    default_policy=default_policy,
+                    region=region,
+                ),
+                self.client,
+            ),
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_SetAclResponse(res.json())

scaleway-async/scaleway_async/vpc/v2/marshalling.py

@@ -12,15 +12,19 @@
     VPC,
     AddSubnetsResponse,
     DeleteSubnetsResponse,
+    AclRule,
+    GetAclResponse,
     ListPrivateNetworksResponse,
     ListSubnetsResponse,
     ListVPCsResponse,
+    SetAclResponse,
     SetSubnetsResponse,
     AddSubnetsRequest,
     CreatePrivateNetworkRequest,
     CreateRouteRequest,
     CreateVPCRequest,
     DeleteSubnetsRequest,
+    SetAclRequest,
     SetSubnetsRequest,
     UpdatePrivateNetworkRequest,
     UpdateRouteRequest,
@@ -284,6 +288,76 @@ def unmarshal_DeleteSubnetsResponse(data: Any) -> DeleteSubnetsResponse:
     return DeleteSubnetsResponse(**args)
 
 
+def unmarshal_AclRule(data: Any) -> AclRule:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'AclRule' failed as data isn't a dictionary."
+        )
+
+    args: Dict[str, Any] = {}
+
+    field = data.get("protocol", None)
+    if field is not None:
+        args["protocol"] = field
+
+    field = data.get("source", None)
+    if field is not None:
+        args["source"] = field
+
+    field = data.get("src_port_low", None)
+    if field is not None:
+        args["src_port_low"] = field
+
+    field = data.get("src_port_high", None)
+    if field is not None:
+        args["src_port_high"] = field
+
+    field = data.get("destination", None)
+    if field is not None:
+        args["destination"] = field
+
+    field = data.get("dst_port_low", None)
+    if field is not None:
+        args["dst_port_low"] = field
+
+    field = data.get("dst_port_high", None)
+    if field is not None:
+        args["dst_port_high"] = field
+
+    field = data.get("action", None)
+    if field is not None:
+        args["action"] = field
+
+    field = data.get("description", None)
+    if field is not None:
+        args["description"] = field
+    else:
+        args["description"] = None
+
+    return AclRule(**args)
+
+
+def unmarshal_GetAclResponse(data: Any) -> GetAclResponse:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'GetAclResponse' failed as data isn't a dictionary."
+        )
+
+    args: Dict[str, Any] = {}
+
+    field = data.get("rules", None)
+    if field is not None:
+        args["rules"] = (
+            [unmarshal_AclRule(v) for v in field] if field is not None else None
+        )
+
+    field = data.get("default_policy", None)
+    if field is not None:
+        args["default_policy"] = field
+
+    return GetAclResponse(**args)
+
+
 def unmarshal_ListPrivateNetworksResponse(data: Any) -> ListPrivateNetworksResponse:
     if not isinstance(data, dict):
         raise TypeError(
@@ -345,6 +419,27 @@ def unmarshal_ListVPCsResponse(data: Any) -> ListVPCsResponse:
     return ListVPCsResponse(**args)
 
 
+def unmarshal_SetAclResponse(data: Any) -> SetAclResponse:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'SetAclResponse' failed as data isn't a dictionary."
+        )
+
+    args: Dict[str, Any] = {}
+
+    field = data.get("rules", None)
+    if field is not None:
+        args["rules"] = (
+            [unmarshal_AclRule(v) for v in field] if field is not None else None
+        )
+
+    field = data.get("default_policy", None)
+    if field is not None:
+        args["default_policy"] = field
+
+    return SetAclResponse(**args)
+
+
 def unmarshal_SetSubnetsResponse(data: Any) -> SetSubnetsResponse:
     if not isinstance(data, dict):
         raise TypeError(
@@ -456,6 +551,60 @@ def marshal_DeleteSubnetsRequest(
     return output
 
 
+def marshal_AclRule(
+    request: AclRule,
+    defaults: ProfileDefaults,
+) -> Dict[str, Any]:
+    output: Dict[str, Any] = {}
+
+    if request.protocol is not None:
+        output["protocol"] = str(request.protocol)
+
+    if request.source is not None:
+        output["source"] = request.source
+
+    if request.src_port_low is not None:
+        output["src_port_low"] = request.src_port_low
+
+    if request.src_port_high is not None:
+        output["src_port_high"] = request.src_port_high
+
+    if request.destination is not None:
+        output["destination"] = request.destination
+
+    if request.dst_port_low is not None:
+        output["dst_port_low"] = request.dst_port_low
+
+    if request.dst_port_high is not None:
+        output["dst_port_high"] = request.dst_port_high
+
+    if request.action is not None:
+        output["action"] = str(request.action)
+
+    if request.description is not None:
+        output["description"] = request.description
+
+    return output
+
+
+def marshal_SetAclRequest(
+    request: SetAclRequest,
+    defaults: ProfileDefaults,
+) -> Dict[str, Any]:
+    output: Dict[str, Any] = {}
+
+    if request.rules is not None:
+        output["rules"] = [marshal_AclRule(item, defaults) for item in request.rules]
+
+    if request.is_ipv6 is not None:
+        output["is_ipv6"] = request.is_ipv6
+
+    if request.default_policy is not None:
+        output["default_policy"] = str(request.default_policy)
+
+    return output
+
+
 def marshal_SetSubnetsRequest(
     request: SetSubnetsRequest,
     defaults: ProfileDefaults,