fix: Add the property UseAliasAsEventTarget to AWS::Serverless::StateMachine and make associated events use the state machine alias as a target.

DEVELOPMENT_GUIDE.md

@@ -59,10 +59,10 @@ We format our code using [Black](https://github.com/python/black) and verify the
 during PR checks. Black will be installed automatically with `make init`.
 
 After installing, you can run our formatting through our Makefile by `make format` or integrating Black directly in your favorite IDE (instructions
-can be found [here](https://black.readthedocs.io/en/stable/editor_integration.html))
+can be found [here](https://black.readthedocs.io/en/stable/integrations/editors.html))
 
 ##### (Workaround) Integrating Black directly in your favorite IDE
-Since black is installed in virtualenv, when you follow [this instruction](https://black.readthedocs.io/en/stable/editor_integration.html), `which black` might give you this
+Since black is installed in virtualenv, when you follow [this instruction](https://black.readthedocs.io/en/stable/integrations/editors.html), `which black` might give you this
 
 ```bash
 (sam38) $ where black

samtranslator/internal/schema_source/aws_serverless_statemachine.py

@@ -174,6 +174,7 @@ class Properties(BaseModel):
     Type: Optional[PassThroughProp] = properties("Type")
     AutoPublishAlias: Optional[PassThroughProp]
     DeploymentPreference: Optional[PassThroughProp]
+    UseAliasAsEventTarget: Optional[bool]
 
 
 class Resource(ResourceAttributes):

samtranslator/model/sam_resources.py

@@ -1778,6 +1778,7 @@ class SamStateMachine(SamResourceMacro):
         "PermissionsBoundary": PropertyType(False, IS_STR),
         "AutoPublishAlias": PassThroughProperty(False),
         "DeploymentPreference": MutatedPassThroughProperty(False),
+        "UseAliasAsEventTarget": Property(False, IS_BOOL),
     }
 
     Definition: Optional[Dict[str, Any]]
@@ -1796,6 +1797,7 @@ class SamStateMachine(SamResourceMacro):
     PermissionsBoundary: Optional[Intrinsicable[str]]
     AutoPublishAlias: Optional[PassThrough]
     DeploymentPreference: Optional[PassThrough]
+    UseAliasAsEventTarget: Optional[bool]
 
     event_resolver = ResourceTypeResolver(
         samtranslator.model.stepfunctions.events,
@@ -1834,6 +1836,7 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
             get_managed_policy_map=get_managed_policy_map,
             auto_publish_alias=self.AutoPublishAlias,
             deployment_preference=self.DeploymentPreference,
+            use_alias_as_event_target=self.UseAliasAsEventTarget,
         )
 
         generated_resources = state_machine_generator.to_cloudformation()

samtranslator/model/stepfunctions/generators.py

@@ -54,6 +54,7 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: PLR0913
         get_managed_policy_map=None,
         auto_publish_alias=None,
         deployment_preference=None,
+        use_alias_as_event_target=None,
     ):
         """
         Constructs an State Machine Generator class that generates a State Machine resource
@@ -81,6 +82,7 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: PLR0913
         :param passthrough_resource_attributes: Attributes such as `Condition` that are added to derived resources
         :param auto_publish_alias: Name of the state machine alias to automatically create and update
         :deployment_preference: Settings to enable gradual state machine deployments
+        :param use_alias_as_event_target: Whether to use the state machine alias as the event target
         """
         self.logical_id = logical_id
         self.depends_on = depends_on
@@ -110,6 +112,7 @@ def __init__(  # type: ignore[no-untyped-def] # noqa: PLR0913
         self.get_managed_policy_map = get_managed_policy_map
         self.auto_publish_alias = auto_publish_alias
         self.deployment_preference = deployment_preference
+        self.use_alias_as_event_target = use_alias_as_event_target
 
     @cw_timer(prefix="Generator", name="StateMachine")
     def to_cloudformation(self):  # type: ignore[no-untyped-def]
@@ -300,6 +303,8 @@ def _construct_alias(self, version: StepFunctionsStateMachineVersion) -> StepFun
         deployment_preference["StateMachineVersionArn"] = state_machine_version_arn
         state_machine_alias.DeploymentPreference = deployment_preference
 
+        self.state_machine_alias = state_machine_alias
+
         return state_machine_alias
 
     def _generate_managed_traffic_shifting_resources(
@@ -310,6 +315,10 @@ def _generate_managed_traffic_shifting_resources(
         :returns: a list containing the state machine's version and alias resources
         :rtype: list
         """
+        if not self.auto_publish_alias and self.use_alias_as_event_target:
+            raise InvalidResourceException(
+                self.logical_id, "'UseAliasAsEventTarget' requires 'AutoPublishAlias' property to be specified."
+            )
         if not self.auto_publish_alias and not self.deployment_preference:
             return []
         if not self.auto_publish_alias and self.deployment_preference:
@@ -341,7 +350,12 @@ def _generate_event_resources(self) -> List[Dict[str, Any]]:
                         kwargs[name] = resource
                 except (TypeError, AttributeError) as e:
                     raise InvalidEventException(logical_id, str(e)) from e
-                resources += eventsource.to_cloudformation(resource=self.state_machine, **kwargs)
+                target_resource = (
+                    (self.state_machine_alias or self.state_machine)
+                    if self.use_alias_as_event_target
+                    else self.state_machine
+                )
+                resources += eventsource.to_cloudformation(resource=target_resource, **kwargs)
 
         return resources
 

samtranslator/schema/schema.json

@@ -281000,6 +281000,10 @@
           ],
           "markdownDescription": "The type of the state machine\\.  \n*Valid values*: `STANDARD` or `EXPRESS`  \n*Type*: String  \n*Required*: No  \n*Default*: `STANDARD`  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`StateMachineType`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-statemachinetype) property of an `AWS::StepFunctions::StateMachine` resource\\.",
           "title": "Type"
+        },
+        "UseAliasAsEventTarget": {
+          "title": "Usealiasaseventtarget",
+          "type": "boolean"
         }
       },
       "title": "Properties",

schema_source/sam.schema.json

@@ -8020,6 +8020,10 @@
           ],
           "markdownDescription": "The type of the state machine\\.  \n*Valid values*: `STANDARD` or `EXPRESS`  \n*Type*: String  \n*Required*: No  \n*Default*: `STANDARD`  \n*AWS CloudFormation compatibility*: This property is passed directly to the [`StateMachineType`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-stepfunctions-statemachine.html#cfn-stepfunctions-statemachine-statemachinetype) property of an `AWS::StepFunctions::StateMachine` resource\\.",
           "title": "Type"
+        },
+        "UseAliasAsEventTarget": {
+          "title": "Usealiasaseventtarget",
+          "type": "boolean"
         }
       },
       "title": "Properties",

tests/model/stepfunctions/test_state_machine_generator.py

@@ -148,6 +148,37 @@ def test_state_machine_with_unsupported_event_source(self):
         with self.assertRaises(InvalidEventException):
             StateMachineGenerator(**self.kwargs).to_cloudformation()
 
+    def test_state_machine_with_alias_as_event_source_target(self):
+        self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
+        self.kwargs["role"] = "my-test-role-arn"
+        self.kwargs["use_alias_as_event_target"] = True
+        self.kwargs["auto_publish_alias"] = "live"
+        event_resolver = Mock()
+        event_resolver.resolve_resource_type = Mock(return_value=CloudWatchEvent)
+        self.kwargs["event_resolver"] = event_resolver
+        self.kwargs["events"] = {
+            "CWEEvent": {"Type": "CloudWatchEvent", "Properties": {"Pattern": {"detail": {"state": ["terminated"]}}}}
+        }
+        self.kwargs["event_resources"] = {"CWEEvent": {}}
+        state_machine_generator = StateMachineGenerator(**self.kwargs)
+        state_machine_generator._generate_managed_traffic_shifting_resources()
+        generated_event_resources = state_machine_generator._generate_event_resources()
+        self.assertEqual(generated_event_resources[0].Targets[0]["Arn"], {"Ref": "StateMachineIdAliaslive"})
+
+    def test_state_machine_with_alias_as_event_source_target_requires_alias(self):
+        self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
+        self.kwargs["role"] = "my-test-role-arn"
+        self.kwargs["use_alias_as_event_target"] = True
+        self.kwargs["deployment_preference"] = {"Type": "ALL_AT_ONCE"}
+        # Missing property
+        # self.kwargs["auto_publish_alias"] = "live"
+        with self.assertRaises(InvalidResourceException) as error:
+            StateMachineGenerator(**self.kwargs).to_cloudformation()
+        self.assertEqual(
+            error.exception.message,
+            "Resource with id [StateMachineId] is invalid. 'UseAliasAsEventTarget' requires 'AutoPublishAlias' property to be specified.",
+        )
+
     def test_state_machine_with_managed_traffic_shifting_properties(self):
         self.kwargs["definition_uri"] = "s3://mybucket/myASLfile"
         self.kwargs["role"] = "my-test-role-arn"