Updated docstring, added describe call to check config

Author: Ameen Khan

src/sagemaker/workflow/pipeline.py

@@ -105,8 +105,9 @@ def create(
             description (str): A description of the pipeline.
             tags (List[Dict[str, str]]): A list of {"Key": "string", "Value": "string"} dicts as
                 tags.
-            parallelism_config (Optional[Config for parallel steps, Parallelism configuration that
-                is applied to each of. the executions
+            parallelism_config (Optional[ParallelismConfiguration]): Parallelism configuration
+                that is applied to each of the executions of the pipeline. It takes precedence
+                over the parallelism configuration of the parent pipeline.
 
         Returns:
             A response dict from the service.
@@ -130,8 +131,9 @@ def _create_args(
         Args:
             role_arn (str): The role arn that is assumed by pipelines to create step artifacts.
             description (str): A description of the pipeline.
-            parallelism_config (Optional[ParallelismConfiguration]): Config for parallel steps, that
-                is applied to each of the executions.
+            parallelism_config (Optional[ParallelismConfiguration]): Parallelism configuration
+                that is applied to each of the executions of the pipeline. It takes precedence
+                over the parallelism configuration of the parent pipeline.
 
         Returns:
             A keyword argument dict for calling create_pipeline.
@@ -188,8 +190,9 @@ def update(
         Args:
             role_arn (str): The role arn that is assumed by pipelines to create step artifacts.
             description (str): A description of the pipeline.
-            parallelism_config (Optional[ParallelismConfiguration]): Config for parallel steps, that
-                is applied to each of the executions.
+            parallelism_config (Optional[ParallelismConfiguration]): Parallelism configuration
+                that is applied to each of the executions of the pipeline. It takes precedence
+                over the parallelism configuration of the parent pipeline.
 
         Returns:
             A response dict from the service.
@@ -265,8 +268,9 @@ def start(
                 pipeline parameters.
             execution_display_name (str): The display name of the pipeline execution.
             execution_description (str): A description of the execution.
-            parallelism_config (Optional[ParallelismConfiguration]): Config for parallel steps, that
-                is applied to each of the executions.
+            parallelism_config (Optional[ParallelismConfiguration]): Parallelism configuration
+                that is applied to each of the executions of the pipeline. It takes precedence
+                over the parallelism configuration of the parent pipeline.
 
         Returns:
             A `_PipelineExecution` instance, if successful.

tests/integ/test_workflow.py

@@ -2307,6 +2307,8 @@ def test_large_pipeline(sagemaker_session, role, pipeline_name, region_name):
             fr"arn:aws:sagemaker:{region_name}:\d{{12}}:pipeline/{pipeline_name}",
             create_arn,
         )
+        response = pipeline.describe()
+        assert len(json.loads(pipeline.describe()["PipelineDefinition"])["Steps"]) == 2000
 
         pipeline.parameters = [ParameterInteger(name="InstanceCount", default_value=1)]
         response = pipeline.update(role)
@@ -2321,7 +2323,7 @@ def test_large_pipeline(sagemaker_session, role, pipeline_name, region_name):
         except Exception:
             pass
 
-def test_create_parallelism_config(sagemaker_session, role, pipeline_name, region_name):
+def test_create_and_update_with_parallelism_config(sagemaker_session, role, pipeline_name, region_name):
     instance_count = ParameterInteger(name="InstanceCount", default_value=2)
 
     outputParam = CallbackOutput(output_name="output", output_type=CallbackOutputTypeEnum.String)
@@ -2342,21 +2344,28 @@ def test_create_parallelism_config(sagemaker_session, role, pipeline_name, regio
         sagemaker_session=sagemaker_session,
     )
 
+
     try:
         response = pipeline.create(role, parallelism_config={"MaxParallelExecutionSteps": 50})
         create_arn = response["PipelineArn"]
         assert re.match(
             fr"arn:aws:sagemaker:{region_name}:\d{{12}}:pipeline/{pipeline_name}",
             create_arn,
         )
+        response = pipeline.describe()
+        assert response["ParallelismConfiguration"]["MaxParallelExecutionSteps"] == 50
 
         pipeline.parameters = [ParameterInteger(name="InstanceCount", default_value=1)]
-        response = pipeline.update(role, parallelism_config={"MaxParallelExecutionSteps": 50})
+        response = pipeline.update(role, parallelism_config={"MaxParallelExecutionSteps": 55})
         update_arn = response["PipelineArn"]
         assert re.match(
             fr"arn:aws:sagemaker:{region_name}:\d{{12}}:pipeline/{pipeline_name}",
             update_arn,
         )
+
+        response = pipeline.describe()
+        assert response["ParallelismConfiguration"]["MaxParallelExecutionSteps"] == 55
+
     finally:
         try:
             pipeline.delete()