feat: Add 'job run-rm' command

docs/commands/jobs/index.md

@@ -161,4 +161,22 @@ fab job run-cancel <path> --id <job_id> [--wait]
 
 ---
 
+### run-rm
+
+Remove a scheduled job.
+
+**Usage:**
+
+```
+fab job run-rm <path> --id <scheduled-id> [-f]
+```
+
+**Parameters:**
+
+- `<path>`: Path to the resource.
+- `--id`: Job or schedule ID to remove.
+- `-f, --force`: Bypass confirmation prompt. Optional.
+
+---
+
 For more examples and detailed scenarios, see [Job Management Examples](../../examples/job_examples.md).

src/fabric_cli/client/fab_api_jobs.py

@@ -89,3 +89,10 @@ def list_item_runs(args: Namespace) -> ApiResponse:
     args.method = "get"
 
     return fabric_api.do_request(args)
+
+def remove_item_schedule(args: Namespace) -> ApiResponse:
+    """https://learn.microsoft.com/en-us/rest/api/fabric/core/job-scheduler/delete-item-schedule?tabs=HTTP"""
+    args.uri = f"workspaces/{args.ws_id}/items/{args.item_id}/jobs/{args.jobType}/schedules/{args.schedule_id}"
+    args.method = "delete"
+
+    return fabric_api.do_request(args)

src/fabric_cli/commands/jobs/fab_jobs.py

@@ -9,6 +9,7 @@
 from fabric_cli.commands.jobs import fab_jobs_run_sch as jobs_run_sch
 from fabric_cli.commands.jobs import fab_jobs_run_status as jobs_run_status
 from fabric_cli.commands.jobs import fab_jobs_run_update as jobs_run_update
+from fabric_cli.commands.jobs import fab_jobs_run_rm as jobs_run_rm
 from fabric_cli.core import fab_handle_context as handle_context
 from fabric_cli.core.fab_commands import Command
 from fabric_cli.core.fab_decorators import handle_exceptions, set_command_context
@@ -86,3 +87,13 @@ def run_update_command(args: Namespace) -> None:
     utils_job.add_item_props_to_args(args, context)
     utils_job.build_config_from_args(args, context, schedule=True)
     jobs_run_update.exec_command(args, context)
+
+
+@handle_exceptions()
+@set_command_context()
+def run_rm_command(args: Namespace) -> None:
+    context = handle_context.get_command_context(args.path)
+    context.check_command_support(Command.JOB_RUN_RM)
+    assert isinstance(context, Item)
+    utils_job.add_item_props_to_args(args, context)
+    jobs_run_rm.exec_command(args, context)

src/fabric_cli/commands/jobs/fab_jobs_run_rm.py

@@ -0,0 +1,38 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+from argparse import Namespace
+
+from fabric_cli.client import fab_api_jobs as jobs_api
+from fabric_cli.core import fab_constant
+from fabric_cli.core.fab_exceptions import FabricCLIError
+from fabric_cli.core.hiearchy.fab_hiearchy import Item
+from fabric_cli.utils import fab_ui as utils_ui
+
+
+def exec_command(args: Namespace, context: Item) -> None:
+    if not args.force:
+        utils_ui.print_warning(f"You are about to delete schedule '{args.schedule_id}' from '{context.full_name}' This action cannot be undone.")
+
+    if args.force or utils_ui.prompt_confirm():
+        utils_ui.print_grey(f"Removing job schedule '{args.schedule_id}'... from '{context.full_name}'")
+
+        response = jobs_api.remove_item_schedule(args)
+
+        if response.status_code == 200:
+            utils_ui.print_output_format(
+                args,
+                message=f"Job schedule {args.schedule_id} removed",
+            )
+        else:
+            try:
+                content = json.loads(response.text)
+                error_message = content.get("error", {}).get("message", "")
+            except json.JSONDecodeError:
+                error_message = response.text
+
+            raise FabricCLIError(
+                f"Failed to remove job schedule {args.schedule_id}: {error_message}",
+                fab_constant.ERROR_API_FAILURE,
+            )

src/fabric_cli/core/fab_commands.py

@@ -55,6 +55,7 @@ class Command(Enum):
     JOB_RUN_CANCEL = "job run-cancel"
     JOB_RUN_LIST = "job run-list"
     JOB_RUN_UPDATE = "job run-update"
+    JOB_RUN_RM = "job run-rm"
     JOB_RUN_SCH = "job run-sch"
     JOB_RUN_STATUS = "job run-status"
 

src/fabric_cli/core/fab_config/command_support.yaml

@@ -51,6 +51,7 @@ commands:
       run_list:
       run_update:
       run_sch:
+      run-rm:
       run_status:
       run_wait:
 

src/fabric_cli/parsers/fab_jobs_parser.py

@@ -16,6 +16,7 @@
         "run-cancel": "Cancel an item or scheduled run.",
         "run-list": "Retrieve the status of an item or scheduled job run.",
         "run-update": "Update a scheduled job.",
+        "run-rm": "Remove a scheduled job.",
         "run-sch": "Schedule a job for an item (pipelines, notebooks, and Spark job definitions).",
         "run-status": "Get details of an item or scheduled job run.",
     },
@@ -275,6 +276,43 @@ def register_parser(subparsers: _SubParsersAction) -> None:
     )
     run_update_parser.set_defaults(func=jobs.run_update_command)
 
+    # Subcommand for 'run_rm'
+    rm_examples = [
+        "# remove pipeline schedule",
+        "$ job run-rm pip1.DataPipeline --id <schedule_id>\n",
+        "# remove notebook schedule",
+        "$ job run-rm nb1.Notebook --id <schedule_id>\n",
+        "# remove Spark job definition schedule",
+        "$ job run-rm sjd1.SparkJobDefinition --id <schedule_id>\n",
+        "# remove lakehouse schedule",
+        "$ job run-rm lh1.Lakehouse --id <schedule_id>\n",
+        "# Force remove a scheduled job without confirmation prompt",
+        "$ job run-rm pip1.DataPipeline --id <schedule_id> -f\n",
+    ]
+
+    run_rm_parser = jobs_subparsers.add_parser(
+        "run-rm",
+        help="Remove a scheduled job",
+        fab_examples=rm_examples,
+        fab_learnmore=["_"],
+    )
+    run_rm_parser.add_argument("path", nargs="+", help="Path to the item")
+    run_rm_parser.add_argument(
+        "--id",
+        required=True,
+        metavar="",
+        dest="schedule_id",
+        help="Job Schedule ID",
+    )
+    run_rm_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        help="Force delete the schedule without confirmation. Optional",
+    )
+    run_rm_parser.usage = f"{utils_error_parser.get_usage_prog(run_rm_parser)}"
+    run_rm_parser.set_defaults(func=jobs.run_rm_command)
+
     # Subcommand for 'run_status'
     status_examples = [
         "# Check the status of a pipeline instance job",