Add create private nat gateway endpoint

doc/source/sdk/guides/privatenat.rst

@@ -33,3 +33,12 @@ This interface is used to get a Private NAT gateway by ID
 
 .. literalinclude:: ../examples/natv3/get_private_gateway.py
    :lines: 16-24
+
+Create Private NAT Gateway
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+This interface is used to create a Private NAT gateway
+:class:`~otcextensions.sdk.natv3.v3.gateway.Gateway`.
+
+.. literalinclude:: ../examples/natv3/create_private_gateway.py
+   :lines: 16-35

examples/natv3/create_private_gateway.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+"""
+Create a NAT Gateway
+"""
+
+import openstack
+
+openstack.enable_logging(True)
+conn = openstack.connect(cloud="otc")
+
+downlink_vpc = {
+    "vpc_id": "3cb66d44-9f75-4237-bfff-e37b14d23ad2",
+    "virsubnet_id": "373979ee-f4f0-46c5-80e3-0fbf72646b70",
+    "ngport_ip_address": "10.0.0.17",
+}
+gateway_attrs = {
+    "name": "nat_001",
+    "description": "my private nat gateway 01",
+    "downlink_vpcs": [downlink_vpc],
+    "spec": "1",
+}
+
+gateway = conn.natv3.create_private_nat_gateway(**gateway_attrs)
+print(gateway)

otcextensions/common/utils.py

@@ -170,3 +170,14 @@ def extract_url_parts(url: str, project_id: str) -> list:
 
     # Strip out empty or None segments and return
     return [part for part in url_parts if part]
+
+
+def normalize_tags(tags):
+    result = []
+    for tag in tags:
+        try:
+            tag = tag.split("=")
+            result.append({"key": tag[0], "value": tag[1]})
+        except IndexError:
+            result.append({"key": tag[0], "value": ""})
+    return result

otcextensions/osclient/cbr/v3/vault.py

@@ -19,6 +19,7 @@
 from osc_lib.cli import parseractions
 from osc_lib.command import command
 
+from otcextensions.common.utils import normalize_tags
 from otcextensions.i18n import _
 
 LOG = logging.getLogger(__name__)
@@ -116,17 +117,6 @@ def _add_associated_resources_to_vault_obj(data, columns):
     return return_data, columns
 
 
-def _normalize_tags(tags):
-    result = []
-    for tag in tags:
-        try:
-            tag = tag.split("=")
-            result.append({"key": tag[0], "value": tag[1]})
-        except IndexError:
-            result.append({"key": tag[0], "value": ""})
-    return result
-
-
 class ListVaults(command.Lister):
     _description = _("List CBR Vaults")
     columns = ("ID", "name", "backup_policy_id", "description", "created_at")
@@ -428,15 +418,15 @@ def take_action(self, parsed_args):
             attrs["backup_policy_id"] = parsed_args.backup_policy
         if parsed_args.bind_rule:
             attrs["bind_rules"] = {"tags": []}
-            attrs["bind_rules"]["tags"] = _normalize_tags(parsed_args.bind_rule)
+            attrs["bind_rules"]["tags"] = normalize_tags(parsed_args.bind_rule)
         if parsed_args.description:
             attrs["description"] = parsed_args.description
         if parsed_args.enterprise_project_id:
             attrs["enterprise_project_id"] = parsed_args.enterprise_project_id
         if parsed_args.auto_bind:
             attrs["auto_bind"] = parsed_args.auto_bind
         if parsed_args.tag:
-            attrs["tags"] = _normalize_tags(parsed_args.tag)
+            attrs["tags"] = normalize_tags(parsed_args.tag)
 
         client = self.app.client_manager.cbr
         obj = client.create_vault(**attrs)
@@ -547,7 +537,7 @@ def take_action(self, parsed_args):
             attrs["auto_bind"] = parsed_args.auto_bind
         if parsed_args.bind_rule:
             attrs["bind_rules"] = {"tags": []}
-            attrs["bind_rules"]["tags"] = _normalize_tags(parsed_args.bind_rule)
+            attrs["bind_rules"]["tags"] = normalize_tags(parsed_args.bind_rule)
         if parsed_args.auto_expand:
             attrs["auto_expand"] = parsed_args.auto_expand
         if parsed_args.smn_notify:

otcextensions/osclient/privatenat/v3/private_nat_gateway.py

@@ -15,9 +15,11 @@
 import logging
 
 from osc_lib import utils
+from osc_lib.cli import parseractions
 from osc_lib.command import command
 
 from otcextensions.common import sdk_utils
+from otcextensions.common.utils import normalize_tags
 from otcextensions.i18n import _
 
 LOG = logging.getLogger(__name__)
@@ -28,6 +30,51 @@ def _get_columns(item):
     return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
 
 
+def _flatten_private_nat_gateway(obj):
+    return {
+        "id": obj.id,
+        "name": obj.name,
+        "description": obj.description,
+        "spec": obj.spec,
+        "project_id": obj.project_id,
+        "enterprise_project_id": obj.enterprise_project_id,
+        "status": obj.status,
+        "created_at": obj.created_at,
+        "updated_at": obj.updated_at,
+        "rule_max": obj.rule_max,
+        "transit_ip_pool_size_max": obj.transit_ip_pool_size_max,
+    }
+
+
+def _add_downlink_vpcs_to_obj(obj, data, columns):
+    for index, downlink_vpc in enumerate(obj.downlink_vpcs or [], start=1):
+        data += (downlink_vpc.get("vpc_id", ""),)
+        columns += ("downlink_vpc_id_%s" % index,)
+
+        data += (downlink_vpc.get("virsubnet_id", ""),)
+        columns += ("downlink_virsubnet_id_%s" % index,)
+
+        data += (downlink_vpc.get("ngport_ip_address", ""),)
+        columns += ("downlink_ngport_ip_address_%s" % index,)
+
+    return data, columns
+
+
+def _add_tags_to_obj(obj, data, columns):
+    data += (
+        "\n".join(
+            "key=%s, value=%s"
+            % (
+                tag.get("key", ""),
+                tag.get("value", ""),
+            )
+            for tag in obj.tags
+        ),
+    )
+    columns += ("tags",)
+    return data, columns
+
+
 class ListPrivateNatGateways(command.Lister):
 
     _description = _("List Private NAT Gateways.")
@@ -177,3 +224,123 @@ def take_action(self, parsed_args):
         data = utils.get_item_properties(obj, columns)
 
         return display_columns, data
+
+
+class CreatePrivateNatGateway(command.ShowOne):
+    _description = _("Create new Private NAT Gateway")
+    columns = (
+        "id",
+        "name",
+        "description",
+        "spec",
+        "project_id",
+        "enterprise_project_id",
+        "status",
+        "created_at",
+        "updated_at",
+        "rule_max",
+        "transit_ip_pool_size_max",
+    )
+
+    def get_parser(self, prog_name):
+        parser = super(CreatePrivateNatGateway, self).get_parser(prog_name)
+        parser.add_argument(
+            "--name",
+            metavar="<name>",
+            required=True,
+            help=_("Specifies the name of the Private NAT Gateway."),
+        )
+        parser.add_argument(
+            "--description",
+            metavar="<description>",
+            help=_(
+                "Provides supplementary information about " "the Private NAT Gateway."
+            ),
+        )
+        parser.add_argument(
+            "--spec",
+            metavar="<spec>",
+            default="Small",
+            help=_(
+                "Specifies the type of the Private NAT Gateway. "
+                "The value can be:\n"
+                "1: small type, which supports up to 10,000 "
+                "SNAT connections.\n"
+                "2: medium type, which supports up to 50,000 "
+                "SNAT connections.\n"
+                "3: large type, which supports up to 200,000 "
+                "SNAT connections.\n"
+                "4: extra-large type, which supports up to "
+                "1,000,000 SNAT connections."
+            ),
+        )
+        parser.add_argument(
+            "--enterprise-project-id",
+            metavar="<enterprise_project_id>",
+            default=0,
+            help=_(
+                "Specifies the ID of the enterprise project"
+                " that is associated with the private NAT gateway"
+                " when the private NAT gateway is created."
+            ),
+        )
+        parser.add_argument(
+            "--downlink-vpc",
+            metavar="<virsubnet_id=virsubnet_id[,ngport_ip_address=ip]>",
+            action=parseractions.MultiKeyValueAction,
+            dest="downlink_vpcs",
+            required_keys=["virsubnet_id"],
+            optional_keys=["ngport_ip_address"],
+            help=_("Specifies the VPC where the private NAT gateway works."),
+        )
+        parser.add_argument(
+            "--tags",
+            metavar="<tags>",
+            action="append",
+            help=_(
+                "Specifies the tag list in KEY=VALUE format."
+                "Repeat for multiple values."
+            ),
+        )
+
+        return parser
+
+    def _build_attrs(self, parsed_args):
+        attrs = {
+            "name": parsed_args.name,
+            "downlink_vpcs": parsed_args.downlink_vpcs,
+        }
+
+        optional_attrs = (
+            "description",
+            "spec",
+            "enterprise_project_id",
+        )
+        for key in optional_attrs:
+            value = getattr(parsed_args, key, None)
+            if value is not None:
+                attrs[key] = value
+
+        if parsed_args.tags:
+            attrs["tags"] = normalize_tags(parsed_args.tags)
+
+        return attrs
+
+    def take_action(self, parsed_args):
+        client = self.app.client_manager.privatenat
+
+        attrs = self._build_attrs(parsed_args)
+        obj = client.create_private_nat_gateway(**attrs)
+
+        columns = self.columns
+        data = utils.get_dict_properties(
+            _flatten_private_nat_gateway(obj),
+            columns,
+        )
+
+        if obj.downlink_vpcs:
+            data, columns = _add_downlink_vpcs_to_obj(obj, data, columns)
+        if obj.tags:
+            data, columns = _add_tags_to_obj(obj, data, columns)
+
+        return columns, data

otcextensions/sdk/natv3/v3/_proxy.py

@@ -42,3 +42,16 @@ def get_private_nat_gateway(self, gateway):
             when no resource can be found.
         """
         return self._get(_gateway.PrivateNatGateway, gateway)
+
+    def create_private_nat_gateway(self, **attrs):
+        """Create a new Private gateway from attributes
+
+        :param dict attrs: Keyword arguments which will be used to create
+            a :class:`~otcextensions.sdk.nat.v2.gateway.Gateway`,
+            comprised of the properties on the Gateway class.
+
+        :returns: The results of the Gateway Creation
+
+        :rtype: :class:`~otcextensions.sdk.nat.v3.gateway.PrivateNatGateway`
+        """
+        return self._create(_gateway.PrivateNatGateway, **attrs)

otcextensions/sdk/natv3/v3/gateway.py

@@ -39,6 +39,7 @@ class PrivateNatGateway(resource.Resource):
 
     allow_list = True
     allow_fetch = True
+    allow_create = True
 
     _query_mapping = resource.QueryParameters(
         "description",

otcextensions/tests/unit/common/test_utils.py

@@ -0,0 +1,30 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from openstack.tests.unit import base
+from otcextensions.common.utils import normalize_tags
+
+
+class TestUtils(base.TestCase):
+
+    def test_normalize_tags(self):
+        tags = ["key1=value", "key2=", "key3"]
+
+        verify_result = [
+            {"key": "key1", "value": "value"},
+            {"key": "key2", "value": ""},
+            {"key": "key3", "value": ""},
+        ]
+
+        result = normalize_tags(tags)
+
+        self.assertEqual(result, verify_result)

otcextensions/tests/unit/osclient/cbr/v3/test_vault.py

@@ -164,19 +164,6 @@ def test_add_associated_resources_to_vault_output(self):
         self.assertEqual(data, verify_data)
         self.assertEqual(column, verify_column)
 
-    def test_normalize_tags(self):
-        tags = ["key1=value", "key2=", "key3"]
-
-        verify_result = [
-            {"key": "key1", "value": "value"},
-            {"key": "key2", "value": ""},
-            {"key": "key3", "value": ""},
-        ]
-
-        result = vault._normalize_tags(tags)
-
-        self.assertEqual(result, verify_result)
-
 
 class TestListVault(fakes.TestCBR):