diff --git a/sdk/ml/azure-ai-ml/CHANGELOG.md b/sdk/ml/azure-ai-ml/CHANGELOG.md index 73d658a90ffb..016968c92465 100644 --- a/sdk/ml/azure-ai-ml/CHANGELOG.md +++ b/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`. diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/create_job.py b/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/create_job.py index 084f8a5b74f2..87f3151e5b81 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/create_job.py +++ b/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/create_job.py @@ -16,6 +16,7 @@ FileRefField, NestedField, StringTransformedEnum, + TypeSensitiveUnionField, UnionField, ) from azure.ai.ml._schema.job import BaseJobSchema @@ -23,7 +24,8 @@ 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) diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/schedule.py b/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/schedule.py index fbde3e9b2cea..dfee0e856f79 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/_schema/schedule/schedule.py +++ b/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() diff --git a/sdk/ml/azure-ai-ml/tests/schedule/unittests/test_schedule_schema.py b/sdk/ml/azure-ai-ml/tests/schedule/unittests/test_schedule_schema.py index 2bf677c8503a..81e9594a9675 100644 --- a/sdk/ml/azure-ai-ml/tests/schedule/unittests/test_schedule_schema.py +++ b/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)