Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changes/unreleased/added-20261026-191106.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
kind: added
body: Add --wait (-w), --timeout, and --polling_interval to job start to wait for completion like job run.
time: 2026-10-26T19:11:06.155500618Z
custom:
Author: tarostok_microsoft
AuthorLink: https://github.com/tarostok
13 changes: 10 additions & 3 deletions docs/commands/jobs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ The `job` commands provide tools for starting, running, monitoring, and scheduli

| Command | Description | Usage |
|-----------------|---------------------------|-----------------------------------------------------------------------|
| `job start` | Start an item (async) | `job start <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>]` |
| `job start` | Start an item (async by default) | `job start <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>] [--wait [--timeout <seconds>] [--polling_interval <seconds>]]` |
| `job run` | Run an item (sync) | `job run <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>] [--timeout <seconds>]` |
| `job run-cancel`| Cancel an item run | `job run-cancel <path> --id <id> [--wait]` |
| `job run-list` | List item or scheduled job runs | `job run-list <path> [--schedule]` |
Expand All @@ -27,11 +27,13 @@ The `job` commands provide tools for starting, running, monitoring, and scheduli
### start

Start an item asynchronously and return immediately with a job ID.
Use `--wait` (or `-w`) to wait for completion with the same completion and
failure handling as `job run`.

**Usage:**

```
fab job start <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>]
fab job start <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>] [--wait [--timeout <seconds>] [--polling_interval <seconds>]]
```

**Parameters:**
Expand All @@ -40,12 +42,17 @@ fab job start <path> [-P <params>] [-C <config>] [-i <json_inline_or_path>]
- `-P, --params`: Parameters in name:type=value format, comma-separated. Optional.
- `-C, --config`: JSON configuration for notebooks. Optional.
- `-i, --input`: JSON payload for additional configuration. Optional.
- `-w, --wait`: Wait for the job to complete. Optional.
- `--timeout`: Maximum time to wait for completion in seconds (0 or greater). Requires `--wait`. Optional.
- `--polling_interval`: Custom job status polling interval in seconds. Requires `--wait`. Optional.

**Examples:**
```
fab job start pipeline1.datapipeline
fab job start notebook1.notebook -P param1:string=value1,param2:int=42
fab job start notebook1.notebook -C '{"spark.executor.memory": "4g"}'
fab job start ws1.Workspace/pipeline1.DataPipeline --wait
fab job start ws1.Workspace/notebook1.Notebook --wait --timeout 600
```

---
Expand All @@ -66,7 +73,7 @@ fab job run <path> [-P <params>] [-C <config>] [-i <json_path>] [--timeout <seco
- `-P, --params`: Parameters in name:type=value format, comma-separated. Optional.
- `-C, --config`: JSON configuration for notebooks. Optional.
- `-i, --input`: JSON payload for additional configuration. Optional.
- `--timeout`: Maximum time to wait for completion in seconds. Optional.
- `--timeout`: Maximum time to wait for completion in seconds (0 or greater). Optional.
- `--polling_interval`: Custom job status polling interval in seconds. Optional.

**Examples:**
Expand Down
8 changes: 8 additions & 0 deletions docs/examples/job_examples.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,14 @@ Execute a job synchronously, waiting for it to complete before returning control
fab job run ws1.Workspace/pip1.DataPipeline
```

Alternatively, use `job start --wait` (or `-w`) to wait for completion with the
same completion and failure handling as `job run`. `--timeout` and
`--polling_interval` are also supported with `--wait`.

```
fab job start ws1.Workspace/pip1.DataPipeline --wait --timeout 120
```

#### Run Job with Timeout

Execute a job with a specific timeout limit in seconds.
Expand Down
4 changes: 2 additions & 2 deletions src/fabric_cli/commands/jobs/fab_jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@ def run_command(args: Namespace) -> None:
utils_job.add_item_props_to_args(args, context)
utils_job.build_config_from_args(args, context)
if args.jobs_command == "start":
fab_ui.print_grey(f"Starting job (async) for '{args.item}'...")
args.wait = False
mode = "sync" if args.wait else "async"
fab_ui.print_grey(f"Starting job ({mode}) for '{args.item}'...")
if args.jobs_command == "run":
fab_ui.print_grey(f"Running job (sync) for '{args.item}'...")
args.wait = True
Expand Down
8 changes: 8 additions & 0 deletions src/fabric_cli/errors/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,14 @@ def universal_security_disabled(name: str) -> str:
def job_instance_failed(job_id: str) -> str:
return f"Job instance '{job_id}' Failed"

@staticmethod
def options_require_wait(options: list[str]) -> str:
return f"{', '.join(options)} can only be used with --wait"

@staticmethod
def negative_job_timeout(timeout: int) -> str:
return f"Timeout ({timeout}s) must be greater than or equal to 0"

@staticmethod
def file_not_accessible(file_path: str) -> str:
return f"Cannot access '{file_path}'. No such file or directory"
Expand Down
23 changes: 21 additions & 2 deletions src/fabric_cli/parsers/fab_jobs_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@

commands = {
"Commands": {
"start": "Start an item asynchronously.",
"start": "Start an item asynchronously, or wait for completion with --wait.",
"run": "Run an item synchronously.",
"run-cancel": "Cancel a running or scheduled job.",
"run-list": "Retrieve the status of an item or scheduled job run.",
Expand All @@ -35,7 +35,11 @@ def register_parser(subparsers: _SubParsersAction) -> None:
"# run a pipeline async",
"$ job start pip1.DataPipeline\n",
"# run a notebook async",
"$ job start nb1.Notebook --input <json_path>",
"$ job start nb1.Notebook --input <json_path>\n",
"# start a pipeline and wait for completion",
"$ job start pip1.DataPipeline --wait\n",
"# start a pipeline and wait with a 60 second timeout",
"$ job start pip1.DataPipeline --wait --timeout 60",
]

start_parser = jobs_subparsers.add_parser(
Expand All @@ -62,6 +66,21 @@ def register_parser(subparsers: _SubParsersAction) -> None:
start_parser.add_argument(
"-i", "--input", nargs="+", help="JSON payload, inline or path. Optional"
)
start_parser.add_argument(
"-w", "--wait", action="store_true", help="Wait for the job to complete. Optional"
)
start_parser.add_argument(
"--timeout",
metavar="",
help="Timeout in seconds (0 or greater). Requires --wait. Optional",
type=int,
Comment thread
tarostok marked this conversation as resolved.
Resolved
)
start_parser.add_argument(
"--polling_interval",
metavar="",
help="Custom job status polling interval in seconds. Requires --wait. Optional",
type=validate_positive_int,
)

start_parser.usage = f"{utils_error_parser.get_usage_prog(start_parser)}"
start_parser.set_defaults(func=lazy_command(_jobs_module_path, 'run_command'))
Expand Down
29 changes: 28 additions & 1 deletion src/fabric_cli/utils/fab_cmd_job_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -424,10 +424,37 @@ def _process_params(args: Namespace, item: Item) -> None:


def validate_timeout_polling_interval(args: Namespace) -> None:
"""Validate that polling interval is not greater than or equal to timeout."""
"""Validate timeout and polling interval options for job execution.

Wait options require --wait for job start, and the polling
interval must be lower than the timeout. Timeouts must be non-negative.
"""
timeout = getattr(args, "timeout", None)
polling_interval = getattr(args, "polling_interval", None)

if getattr(args, "jobs_command", None) == "start" and not getattr(
args, "wait", False
):
options = [
option
for option, value in (
("--timeout", timeout),
("--polling_interval", polling_interval),
)
if value is not None
]
if options:
raise FabricCLIError(
ErrorMessages.Common.options_require_wait(options),
fab_constant.ERROR_INVALID_INPUT,
)

if timeout is not None and timeout < 0:
raise FabricCLIError(
ErrorMessages.Common.negative_job_timeout(timeout),
fab_constant.ERROR_INVALID_INPUT,
Comment thread
tarostok marked this conversation as resolved.
Resolved
)

if (
timeout is not None
and polling_interval is not None
Expand Down
201 changes: 201 additions & 0 deletions tests/test_core/test_fab_jobs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,201 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

"""Unit tests for job start waiting and job run compatibility."""

import inspect
import json
from argparse import Namespace
from unittest.mock import Mock

import pytest

from fabric_cli.commands.jobs import fab_jobs
from fabric_cli.core import fab_constant
from fabric_cli.core.fab_exceptions import FabricCLIError
from fabric_cli.core.fab_parser_setup import CustomArgumentParser
from fabric_cli.core.hiearchy.fab_item import Item
from fabric_cli.parsers import fab_jobs_parser


def _parse_job_args(command: str, flags: list[str]) -> Namespace:
parser = CustomArgumentParser()
fab_jobs_parser.register_parser(parser.add_subparsers(dest="command"))
args = parser.parse_args(["job", command, "ws.Workspace/nb.Notebook", *flags])
args.item = "nb.Notebook"
args.output_format = "json"
args.command_path = f"job {command}"
return args


@pytest.fixture
def job_mocks(monkeypatch: pytest.MonkeyPatch) -> Namespace:
"""Isolate context resolution, HTTP requests, output, and polling delays."""
item = Mock(spec=Item)
monkeypatch.setattr(
fab_jobs.handle_context, "get_command_context", Mock(return_value=item)
)
monkeypatch.setattr(fab_jobs.utils_job, "add_item_props_to_args", Mock())
monkeypatch.setattr(fab_jobs.utils_job, "build_config_from_args", Mock())

response = Mock(status_code=202, headers={})
start = Mock(return_value=(response, "test-job-id"))
poll = Mock(
side_effect=[
Mock(status_code=200, headers={}, text=json.dumps({"status": status}))
for status in ("InProgress", "Completed")
]
)
sleep = Mock()
output = Mock()
monkeypatch.setattr(fab_jobs.jobs_run.jobs_api, "run_on_demand_item_job", start)
monkeypatch.setattr(fab_jobs.utils_job.jobs_api, "get_item_job_instance", poll)
monkeypatch.setattr(fab_jobs.utils_job.time, "sleep", sleep)
monkeypatch.setattr(fab_jobs.fab_ui, "print_grey", Mock())
monkeypatch.setattr(fab_jobs.fab_ui, "print_progress", Mock())
monkeypatch.setattr(fab_jobs.fab_ui, "print_output_format", output)
monkeypatch.setattr(fab_jobs.fab_ui, "print_entries_unix_style", Mock())
return Namespace(start=start, poll=poll, sleep=sleep, output=output)


@pytest.mark.parametrize(
("command", "flags", "wait"),
[
("start", [], False),
("start", ["--wait"], True),
("start", ["-w"], True),
("run", [], True),
],
)
def test_job_wait_behavior(
command: str, flags: list[str], wait: bool, job_mocks: Namespace
) -> None:
args = _parse_job_args(command, flags)

inspect.unwrap(fab_jobs.run_command)(args)

assert args.wait is wait
job_mocks.start.assert_called_once_with(args, None)
assert job_mocks.poll.call_count == (2 if wait else 0)
assert job_mocks.sleep.call_count == (2 if wait else 0)
job_mocks.output.assert_called_once()
assert job_mocks.output.call_args.kwargs["data"] == {"id": "test-job-id"}
assert job_mocks.output.call_args.kwargs["message"] == (
f"Job instance 'test-job-id' {'completed' if wait else 'created'}"
)


@pytest.mark.parametrize("command", ["start", "run"])
def test_job_wait_reports_failure(command: str, job_mocks: Namespace) -> None:
args = _parse_job_args(command, ["--wait"] if command == "start" else [])
job_mocks.poll.side_effect = None
job_mocks.poll.return_value = Mock(
status_code=200, headers={}, text=json.dumps({"status": "Failed"})
)

with pytest.raises(FabricCLIError) as exc_info:
inspect.unwrap(fab_jobs.run_command)(args)

assert exc_info.value.status_code == fab_constant.ERROR_JOB_FAILED
job_mocks.output.assert_not_called()


@pytest.mark.parametrize(
("flags", "expected"),
[
(["--timeout", "60"], "--timeout can only be used with --wait"),
(
["--polling_interval", "10"],
"--polling_interval can only be used with --wait",
),
(
["--timeout", "60", "--polling_interval", "10"],
"--timeout, --polling_interval can only be used with --wait",
),
],
)
def test_job_start_wait_options_without_wait_failure(
flags: list[str], expected: str, job_mocks: Namespace
) -> None:
args = _parse_job_args("start", flags)

with pytest.raises(FabricCLIError) as exc_info:
inspect.unwrap(fab_jobs.run_command)(args)

assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT
assert exc_info.value.message == expected
job_mocks.start.assert_not_called()


@pytest.mark.parametrize("command", ["start", "run"])
def test_job_negative_timeout_failure(command: str, job_mocks: Namespace) -> None:
flags = ["--wait"] if command == "start" else []
args = _parse_job_args(command, [*flags, "--timeout", "-1"])

with pytest.raises(FabricCLIError) as exc_info:
inspect.unwrap(fab_jobs.run_command)(args)

assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT
job_mocks.start.assert_not_called()
job_mocks.poll.assert_not_called()
job_mocks.sleep.assert_not_called()


@pytest.mark.parametrize("command", ["start", "run"])
def test_job_zero_timeout_success(
command: str, job_mocks: Namespace, monkeypatch: pytest.MonkeyPatch
) -> None:
flags = ["--wait"] if command == "start" else []
args = _parse_job_args(command, [*flags, "--timeout", "0"])
monkeypatch.setattr(
fab_jobs.jobs_run.config, "get_config", Mock(return_value="false")
)
warning = Mock()
monkeypatch.setattr(fab_jobs.fab_ui, "print_warning", warning)

inspect.unwrap(fab_jobs.run_command)(args)

job_mocks.start.assert_called_once_with(args, None)
job_mocks.sleep.assert_called_once_with(0)
job_mocks.poll.assert_not_called()
warning.assert_called_once_with(
"Job instance 'test-job-id' timed out after 0 seconds"
)


def test_job_start_wait_timeout_success(
job_mocks: Namespace, monkeypatch: pytest.MonkeyPatch
) -> None:
args = _parse_job_args(
"start", ["--wait", "--timeout", "60", "--polling_interval", "10"]
)
wait = Mock()
monkeypatch.setattr(fab_jobs.jobs_run.utils_job, "wait_for_job_completion", wait)

inspect.unwrap(fab_jobs.run_command)(args)

wait.assert_called_once()
assert wait.call_args.kwargs["timeout"] == 60
assert wait.call_args.kwargs["custom_polling_interval"] == 10


def test_job_start_wait_timeout_cancels_job(
job_mocks: Namespace,
monkeypatch: pytest.MonkeyPatch,
mock_fab_set_state_config,
) -> None:
args = _parse_job_args("start", ["--wait", "--timeout", "5"])
mock_fab_set_state_config(fab_constant.FAB_JOB_CANCEL_ONTIMEOUT, "true")
monkeypatch.setattr(
fab_jobs.jobs_run.utils_job,
"wait_for_job_completion",
Mock(side_effect=TimeoutError("timed out")),
)
monkeypatch.setattr(fab_jobs.fab_ui, "print_warning", Mock())
cancel = Mock(return_value=Mock(status_code=202))
monkeypatch.setattr(fab_jobs.jobs_run.jobs_api, "cancel_item_job_instance", cancel)

inspect.unwrap(fab_jobs.run_command)(args)

cancel.assert_called_once_with(args)
assert args.instance_id == "test-job-id"
Loading
Loading