Merge branch 'main' into master

Author: ca-nguyen

CONTRIBUTING.md

@@ -14,11 +14,15 @@ information to effectively respond to your bug report or contribution.
 * [Contributing via Pull Requests (PRs)](#contributing-via-pull-requests-prs)
   * [Pulling Down the Code](#pulling-down-the-code)
   * [Running the Unit Tests](#running-the-unit-tests)
+    * [Running Unit Tests and Debugging in PyCharm](#running-unit-tests-and-debugging-in-pycharm)
   * [Running the Integration Tests](#running-the-integration-tests)
   * [Making and Testing Your Change](#making-and-testing-your-change)
   * [Committing Your Change](#committing-your-change)
   * [Sending a Pull Request](#sending-a-pull-request)
 * [Finding Contributions to Work On](#finding-contributions-to-work-on)
+* [Setting Up Your Development Environment](#setting-up-your-development-environment)
+  * [Setting Up Your Environment for Debugging](#setting-up-your-environment-for-debugging)
+    * [PyCharm](#pycharm)
 * [Code of Conduct](#code-of-conduct)
 * [Security Issue Notifications](#security-issue-notifications)
 * [Licensing](#licensing)
@@ -65,6 +69,11 @@ You can also run a single test with the following command: `tox -e py36 -- -s -v
   * Note that the coverage test will fail if you only run a single test, so make sure to surround the command with `export IGNORE_COVERAGE=-` and `unset IGNORE_COVERAGE`
   * Example: `export IGNORE_COVERAGE=- ; tox -e py36 -- -s -vv tests/unit/test_sagemaker_steps.py::test_training_step_creation_with_model ; unset IGNORE_COVERAGE`
 
+#### Running Unit Tests and Debugging in PyCharm
+You can also run the unit tests with the following options:
+* Right click on a test file in the Project tree and select `Run/Debug 'pytest' for ...`
+* Right click on the test definition and select `Run/Debug 'pytest' for ...`
+* Click on the green arrow next to test definition
 
 ### Running the Integration Tests
 
@@ -168,6 +177,33 @@ Please remember to:
 Looking at the existing issues is a great way to find something to contribute on. As our projects, by default, use the default GitHub issue labels ((enhancement/bug/duplicate/help wanted/invalid/question/wontfix), looking at any ['help wanted'](https://github.com/aws/aws-step-functions-data-science-sdk-python/labels/help%20wanted) issues is a great place to start.
 
 
+## Setting Up Your Development Environment
+
+### Setting Up Your Environment for Debugging
+
+Setting up your IDE for debugging tests locally will save you a lot of time.
+You might be able to `Run` and `Debug` the tests directly in your IDE with your default settings, but if it's not the case,
+follow the steps described in this section.
+
+#### PyCharm
+1. Set your Default test runner to `pytest` in _Preferences → Tools → Python Integrated Tools_
+1. If you are using `PyCharm Professional Edition`, go to _Preferences → Build, Execution, Deployment → Python Debugger_ and set the options with following values:
+   
+   | Option                                                      | Value                 |
+   |:------------------------------------------------------------ |:----------------------|
+   | Attach subprocess automatically while debugging             | `Enabled`             |
+   | Collect run-time types information for code insight         | `Enabled`             |
+   | Gevent compatible                                           | `Disabled`            |
+   | Drop into debugger on failed tests                          | `Enabled`             |
+   | PyQt compatible                                             | `Auto`                |
+   | For Attach to Process show processes with names containing  | `python`              |
+   
+    This will allow you to break into all subprocesses of the process being debugged and preserve functions types while debugging.
+1. Debug tests in PyCharm as per [Running Unit Tests and Debugging in PyCharm](#running-unit-tests-and-debugging-in-pycharm)
+
+_Note: This setup was tested and confirmed to work with
+`PyCharm 2020.3.5 (Professional Edition)` and `PyCharm 2021.1.1 (Professional Edition)`_
+
 ## Code of Conduct
 
 This project has adopted the [Amazon Open Source Code of Conduct](https://aws.github.io/code-of-conduct).

doc/services.rst

@@ -8,6 +8,8 @@ This module provides classes to build steps that integrate with Amazon DynamoDB,
 
 - `Amazon DynamoDB <#amazon-dynamodb>`__
 
+- `Amazon EKS <#amazon-eks>`__
+
 - `Amazon EMR <#amazon-emr>`__
 
 - `Amazon EventBridge <#amazon-eventbridge>`__
@@ -29,6 +31,26 @@ Amazon DynamoDB
 
 .. autoclass:: stepfunctions.steps.service.DynamoDBUpdateItemStep
 
+
+Amazon EKS
+----------
+.. autoclass:: stepfunctions.steps.service.EksCallStep
+
+.. autoclass:: stepfunctions.steps.service.EksCreateClusterStep
+
+.. autoclass:: stepfunctions.steps.service.EksCreateFargateProfileStep
+
+.. autoclass:: stepfunctions.steps.service.EksCreateNodeGroupStep
+
+.. autoclass:: stepfunctions.steps.service.EksDeleteClusterStep
+
+.. autoclass:: stepfunctions.steps.service.EksDeleteFargateProfileStep
+
+.. autoclass:: stepfunctions.steps.service.EksDeleteNodegroupStep
+
+.. autoclass:: stepfunctions.steps.service.EksRunJobStep
+
+
 Amazon EMR
 -----------
 .. autoclass:: stepfunctions.steps.service.EmrCreateClusterStep

src/stepfunctions/steps/__init__.py

@@ -19,6 +19,17 @@
 from stepfunctions.steps.sagemaker import TrainingStep, TransformStep, ModelStep, EndpointConfigStep, EndpointStep, TuningStep, ProcessingStep
 from stepfunctions.steps.compute import LambdaStep, BatchSubmitJobStep, GlueStartJobRunStep, EcsRunTaskStep
 from stepfunctions.steps.service import DynamoDBGetItemStep, DynamoDBPutItemStep, DynamoDBUpdateItemStep, DynamoDBDeleteItemStep
+
+from stepfunctions.steps.service import (
+    EksCallStep,
+    EksCreateClusterStep,
+    EksCreateFargateProfileStep,
+    EksCreateNodeGroupStep,
+    EksDeleteClusterStep,
+    EksDeleteFargateProfileStep,
+    EksDeleteNodegroupStep,
+    EksRunJobStep,
+)
 from stepfunctions.steps.service import EmrCreateClusterStep, EmrTerminateClusterStep, EmrAddStepStep, EmrCancelStepStep, EmrSetClusterTerminationProtectionStep, EmrModifyInstanceFleetByNameStep, EmrModifyInstanceGroupByNameStep
 from stepfunctions.steps.service import EventBridgePutEventsStep
 from stepfunctions.steps.service import GlueDataBrewStartJobRunStep

src/stepfunctions/steps/fields.py

@@ -60,7 +60,6 @@ class Field(Enum):
     HeartbeatSeconds     = 'heartbeat_seconds'
     HeartbeatSecondsPath = 'heartbeat_seconds_path'
 
-
     # Retry and catch fields
     ErrorEquals      = 'error_equals'
     IntervalSeconds  = 'interval_seconds'

src/stepfunctions/steps/sagemaker.py

@@ -19,7 +19,7 @@
 from stepfunctions.inputs import Placeholder
 from stepfunctions.steps.states import Task
 from stepfunctions.steps.fields import Field
-from stepfunctions.steps.utils import tags_dict_to_kv_list
+from stepfunctions.steps.utils import merge_dicts, tags_dict_to_kv_list
 from stepfunctions.steps.integration_resources import IntegrationPattern, get_service_integration_arn
 
 from sagemaker.workflow.airflow import training_config, transform_config, model_config, tuning_config, processing_config
@@ -30,6 +30,7 @@
 
 SAGEMAKER_SERVICE_NAME = "sagemaker"
 
+
 class SageMakerApi(Enum):
     CreateTrainingJob = "createTrainingJob"
     CreateTransformJob = "createTransformJob"
@@ -479,7 +480,9 @@ class ProcessingStep(Task):
     Creates a Task State to execute a SageMaker Processing Job.
     """
 
-    def __init__(self, state_id, processor, job_name, inputs=None, outputs=None, experiment_config=None, container_arguments=None, container_entrypoint=None, kms_key_id=None, wait_for_completion=True, tags=None, **kwargs):
+    def __init__(self, state_id, processor, job_name, inputs=None, outputs=None, experiment_config=None,
+                 container_arguments=None, container_entrypoint=None, kms_key_id=None, wait_for_completion=True,
+                 tags=None, **kwargs):
         """
         Args:
             state_id (str): State name whose length **must be** less than or equal to 128 unicode characters. State names **must be** unique within the scope of the whole state machine.
@@ -491,15 +494,18 @@ def __init__(self, state_id, processor, job_name, inputs=None, outputs=None, exp
             outputs (list[:class:`~sagemaker.processing.ProcessingOutput`]): Outputs for
                 the processing job. These can be specified as either path strings or
                 :class:`~sagemaker.processing.ProcessingOutput` objects (default: None).
-            experiment_config (dict, optional): Specify the experiment config for the processing. (Default: None)
-            container_arguments ([str]): The arguments for a container used to run a processing job.
-            container_entrypoint ([str]): The entrypoint for a container used to run a processing job.
-            kms_key_id (str): The AWS Key Management Service (AWS KMS) key that Amazon SageMaker
+            experiment_config (dict or Placeholder, optional): Specify the experiment config for the processing. (Default: None)
+            container_arguments ([str] or Placeholder): The arguments for a container used to run a processing job.
+            container_entrypoint ([str] or Placeholder): The entrypoint for a container used to run a processing job.
+            kms_key_id (str or Placeholder): The AWS Key Management Service (AWS KMS) key that Amazon SageMaker
                 uses to encrypt the processing job output. KmsKeyId can be an ID of a KMS key,
                 ARN of a KMS key, alias of a KMS key, or alias of a KMS key.
                 The KmsKeyId is applied to all outputs.
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait for the processing job to complete before proceeding to the next step in the workflow. Set to `False` if the Task state should submit the processing job and proceed to the next step. (default: True)
-            tags (list[dict], optional): `List to tags <https://docs.aws.amazon.com/sagemaker/latest/dg/API_Tag.html>`_ to associate with the resource.
+            tags (list[dict] or Placeholder, optional): `List to tags <https://docs.aws.amazon.com/sagemaker/latest/dg/API_Tag.html>`_ to associate with the resource.
+            parameters(dict, optional): The value of this field is merged with other arguments to become the request payload for SageMaker `CreateProcessingJob<https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_CreateProcessingJob.html>`_. 
+                You can use `parameters` to override the value provided by other arguments and specify any field's value dynamically using `Placeholders<https://aws-step-functions-data-science-sdk.readthedocs.io/en/stable/placeholders.html?highlight=placeholder#stepfunctions.inputs.Placeholder>`_.
+
         """
         if wait_for_completion:
             """
@@ -518,22 +524,25 @@ def __init__(self, state_id, processor, job_name, inputs=None, outputs=None, exp
                                                                        SageMakerApi.CreateProcessingJob)
 
         if isinstance(job_name, str):
-            parameters = processing_config(processor=processor, inputs=inputs, outputs=outputs, container_arguments=container_arguments, container_entrypoint=container_entrypoint, kms_key_id=kms_key_id, job_name=job_name)
+            processing_parameters = processing_config(processor=processor, inputs=inputs, outputs=outputs, container_arguments=container_arguments, container_entrypoint=container_entrypoint, kms_key_id=kms_key_id, job_name=job_name)
         else:
-            parameters = processing_config(processor=processor, inputs=inputs, outputs=outputs, container_arguments=container_arguments, container_entrypoint=container_entrypoint, kms_key_id=kms_key_id)
+            processing_parameters = processing_config(processor=processor, inputs=inputs, outputs=outputs, container_arguments=container_arguments, container_entrypoint=container_entrypoint, kms_key_id=kms_key_id)
 
         if isinstance(job_name, Placeholder):
-            parameters['ProcessingJobName'] = job_name
+            processing_parameters['ProcessingJobName'] = job_name
 
         if experiment_config is not None:
-            parameters['ExperimentConfig'] = experiment_config
-        
+            processing_parameters['ExperimentConfig'] = experiment_config
+
         if tags:
-            parameters['Tags'] = tags_dict_to_kv_list(tags)
-        
-        if 'S3Operations' in parameters:
-            del parameters['S3Operations']
-        
-        kwargs[Field.Parameters.value] = parameters
+            processing_parameters['Tags'] = tags if isinstance(tags, Placeholder) else tags_dict_to_kv_list(tags)
+
+        if 'S3Operations' in processing_parameters:
+            del processing_parameters['S3Operations']
+
+        if Field.Parameters.value in kwargs and isinstance(kwargs[Field.Parameters.value], dict):
+            # Update processing_parameters with input parameters
+            merge_dicts(processing_parameters, kwargs[Field.Parameters.value])
 
+        kwargs[Field.Parameters.value] = processing_parameters
         super(ProcessingStep, self).__init__(state_id, **kwargs)