Added the CLI command for migrating DLT pipelines (#3579)

## Changes
Added cli command for migration DLT pipelines with the flags for
limiting the pipelines ids

### Linked issues
Introduces #3107 

### Functionality
- [x] added new CLI command

### Tests
- [x] manually tested
- [ ] added unit tests
- [ ] added integration tests
- [ ] verified on staging environment (screenshot attached)

Author: pritishpai

labs.yml

@@ -368,3 +368,11 @@ commands:
 
   - name: assign-owner-group
     description: Assign owner group to the workspace. This group will be assigned as an owner to all migrated tables and views.
+
+  - name: migrate-dlt-pipelines
+    description: Migrate DLT pipelines from HMS to UC using the DLT Migration API
+    flags:
+      - name: include-pipeline-ids
+        description: (Optional) Comma separated list of pipeline IDs to include in the migration
+      - name: exclude-pipeline-ids
+        description: (Optional) Comma separated list of pipeline IDs to exclude from the migration

src/databricks/labs/ucx/cli.py

@@ -911,12 +911,18 @@ def migrate_dlt_pipelines(
     ctx: WorkspaceContext | None = None,
     run_as_collection: bool = False,
     a: AccountClient | None = None,
+    **named_parameters,
 ) -> None:
-    """Migrate DLT pipelines to UC"""
+    """
+    Migrate DLT pipelines to UC that were crawled in the assessment workflow
+    Flags
+    -- include-pipeline-ids: Comma separated list of pipeline ids to migrate
+    -- exclude-pipeline-ids: Comma separated list of pipeline ids to exclude from migration
+    """
     if ctx:
         workspace_contexts = [ctx]
     else:
-        workspace_contexts = _get_workspace_contexts(w, a, run_as_collection)
+        workspace_contexts = _get_workspace_contexts(w, a, run_as_collection, **named_parameters)
 
     for workspace_context in workspace_contexts:
         workspace_context.pipelines_migrator.migrate_pipelines()

src/databricks/labs/ucx/contexts/application.py

@@ -374,8 +374,23 @@ def table_ownership_grant_loader(self) -> TableOwnershipGrantLoader:
 
     @cached_property
     def pipelines_migrator(self) -> PipelinesMigrator:
+        include_pipeline_ids = (
+            self.named_parameters.get('include_pipeline_ids', '').split(',')
+            if 'include_pipeline_ids' in self.named_parameters
+            else None
+        )
+        exclude_pipeline_ids = (
+            self.named_parameters.get('exclude_pipeline_ids', '').split(',')
+            if 'exclude_pipeline_ids' in self.named_parameters
+            else None
+        )
         return PipelinesMigrator(
-            self.workspace_client, self.pipelines_crawler, self.jobs_crawler, self.config.ucx_catalog
+            self.workspace_client,
+            self.pipelines_crawler,
+            self.jobs_crawler,
+            self.config.ucx_catalog,
+            include_pipeline_ids=include_pipeline_ids,
+            exclude_pipeline_ids=exclude_pipeline_ids,
         )
 
     @cached_property