ADT CLI features and functionality around private endpoints + links.

azext_iot/digitaltwins/_help.py

@@ -209,6 +209,95 @@ def load_digitaltwins_help():
             az dt endpoint delete -n {instance_name} --endpoint-name {endpoint_name} -y --no-wait
     """
 
+    helps["dt network"] = """
+        type: group
+        short-summary: Manage Digital Twins network configuration including private links and endpoint connections.
+    """
+
+    helps["dt network private-link"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-link operations.
+    """
+
+    helps["dt network private-link show"] = """
+        type: command
+        short-summary: Show a private-link associated with the instance.
+
+        examples:
+        - name: Show the private-link named "API" associated with the instance.
+          text: >
+            az dt network private-link show -n {instance_name} --link-name API
+    """
+
+    helps["dt network private-link list"] = """
+        type: command
+        short-summary: List private-links associated with the Digital Twins instance.
+
+        examples:
+        - name: List all private-links associated with the instance.
+          text: >
+            az dt network private-link list -n {instance_name}
+    """
+
+    helps["dt network private-endpoint"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-endpoints.
+        long-summary: Use 'az network private-endpoint create' to create a private-endpoint and link to a Digital Twins resource.
+    """
+
+    helps["dt network private-endpoint connection"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-endpoint connections.
+    """
+
+    helps["dt network private-endpoint connection list"] = """
+        type: command
+        short-summary: List private-endpoint connections associated with the Digital Twins instance.
+
+        examples:
+        - name: List all private-endpoint connections associated with the instance.
+          text: >
+            az dt network private-endpoint connection list -n {instance_name}
+    """
+
+    helps["dt network private-endpoint connection show"] = """
+        type: command
+        short-summary: Show a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Show details of the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f.
+          text: >
+            az dt network private-endpoint connection show -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f
+    """
+
+    helps["dt network private-endpoint connection set"] = """
+        type: command
+        short-summary: Set the state of a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Approve a pending private-endpoint connection associated with the instance and add a description.
+          text: >
+            az dt network private-endpoint connection set -n {instance_name} --cn {connection_name} --status Approved --desc "A description."
+
+        - name: Reject a private-endpoint connection associated with the instance and add a description.
+          text: >
+            az dt network private-endpoint connection set -n {instance_name} --cn {connection_name} --status Rejected --desc "Does not comply."
+    """
+
+    helps["dt network private-endpoint connection delete"] = """
+        type: command
+        short-summary: Delete a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Delete the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f with confirmation. Block until finished.
+          text: >
+            az dt network private-endpoint connection delete -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f
+
+        - name: Delete the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f no confirmation. Return immediately.
+          text: >
+            az dt network private-endpoint connection delete -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f -y --no-wait
+    """
+
     helps["dt role-assignment"] = """
         type: group
         short-summary: Manage RBAC role assignments for a Digital Twins instance.

azext_iot/digitaltwins/command_map.py

@@ -4,8 +4,6 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from azure.cli.core.profiles import ResourceType
-
 """
 Load CLI commands
 """
@@ -39,7 +37,6 @@ def load_digitaltwins_commands(self, _):
     with self.command_group(
         "dt",
         command_type=digitaltwins_resource_ops,
-        resource_type=ResourceType.MGMT_RESOURCE_RESOURCES,
     ) as cmd_group:
         cmd_group.command("create", "create_instance")
         cmd_group.show_command("show", "show_instance")
@@ -140,3 +137,31 @@ def load_digitaltwins_commands(self, _):
         )
         cmd_group.command("update", "update_model")
         cmd_group.command("delete", "delete_model")
+
+    with self.command_group(
+        "dt network",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        pass
+
+    with self.command_group(
+        "dt network private-link",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        cmd_group.show_command("show", "show_private_link")
+        cmd_group.command("list", "list_private_links")
+
+    with self.command_group(
+        "dt network private-endpoint",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        pass
+
+    with self.command_group(
+        "dt network private-endpoint connection",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        cmd_group.command("set", "set_private_endpoint_conn")
+        cmd_group.show_command("show", "show_private_endpoint_conn")
+        cmd_group.command("list", "list_private_endpoint_conns")
+        cmd_group.command("delete", "delete_private_endpoint_conn", confirmation=True, supports_no_wait=True)

azext_iot/digitaltwins/commands_resource.py

@@ -5,7 +5,11 @@
 # --------------------------------------------------------------------------------------------
 
 from azext_iot.digitaltwins.providers.resource import ResourceProvider
-from azext_iot.digitaltwins.common import ADTEndpointType, ADTEndpointAuthType
+from azext_iot.digitaltwins.common import (
+    ADTEndpointType,
+    ADTEndpointAuthType,
+    ADTPublicNetworkAccessType,
+)
 from knack.log import get_logger
 
 logger = get_logger(__name__)
@@ -20,6 +24,7 @@ def create_instance(
     assign_identity=None,
     scopes=None,
     role_type="Contributor",
+    public_network_access=ADTPublicNetworkAccessType.enabled.value,
 ):
     rp = ResourceProvider(cmd)
     return rp.create(
@@ -30,6 +35,7 @@ def create_instance(
         assign_identity=assign_identity,
         scopes=scopes,
         role_type=role_type,
+        public_network_access=public_network_access,
     )
 
 
@@ -157,3 +163,58 @@ def add_endpoint_eventhub(
         dead_letter_secret=dead_letter_secret,
         auth_type=auth_type,
     )
+
+
+def show_private_link(cmd, name, link_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.get_private_link(
+        name=name, resource_group_name=resource_group_name, link_name=link_name
+    )
+
+
+def list_private_links(cmd, name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.list_private_links(name=name, resource_group_name=resource_group_name)
+
+
+def set_private_endpoint_conn(
+    cmd,
+    name,
+    conn_name,
+    status,
+    description=None,
+    group_ids=None,
+    actions_required=None,
+    resource_group_name=None,
+):
+    rp = ResourceProvider(cmd)
+    return rp.set_private_endpoint_conn(
+        name=name,
+        resource_group_name=resource_group_name,
+        conn_name=conn_name,
+        status=status,
+        description=description,
+        group_ids=group_ids,
+        actions_required=actions_required,
+    )
+
+
+def show_private_endpoint_conn(cmd, name, conn_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.get_private_endpoint_conn(
+        name=name, resource_group_name=resource_group_name, conn_name=conn_name
+    )
+
+
+def list_private_endpoint_conns(cmd, name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.list_private_endpoint_conns(
+        name=name, resource_group_name=resource_group_name
+    )
+
+
+def delete_private_endpoint_conn(cmd, name, conn_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.delete_private_endpoint_conn(
+        name=name, resource_group_name=resource_group_name, conn_name=conn_name
+    )

azext_iot/digitaltwins/common.py

@@ -14,7 +14,7 @@
 
 class ADTEndpointType(Enum):
     """
-    ADT Endpoint Type.
+    ADT endpoint type.
     """
 
     eventgridtopic = "eventgridtopic"
@@ -24,8 +24,28 @@ class ADTEndpointType(Enum):
 
 class ADTEndpointAuthType(Enum):
     """
-    ADT Endpoint Auth Type.
+    ADT endpoint auth type.
     """
 
     identitybased = "IdentityBased"
     keybased = "KeyBased"
+
+
+class ADTPrivateConnectionStatusType(Enum):
+    """
+    ADT private endpoint connection status type.
+    """
+
+    pending = "Pending"
+    approved = "Approved"
+    rejected = "Rejected"
+    disconnected = "Disconnected"
+
+
+class ADTPublicNetworkAccessType(Enum):
+    """
+    ADT private endpoint connection status type.
+    """
+
+    enabled = "Enabled"
+    disabled = "Disabled"

azext_iot/digitaltwins/params.py

@@ -15,7 +15,11 @@
     get_enum_type,
     tags_type,
 )
-from azext_iot.digitaltwins.common import ADTEndpointAuthType
+from azext_iot.digitaltwins.common import (
+    ADTEndpointAuthType,
+    ADTPrivateConnectionStatusType,
+    ADTPublicNetworkAccessType,
+)
 
 depfor_type = CLIArgumentType(
     options_list=["--dependencies-for"],
@@ -72,10 +76,14 @@ def load_digitaltwins_arguments(self, _):
             help="Event route name.",
         )
         context.argument(
-            "filter", options_list=["--filter"], help="Event route filter.",
+            "filter",
+            options_list=["--filter"],
+            help="Event route filter.",
         )
         context.argument(
-            "role_type", options_list=["--role"], help="Role name or Id.",
+            "role_type",
+            options_list=["--role"],
+            help="Role name or Id.",
         )
         context.argument(
             "assignee",
@@ -89,7 +97,9 @@ def load_digitaltwins_arguments(self, _):
             help="Digital Twins model Id. Example: dtmi:com:example:Room;2",
         )
         context.argument(
-            "twin_id", options_list=["--twin-id", "-t"], help="The digital twin Id.",
+            "twin_id",
+            options_list=["--twin-id", "-t"],
+            help="The digital twin Id.",
         )
         context.argument(
             "include_inherited",
@@ -103,6 +113,13 @@ def load_digitaltwins_arguments(self, _):
             options_list=["--top"],
             help="Maximum number of elements to return.",
         )
+        context.argument(
+            "public_network_access",
+            options_list=["--public-network-access", "--pna"],
+            help="Determines if the Digital Twins instance can be accessed from a public network.",
+            arg_group="Networking",
+            arg_type=get_enum_type(ADTPublicNetworkAccessType),
+        )
 
     with self.argument_context("dt create") as context:
         context.argument(
@@ -130,19 +147,19 @@ def load_digitaltwins_arguments(self, _):
             "dead_letter_secret",
             options_list=["--deadletter-sas-uri", "--dsu"],
             help="Dead-letter storage container URL with SAS token for Key based authentication.",
-            arg_group="Dead-letter Endpoint"
+            arg_group="Dead-letter Endpoint",
         )
         context.argument(
             "dead_letter_uri",
             options_list=["--deadletter-uri", "--du"],
             help="Dead-letter storage container URL for Identity based authentication.",
-            arg_group="Dead-letter Endpoint"
+            arg_group="Dead-letter Endpoint",
         )
         context.argument(
             "auth_type",
             options_list=["--auth-type"],
             help="Endpoint authentication type.",
-            arg_type=get_enum_type(ADTEndpointAuthType)
+            arg_type=get_enum_type(ADTEndpointAuthType),
         )
 
     with self.argument_context("dt endpoint create eventgrid") as context:
@@ -352,5 +369,48 @@ def load_digitaltwins_arguments(self, _):
             help="Indicates intent to decommission a target model.",
         )
         context.argument(
-            "dependencies_for", arg_type=depfor_type,
+            "dependencies_for",
+            arg_type=depfor_type,
+        )
+
+    with self.argument_context("dt network private-link") as context:
+        context.argument(
+            "link_name",
+            options_list=["--link-name", "--ln"],
+            help="Private link name.",
+            arg_group="Private Connection",
+        )
+
+    with self.argument_context("dt network private-endpoint") as context:
+        context.argument(
+            "conn_name",
+            options_list=["--conn-name", "--cn"],
+            help="Private endpoint connection name.",
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "group_ids",
+            options_list=["--group-ids"],
+            help="Space seperated list of group ids that the private endpoint should connect to.",
+            arg_group="Private-Endpoint",
+            nargs="+",
+        )
+        context.argument(
+            "status",
+            options_list=["--status"],
+            help="The status of a private endpoint connection.",
+            arg_type=get_enum_type(ADTPrivateConnectionStatusType),
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "description",
+            options_list=["--description", "--desc"],
+            help="Description for the private endpoint connection.",
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "actions_required",
+            options_list=["--actions-required", "--ar"],
+            help="A message indicating if changes on the service provider require any updates on the consumer.",
+            arg_group="Private-Endpoint",
         )