fix: add warning message for job-prefixed pipeline steps when no job name is provided

Author: svia3

src/sagemaker/workflow/utilities.py

@@ -41,6 +41,13 @@
     "if desired."
 )
 
+JOB_KEY_NONE_WARN_MSG_TEMPLATE = (
+    "The provided job field [{}] has not been specified. Please refer to the AWS Docs to "
+    "identify which field should be set to enable the custom-prefixing feature for jobs "
+    "created via a pipeline execution. "
+    "https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-access.html#build-and-manage-step-permissions-prefix"
+)
+
 if TYPE_CHECKING:
     from sagemaker.workflow.step_collections import StepCollection
 
@@ -458,6 +465,8 @@ def trim_request_dict(request_dict, job_key, config):
         request_dict.pop(job_key, None)  # safely return null in case of KeyError
     else:
         if job_key in request_dict:
+            if request_dict[job_key] is None:
+                raise ValueError(JOB_KEY_NONE_WARN_MSG_TEMPLATE.format(job_key))
             request_dict[job_key] = base_from_name(request_dict[job_key])  # trim timestamp
 
     return request_dict

tests/unit/sagemaker/workflow/test_model_step.py

@@ -1171,3 +1171,29 @@ def test_create_model_step_using_custom_model_name(pipeline_session):
     steps = json.loads(pipeline.definition())["Steps"]
     assert len(steps) == 1
     assert "ModelName" not in steps[0]["Arguments"]
+
+
+def test_create_model_step_using_custom_model_name_set_to_none(pipeline_session):
+
+    # Name of the model not specified, will resolve to None.
+    model = Model(
+        image_uri="my-image",
+        sagemaker_session=pipeline_session,
+        model_data="s3://",
+        role=ROLE,
+    )
+    step_create_model = ModelStep(name="MyModelStep", step_args=model.create())
+
+    # 1. Toggle on custom-prefixing model package name popped
+    config = PipelineDefinitionConfig(use_custom_job_prefix=True)
+
+    with pytest.raises(ValueError) as error:
+        pipeline = Pipeline(
+            name="MyPipeline",
+            steps=[step_create_model],
+            sagemaker_session=pipeline_session,
+            pipeline_definition_config=config,
+        )
+        pipeline.definition()
+
+    assert "The provided job field [ModelName] has not been specified" in str(error.value)