[azure-ai-ml] Simplify schedule job validation errors

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features Added
 
 ### Bugs Fixed
+- Simplified schedule validation errors so invalid local job paths report the relevant file error instead of errors from every supported job schema.
 - Fixed internal pipeline `Command` node dropping node-level interactive `services` (SSH, JupyterLab, TensorBoard, VS Code, etc.) during serialization, which prevented interactive endpoints from being created for Singularity jobs. The `services` are now serialized into the pipeline REST request and round-tripped on deserialization, matching the public `Command` node behavior.
 - Fixed `MLClient.jobs.create_or_update`, `archive`, and `restore` failing for previously-fetched jobs across all job types by routing metadata-only edits through the RunHistory PATCH endpoint.
 - Fixed `DeploymentTemplate.creation_context` always being `None` when retrieved via `get()` or `list()`. The created/modified timestamps and identity returned by the service (as `createdTime` / `modifiedTime` / `createdBy`) are now populated on `creation_context`, making `DeploymentTemplate` consistent with `Model` and `Environment`.

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/create_job.py

@@ -16,14 +16,16 @@
     FileRefField,
     NestedField,
     StringTransformedEnum,
+    TypeSensitiveUnionField,
     UnionField,
 )
 from azure.ai.ml._schema.job import BaseJobSchema
 from azure.ai.ml._schema.job.input_output_fields_provider import InputsField, OutputsField
 from azure.ai.ml._schema.pipeline.settings import PipelineJobSettingsSchema
 from azure.ai.ml._utils.utils import load_file, merge_dict
 from azure.ai.ml.constants import JobType
-from azure.ai.ml.constants._common import BASE_PATH_CONTEXT_KEY, AzureMLResourceType
+from azure.ai.ml.constants._common import ARM_ID_PREFIX, BASE_PATH_CONTEXT_KEY, AzureMLResourceType
+from azure.ai.ml.exceptions import ValidationException
 
 _SCHEDULED_JOB_UPDATES_KEY = "scheduled_job_updates"
 
@@ -56,15 +58,32 @@ def _deserialize(self, value, attr, data, **kwargs) -> "Job":
         )
 
 
+class CreateJobReferenceField(UnionField):
+    """A schedule job reference that selects remote or local validation based on its prefix."""
+
+    def __init__(self, **kwargs):
+        super().__init__(
+            [
+                ArmStr(azureml_type=AzureMLResourceType.JOB),
+                CreateJobFileRefField(),
+            ],
+            **kwargs,
+        )
+
+    def _deserialize(self, value, attr, data, **kwargs):
+        if not isinstance(value, str):
+            return super()._deserialize(value, attr, data, **kwargs)
+
+        field = self._union_fields[0] if value.startswith(ARM_ID_PREFIX) else self._union_fields[1]
+        try:
+            return field.deserialize(value, attr, data, **kwargs)
+        except ValidationException as error:
+            raise ValidationError(error.message) from error
+
+
 class BaseCreateJobSchema(BaseJobSchema):
     compute = ComputeField()
-    job = UnionField(
-        [
-            ArmStr(azureml_type=AzureMLResourceType.JOB),
-            CreateJobFileRefField,
-        ],
-        required=True,
-    )
+    job = CreateJobReferenceField(required=True)
 
     # pylint: disable-next=docstring-missing-param
     def _get_job_instance_for_remote_job(self, id: Optional[str], data: Optional[dict], **kwargs) -> "Job":
@@ -142,3 +161,30 @@ class SparkCreateJobSchema(BaseCreateJobSchema):
     type = StringTransformedEnum(allowed_values=[JobType.SPARK])
     conf = fields.Dict(keys=fields.Str(), values=fields.Raw())
     environment = EnvironmentField(allow_none=True)
+
+
+class ScheduleCreateJobField(TypeSensitiveUnionField):
+    """A schedule job definition that validates only the matching job type."""
+
+    def __init__(self, **kwargs):
+        super().__init__(
+            {
+                JobType.PIPELINE: [NestedField(PipelineCreateJobSchema)],
+                JobType.COMMAND: [NestedField(CommandCreateJobSchema)],
+                JobType.SPARK: [NestedField(SparkCreateJobSchema)],
+            },
+            plain_union_fields=[CreateJobReferenceField()],
+            allow_load_from_file=False,
+            **kwargs,
+        )
+
+    def _deserialize(self, value, attr, data, **kwargs):
+        if isinstance(value, str):
+            return self._union_fields[0].deserialize(value, attr, data, **kwargs)
+        if isinstance(value, dict) and value.get(self.type_field_name) in self.allowed_types:
+            field_index = self.allowed_types.index(value[self.type_field_name]) + 1
+            return self._union_fields[field_index].deserialize(value, attr, data, **kwargs)
+        return super()._deserialize(value, attr, data, **kwargs)
+
+    def _serialize(self, value, attr, obj, **kwargs):
+        return self._union_fields[0]._serialize(value, attr, obj, **kwargs)

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/schedule.py

@@ -4,17 +4,11 @@
 
 from marshmallow import fields
 
-from azure.ai.ml._schema.core.fields import ArmStr, NestedField, UnionField
+from azure.ai.ml._schema.core.fields import NestedField, UnionField
 from azure.ai.ml._schema.core.resource import ResourceSchema
 from azure.ai.ml._schema.job import CreationContextSchema
-from azure.ai.ml._schema.schedule.create_job import (
-    CommandCreateJobSchema,
-    CreateJobFileRefField,
-    PipelineCreateJobSchema,
-    SparkCreateJobSchema,
-)
+from azure.ai.ml._schema.schedule.create_job import ScheduleCreateJobField
 from azure.ai.ml._schema.schedule.trigger import CronTriggerSchema, RecurrenceTriggerSchema
-from azure.ai.ml.constants._common import AzureMLResourceType
 
 
 class ScheduleSchema(ResourceSchema):
@@ -33,12 +27,4 @@ class ScheduleSchema(ResourceSchema):
 
 
 class JobScheduleSchema(ScheduleSchema):
-    create_job = UnionField(
-        [
-            ArmStr(azureml_type=AzureMLResourceType.JOB),
-            CreateJobFileRefField,
-            NestedField(PipelineCreateJobSchema),
-            NestedField(CommandCreateJobSchema),
-            NestedField(SparkCreateJobSchema),
-        ]
-    )
+    create_job = ScheduleCreateJobField()

sdk/ml/azure-ai-ml/tests/schedule/unittests/test_schedule_schema.py

@@ -1,5 +1,6 @@
 import pydash
 import pytest
+import yaml
 from marshmallow import ValidationError
 
 from azure.ai.ml._utils.utils import load_yaml
@@ -12,6 +13,41 @@
 @pytest.mark.unittest
 @pytest.mark.pipeline_test
 class TestScheduleSchema:
+    @pytest.mark.parametrize(
+        "create_job",
+        [
+            "./missing-job.yml",
+            {"type": "pipeline", "job": "./missing-job.yml"},
+            {"type": "command", "job": "./missing-job.yml"},
+            {"type": "spark", "job": "./missing-job.yml"},
+        ],
+    )
+    def test_missing_local_job_has_concise_error(self, tmp_path, create_job):
+        schedule_data = load_yaml("./tests/test_configs/schedule/hello_cron_schedule_with_file_reference.yml")
+        schedule_data["create_job"] = create_job
+        schedule_path = tmp_path / "schedule.yml"
+        schedule_path.write_text(yaml.safe_dump(schedule_data), encoding="utf-8")
+
+        with pytest.raises(ValidationError) as error:
+            load_schedule(schedule_path)
+
+        error_message = str(error.value)
+        assert "No such file or directory" in error_message
+        assert "In order to specify an existing jobs" not in error_message
+        assert "Not supporting non file for create_job" not in error_message
+        assert "passed is not in set" not in error_message
+
+    def test_unsupported_job_type_has_concise_error(self, tmp_path):
+        schedule_data = load_yaml("./tests/test_configs/schedule/hello_cron_schedule_with_file_reference.yml")
+        schedule_data["create_job"] = {"type": "unsupported", "job": "./missing-job.yml"}
+        schedule_path = tmp_path / "schedule.yml"
+        schedule_path.write_text(yaml.safe_dump(schedule_data), encoding="utf-8")
+
+        with pytest.raises(ValidationError) as error:
+            load_schedule(schedule_path)
+
+        assert "Value 'unsupported' passed is not in set" in str(error.value)
+
     def test_load_cron_schedule_with_file_reference(self):
         test_path = "./tests/test_configs/schedule/hello_cron_schedule_with_file_reference.yml"
         schedule = load_schedule(test_path)