Refactor params

Author: Bernd Verst

src/durabletask/azext_durabletask/_help.py

@@ -23,30 +23,17 @@
   - name: Attach a scheduler to a function app with the Worker role
     text: |
         az durabletask scheduler attach -g myResourceGroup -n myScheduler \\
-            --task-hub-name myTaskHub --target-type functionapp \\
-            --target-name myFunctionApp --role-type worker
+            --task-hub-name myTaskHub --role-type worker \\
+            --target /subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/myRG/providers/Microsoft.Web/sites/myFunctionApp
   - name: Attach a scheduler to a container app with the Data Contributor role
     text: |
         az durabletask scheduler attach -g myResourceGroup -n myScheduler \\
-            --task-hub-name myTaskHub --target-type containerapp \\
-            --target-name myContainerApp --role-type contributor
-  - name: Attach with the Data Reader role and a different resource group
-    text: |
-        az durabletask scheduler attach -g schedulerRG -n myScheduler \\
-            --task-hub-name myTaskHub --target-type functionapp \\
-            --target-name myFunctionApp --role-type reader \\
-            --target-resource-group appRG
-  - name: Attach a scheduler to a target in a different subscription
-    text: |
-        az durabletask scheduler attach -g schedulerRG -n myScheduler \\
-            --task-hub-name myTaskHub --target-type containerapp \\
-            --target-name myContainerApp --role-type contributor \\
-            --target-resource-group appRG \\
-            --target-subscription 00000000-0000-0000-0000-000000000000
+            --task-hub-name myTaskHub --role-type contributor \\
+            --target /subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/myRG/providers/Microsoft.App/containerApps/myContainerApp
   - name: Attach using a user-assigned managed identity
     text: |
         az durabletask scheduler attach -g myResourceGroup -n myScheduler \\
-            --task-hub-name myTaskHub --target-type functionapp \\
-            --target-name myFunctionApp --role-type worker \\
+            --task-hub-name myTaskHub --role-type worker \\
+            --target /subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/myRG/providers/Microsoft.Web/sites/myFunctionApp \\
             --identity /subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/myRG/providers/Microsoft.ManagedIdentity/userAssignedIdentities/myIdentity
 """

src/durabletask/azext_durabletask/_params.py

@@ -17,15 +17,8 @@ def load_arguments(self, _):  # pylint: disable=unused-argument
                    help='Name of the Durable Task scheduler.')
         c.argument('task_hub_name', options_list=['--task-hub-name'],
                    help='Name of the Durable Task task hub.')
-        c.argument('target_type', options_list=['--target-type'],
-                   help='The type of compute resource to attach.',
-                   choices=['functionapp', 'containerapp'], required=True)
-        c.argument('target_name', options_list=['--target-name'],
-                   help='Name of the target resource (Function App or Container App) to attach.')
-        c.argument('target_resource_group', options_list=['--target-resource-group'],
-                   help='Resource group of the target resource. Defaults to the scheduler resource group.')
-        c.argument('target_subscription', options_list=['--target-subscription'],
-                   help='Subscription ID of the target resource. Defaults to the current subscription.')
+        c.argument('target', options_list=['--target'],
+                   help='Resource ID of the target Function App or Container App.')
         c.argument('role_type', options_list=['--role-type'],
                    help='The type of role to assign to the target managed identity.',
                    choices=['worker', 'contributor', 'reader'], required=True)

src/durabletask/azext_durabletask/_scheduler.py

@@ -12,6 +12,33 @@
 import logging
 logger = logging.getLogger(__name__)
 
+RESOURCE_TYPE_MAP = {
+    "microsoft.web/sites": "functionapp",
+    "microsoft.app/containerapps": "containerapp",
+}
+
+
+def _parse_target(target):
+    """Parse a target resource ID and return (target_type, name, rg, subscription)."""
+    from azure.mgmt.core.tools import parse_resource_id
+
+    parsed = parse_resource_id(target)
+    name = parsed.get("name")
+    rg = parsed.get("resource_group")
+    sub = parsed.get("subscription")
+    namespace = parsed.get("namespace", "")
+    rtype = parsed.get("type", "")
+    provider_key = f"{namespace}/{rtype}".lower()
+    target_type = RESOURCE_TYPE_MAP.get(provider_key)
+    if not target_type or not name or not rg or not sub:
+        raise ValidationError(
+            f"Invalid target resource ID: '{target}'. "
+            "Expected a Function App (Microsoft.Web/sites) or "
+            "Container App (Microsoft.App/containerApps) resource ID."
+        )
+    return target_type, name, rg, sub
+
+
 ROLE_TYPE_MAP = {
     "worker": "Durable Task Worker",
     "contributor": "Durable Task Data Contributor",
@@ -270,27 +297,22 @@ def _get_containerapp_identity(cli_ctx, target_name, target_resource_group, targ
     return identity.principal_id
 
 
-def _check_target_permissions(cli_ctx, target_type, target_name, target_resource_group, target_subscription):
+def _check_target_permissions(cli_ctx, target_type, target_id):
     """Check that the caller has the required permissions on the target resource."""
-    sub = target_subscription or cli_ctx.data.get("subscription_id", "")
-
     if target_type == "functionapp":
-        scope = (f"/subscriptions/{sub}/resourceGroups/{target_resource_group}"
-                 f"/providers/Microsoft.Web/sites/{target_name}")
         required_actions = [
             "Microsoft.Web/sites/config/list/action",
             "Microsoft.Web/sites/config/write",
         ]
     else:  # containerapp
-        scope = (f"/subscriptions/{sub}/resourceGroups/{target_resource_group}"
-                 f"/providers/Microsoft.App/containerApps/{target_name}")
         required_actions = [
             "Microsoft.App/containerApps/read",
             "Microsoft.App/containerApps/write",
         ]
 
-    logger.info("Checking permissions on %s '%s'...", target_type, target_name)
-    _check_access(cli_ctx, scope, required_actions, f"{target_type} '{target_name}'")
+    logger.info("Checking permissions on target...")
+    _check_access(cli_ctx, target_id, required_actions,
+                  f"{target_type} '{target_id}'")
 
 
 def _update_functionapp_settings(cli_ctx, target_name, target_resource_group, target_subscription,
@@ -368,9 +390,7 @@ def _update_containerapp_env_vars(cli_ctx, target_name, target_resource_group, t
 
 
 def attach_scheduler(cmd, resource_group_name, scheduler_name, task_hub_name,  # pylint: disable=too-many-locals
-                     target_type, target_name, role_type,
-                     target_resource_group=None, target_subscription=None,
-                     identity=None):
+                     target, role_type, identity=None):
     """Attach a Durable Task scheduler to a Function App or Container App."""
     from azure.cli.core.commands.client_factory import get_mgmt_service_client
     from azure.cli.core.commands.arm import resolve_role_id
@@ -379,12 +399,12 @@ def attach_scheduler(cmd, resource_group_name, scheduler_name, task_hub_name,  #
     from azure.core.exceptions import ResourceExistsError, HttpResponseError
     import uuid
 
-    if target_resource_group is None:
-        target_resource_group = resource_group_name
-
     cli_ctx = cmd.cli_ctx
     client_id = None
 
+    # Parse the target resource ID
+    target_type, target_name, target_rg, target_sub = _parse_target(target)
+
     # Step 1: Get the scheduler to retrieve its endpoint
     logger.info("Retrieving scheduler '%s' in resource group '%s'...", scheduler_name, resource_group_name)
     scheduler = _SchedulerShow(cli_ctx=cli_ctx)(command_args={
@@ -429,24 +449,26 @@ def attach_scheduler(cmd, resource_group_name, scheduler_name, task_hub_name,  #
         logger.info("Ensuring identity is attached to %s '%s'...", target_type, target_name)
         if target_type == "functionapp":
             _ensure_identity_on_functionapp(
-                cli_ctx, target_name, target_resource_group,
-                target_subscription, identity)
+                cli_ctx, target_name, target_rg,
+                target_sub, identity)
         else:
             _ensure_identity_on_containerapp(
-                cli_ctx, target_name, target_resource_group,
-                target_subscription, identity)
+                cli_ctx, target_name, target_rg,
+                target_sub, identity)
     else:
         # System-assigned identity: retrieve from the target resource
         logger.info("Retrieving %s '%s' in resource group '%s'...",
-                    target_type, target_name, target_resource_group)
+                    target_type, target_name, target_rg)
         if target_type == "functionapp":
-            principal_id = _get_functionapp_identity(cli_ctx, target_name, target_resource_group, target_subscription)
+            principal_id = _get_functionapp_identity(
+                cli_ctx, target_name, target_rg, target_sub)
         else:
-            principal_id = _get_containerapp_identity(cli_ctx, target_name, target_resource_group, target_subscription)
+            principal_id = _get_containerapp_identity(
+                cli_ctx, target_name, target_rg, target_sub)
     logger.info("Managed identity principal ID: %s", principal_id)
 
     # Step 2b: Check permissions on the target and scheduler
-    _check_target_permissions(cli_ctx, target_type, target_name, target_resource_group, target_subscription)
+    _check_target_permissions(cli_ctx, target_type, target)
 
     scheduler_id = scheduler.get("id")
     logger.info("Checking permissions on scheduler '%s'...", scheduler_name)
@@ -486,21 +508,22 @@ def attach_scheduler(cmd, resource_group_name, scheduler_name, task_hub_name,  #
     logger.info("Updating settings for %s '%s'...", target_type, target_name)
     if target_type == "functionapp":
         _update_functionapp_settings(
-            cli_ctx, target_name, target_resource_group,
-            target_subscription, endpoint, task_hub_name, client_id)
+            cli_ctx, target_name, target_rg,
+            target_sub, endpoint, task_hub_name, client_id)
     else:
         _update_containerapp_env_vars(
-            cli_ctx, target_name, target_resource_group,
-            target_subscription, endpoint, task_hub_name, client_id)
+            cli_ctx, target_name, target_rg,
+            target_sub, endpoint, task_hub_name, client_id)
 
     result = {
         "scheduler": scheduler_name,
         "resourceGroup": resource_group_name,
         "taskHubName": task_hub_name,
         "schedulerEndpoint": endpoint,
+        "target": target,
         "targetType": target_type,
         "targetName": target_name,
-        "targetResourceGroup": target_resource_group,
+        "targetResourceGroup": target_rg,
         "identityPrincipalId": principal_id,
         "roleAssigned": role_name,
     }

src/durabletask/azext_durabletask/commands.py

@@ -15,7 +15,9 @@ def load_command_table(self, _):  # pylint: disable=unused-argument
     from ._format import (scheduler_table_format, scheduler_list_table_format,
                           taskhub_table_format, taskhub_list_table_format)
 
-    with self.command_group('durabletask scheduler'):
+    with self.command_group(
+            'durabletask scheduler',
+            custom_command_type=self.module_kwargs['custom_command_type']) as g:
         from .custom import CreateScheduler, UpdateScheduler
         from .aaz.latest.durabletask.scheduler import Show as _SchedulerShow, List as _SchedulerList
         self.command_table["durabletask scheduler create"] = CreateScheduler(
@@ -26,6 +28,7 @@ def load_command_table(self, _):  # pylint: disable=unused-argument
             loader=self, table_transformer=scheduler_table_format)
         self.command_table["durabletask scheduler list"] = _SchedulerList(
             loader=self, table_transformer=scheduler_list_table_format)
+        g.custom_command('attach', 'attach_scheduler')
 
     with self.command_group('durabletask taskhub'):
         from .aaz.latest.durabletask.taskhub import (
@@ -40,8 +43,3 @@ def load_command_table(self, _):  # pylint: disable=unused-argument
     with self.command_group('durabletask retention-policy'):
         from .custom import CreatePolicy
         self.command_table["durabletask retention-policy create"] = CreatePolicy(loader=self)
-
-    with self.command_group(
-            'durabletask scheduler',
-            custom_command_type=self.module_kwargs['custom_command_type']) as g:
-        g.custom_command('attach', 'attach_scheduler')