Add create private nat gateway endpoint

shhovhan · shhovhan · commit f290fdb59f46 · 2026-03-31T00:33:49.000+04:00
diff --git a/doc/source/sdk/guides/privatenat.rst b/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
diff --git a/examples/natv3/create_private_gateway.py b/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)
diff --git a/otcextensions/common/utils.py b/otcextensions/common/utils.py
@@ -170,3 +170,23 @@ 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):
+    """Convert list of KEY=VALUE strings to API tag dicts."""
+    result = []
+
+    for tag in tags:
+        if "=" not in tag:
+            raise ValueError("Invalid tag '%s'. Expected format key=value" % tag)
+
+        key, value = tag.split("=", 1)
+        key = key.strip()
+        value = value.strip()
+
+        if not key:
+            raise ValueError("Tag key cannot be empty")
+
+        result.append({"key": key, "value": value})
+
+    return result
diff --git a/otcextensions/osclient/cbr/v3/vault.py b/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:
diff --git a/otcextensions/osclient/privatenat/v3/private_nat_gateway.py b/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
diff --git a/otcextensions/sdk/natv3/v3/_proxy.py b/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)
diff --git a/otcextensions/sdk/natv3/v3/gateway.py b/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",
diff --git a/otcextensions/tests/unit/osclient/privatenat/v3/fakes.py b/otcextensions/tests/unit/osclient/privatenat/v3/fakes.py
@@ -38,10 +38,29 @@ def generate(cls):
         object_info = {
             "id": "id-" + uuid.uuid4().hex,
             "name": "name-" + uuid.uuid4().hex,
+            "description": "private_nat_gateway_description",
             "spec": "Small",
             "status": "ACTIVE",
             "project_id": "project-" + uuid.uuid4().hex,
             "enterprise_project_id": "ep-" + uuid.uuid4().hex,
+            "created_at": "2024-01-01T10:00:00",
+            "updated_at": "2024-01-01T10:00:00",
+            "rule_max": 20,
+            "transit_ip_pool_size_max": 1,
+            "downlink_vpcs": [
+                {
+                    "vpc_id": "vpc-" + uuid.uuid4().hex,
+                    "virsubnet_id": "subnet-" + uuid.uuid4().hex,
+                    "ngport_ip_address": "192.168.10.10",
+                }
+            ],
+            "tags": [
+                {
+                    "key": "key1",
+                    "value": "value1",
+                }
+            ],
         }
 
-        return gateway.PrivateNatGateway(**object_info)
+        obj = gateway.PrivateNatGateway.existing(**object_info)
+        return obj
diff --git a/otcextensions/tests/unit/osclient/privatenat/v3/test_private_gateway.py b/otcextensions/tests/unit/osclient/privatenat/v3/test_private_gateway.py
diff --git a/otcextensions/tests/unit/sdk/natv3/v3/test_proxy.py b/otcextensions/tests/unit/sdk/natv3/v3/test_proxy.py
diff --git a/setup.cfg b/setup.cfg
