# AWS Step Functions Construct Library
<!--BEGIN STABILITY BANNER-->---
> AWS CDK v1 has reached End-of-Support on 2023-06-01.
> This package is no longer being updated, and users should migrate to AWS CDK v2.
>
> For more information on how to migrate, see the [*Migrating to AWS CDK v2* guide](https://docs.aws.amazon.com/cdk/v2/guide/migrating-v2.html).
---
<!--END STABILITY BANNER-->
The `@aws-cdk/aws-stepfunctions` package contains constructs for building
serverless workflows using objects. Use this in conjunction with the
`@aws-cdk/aws-stepfunctions-tasks` package, which contains classes used
to call other AWS services.
Defining a workflow looks like this (for the [Step Functions Job Poller
example](https://docs.aws.amazon.com/step-functions/latest/dg/job-status-poller-sample.html)):
## Example
```python
import aws_cdk.aws_lambda as lambda_
# submit_lambda: lambda.Function
# get_status_lambda: lambda.Function
submit_job = tasks.LambdaInvoke(self, "Submit Job",
lambda_function=submit_lambda,
# Lambda's result is in the attribute `Payload`
output_path="$.Payload"
)
wait_x = sfn.Wait(self, "Wait X Seconds",
time=sfn.WaitTime.seconds_path("$.waitSeconds")
)
get_status = tasks.LambdaInvoke(self, "Get Job Status",
lambda_function=get_status_lambda,
# Pass just the field named "guid" into the Lambda, put the
# Lambda's result in a field called "status" in the response
input_path="$.guid",
output_path="$.Payload"
)
job_failed = sfn.Fail(self, "Job Failed",
cause="AWS Batch Job Failed",
error="DescribeJob returned FAILED"
)
final_status = tasks.LambdaInvoke(self, "Get Final Job Status",
lambda_function=get_status_lambda,
# Use "guid" field as input
input_path="$.guid",
output_path="$.Payload"
)
definition = submit_job.next(wait_x).next(get_status).next(sfn.Choice(self, "Job Complete?").when(sfn.Condition.string_equals("$.status", "FAILED"), job_failed).when(sfn.Condition.string_equals("$.status", "SUCCEEDED"), final_status).otherwise(wait_x))
sfn.StateMachine(self, "StateMachine",
definition=definition,
timeout=Duration.minutes(5)
)
```
You can find more sample snippets and learn more about the service integrations
in the `@aws-cdk/aws-stepfunctions-tasks` package.
## State Machine
A `stepfunctions.StateMachine` is a resource that takes a state machine
definition. The definition is specified by its start state, and encompasses
all states reachable from the start state:
```python
start_state = sfn.Pass(self, "StartState")
sfn.StateMachine(self, "StateMachine",
definition=start_state
)
```
State machines execute using an IAM Role, which will automatically have all
permissions added that are required to make all state machine tasks execute
properly (for example, permissions to invoke any Lambda functions you add to
your workflow). A role will be created by default, but you can supply an
existing one as well.
## Accessing State (the JsonPath class)
Every State Machine execution has [State Machine
Data](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-state-machine-data.html):
a JSON document containing keys and values that is fed into the state machine,
gets modified as the state machine progresses, and finally is produced as output.
You can pass fragments of this State Machine Data into Tasks of the state machine.
To do so, use the static methods on the `JsonPath` class. For example, to pass
the value that's in the data key of `OrderId` to a Lambda function as you invoke
it, use `JsonPath.stringAt('$.OrderId')`, like so:
```python
import aws_cdk.aws_lambda as lambda_
# order_fn: lambda.Function
submit_job = tasks.LambdaInvoke(self, "InvokeOrderProcessor",
lambda_function=order_fn,
payload=sfn.TaskInput.from_object({
"OrderId": sfn.JsonPath.string_at("$.OrderId")
})
)
```
The following methods are available:
| Method | Purpose |
|--------|---------|
| `JsonPath.stringAt('$.Field')` | reference a field, return the type as a `string`. |
| `JsonPath.listAt('$.Field')` | reference a field, return the type as a list of strings. |
| `JsonPath.numberAt('$.Field')` | reference a field, return the type as a number. Use this for functions that expect a number argument. |
| `JsonPath.objectAt('$.Field')` | reference a field, return the type as an `IResolvable`. Use this for functions that expect an object argument. |
| `JsonPath.entirePayload` | reference the entire data object (equivalent to a path of `$`). |
| `JsonPath.taskToken` | reference the [Task Token](https://docs.aws.amazon.com/step-functions/latest/dg/connect-to-resource.html#connect-wait-token), used for integration patterns that need to run for a long time. |
You can also call [intrinsic functions](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-intrinsic-functions.html) using the methods on `JsonPath`:
| Method | Purpose |
|--------|---------|
| `JsonPath.array(JsonPath.stringAt('$.Field'), ...)` | make an array from other elements. |
| `JsonPath.format('The value is {}.', JsonPath.stringAt('$.Value'))` | insert elements into a format string. |
| `JsonPath.stringToJson(JsonPath.stringAt('$.ObjStr'))` | parse a JSON string to an object |
| `JsonPath.jsonToString(JsonPath.objectAt('$.Obj'))` | stringify an object to a JSON string |
## Amazon States Language
This library comes with a set of classes that model the [Amazon States
Language](https://states-language.net/spec.html). The following State classes
are supported:
* [`Task`](#task)
* [`Pass`](#pass)
* [`Wait`](#wait)
* [`Choice`](#choice)
* [`Parallel`](#parallel)
* [`Succeed`](#succeed)
* [`Fail`](#fail)
* [`Map`](#map)
* [`Custom State`](#custom-state)
An arbitrary JSON object (specified at execution start) is passed from state to
state and transformed during the execution of the workflow. For more
information, see the States Language spec.
### Task
A `Task` represents some work that needs to be done. The exact work to be
done is determine by a class that implements `IStepFunctionsTask`, a collection
of which can be found in the `@aws-cdk/aws-stepfunctions-tasks` module.
The tasks in the `@aws-cdk/aws-stepfunctions-tasks` module support the
[service integration pattern](https://docs.aws.amazon.com/step-functions/latest/dg/connect-to-resource.html) that integrates Step Functions with services
directly in the Amazon States language.
### Pass
A `Pass` state passes its input to its output, without performing work.
Pass states are useful when constructing and debugging state machines.
The following example injects some fixed data into the state machine through
the `result` field. The `result` field will be added to the input and the result
will be passed as the state's output.
```python
# Makes the current JSON state { ..., "subObject": { "hello": "world" } }
pass = sfn.Pass(self, "Add Hello World",
result=sfn.Result.from_object({"hello": "world"}),
result_path="$.subObject"
)
# Set the next state
next_state = sfn.Pass(self, "NextState")
pass.next(next_state)
```
The `Pass` state also supports passing key-value pairs as input. Values can
be static, or selected from the input with a path.
The following example filters the `greeting` field from the state input
and also injects a field called `otherData`.
```python
pass = sfn.Pass(self, "Filter input and inject data",
parameters={ # input to the pass state
"input": sfn.JsonPath.string_at("$.input.greeting"),
"other_data": "some-extra-stuff"}
)
```
The object specified in `parameters` will be the input of the `Pass` state.
Since neither `Result` nor `ResultPath` are supplied, the `Pass` state copies
its input through to its output.
Learn more about the [Pass state](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-pass-state.html)
### Wait
A `Wait` state waits for a given number of seconds, or until the current time
hits a particular time. The time to wait may be taken from the execution's JSON
state.
```python
# Wait until it's the time mentioned in the the state object's "triggerTime"
# field.
wait = sfn.Wait(self, "Wait For Trigger Time",
time=sfn.WaitTime.timestamp_path("$.triggerTime")
)
# Set the next state
start_the_work = sfn.Pass(self, "StartTheWork")
wait.next(start_the_work)
```
### Choice
A `Choice` state can take a different path through the workflow based on the
values in the execution's JSON state:
```python
choice = sfn.Choice(self, "Did it work?")
# Add conditions with .when()
success_state = sfn.Pass(self, "SuccessState")
failure_state = sfn.Pass(self, "FailureState")
choice.when(sfn.Condition.string_equals("$.status", "SUCCESS"), success_state)
choice.when(sfn.Condition.number_greater_than("$.attempts", 5), failure_state)
# Use .otherwise() to indicate what should be done if none of the conditions match
try_again_state = sfn.Pass(self, "TryAgainState")
choice.otherwise(try_again_state)
```
If you want to temporarily branch your workflow based on a condition, but have
all branches come together and continuing as one (similar to how an `if ... then ... else` works in a programming language), use the `.afterwards()` method:
```python
choice = sfn.Choice(self, "What color is it?")
handle_blue_item = sfn.Pass(self, "HandleBlueItem")
handle_red_item = sfn.Pass(self, "HandleRedItem")
handle_other_item_color = sfn.Pass(self, "HanldeOtherItemColor")
choice.when(sfn.Condition.string_equals("$.color", "BLUE"), handle_blue_item)
choice.when(sfn.Condition.string_equals("$.color", "RED"), handle_red_item)
choice.otherwise(handle_other_item_color)
# Use .afterwards() to join all possible paths back together and continue
ship_the_item = sfn.Pass(self, "ShipTheItem")
choice.afterwards().next(ship_the_item)
```
If your `Choice` doesn't have an `otherwise()` and none of the conditions match
the JSON state, a `NoChoiceMatched` error will be thrown. Wrap the state machine
in a `Parallel` state if you want to catch and recover from this.
#### Available Conditions
see [step function comparison operators](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-choice-state.html#amazon-states-language-choice-state-rules)
* `Condition.isPresent` - matches if a json path is present
* `Condition.isNotPresent` - matches if a json path is not present
* `Condition.isString` - matches if a json path contains a string
* `Condition.isNotString` - matches if a json path is not a string
* `Condition.isNumeric` - matches if a json path is numeric
* `Condition.isNotNumeric` - matches if a json path is not numeric
* `Condition.isBoolean` - matches if a json path is boolean
* `Condition.isNotBoolean` - matches if a json path is not boolean
* `Condition.isTimestamp` - matches if a json path is a timestamp
* `Condition.isNotTimestamp` - matches if a json path is not a timestamp
* `Condition.isNotNull` - matches if a json path is not null
* `Condition.isNull` - matches if a json path is null
* `Condition.booleanEquals` - matches if a boolean field has a given value
* `Condition.booleanEqualsJsonPath` - matches if a boolean field equals a value in a given mapping path
* `Condition.stringEqualsJsonPath` - matches if a string field equals a given mapping path
* `Condition.stringEquals` - matches if a field equals a string value
* `Condition.stringLessThan` - matches if a string field sorts before a given value
* `Condition.stringLessThanJsonPath` - matches if a string field sorts before a value at given mapping path
* `Condition.stringLessThanEquals` - matches if a string field sorts equal to or before a given value
* `Condition.stringLessThanEqualsJsonPath` - matches if a string field sorts equal to or before a given mapping
* `Condition.stringGreaterThan` - matches if a string field sorts after a given value
* `Condition.stringGreaterThanJsonPath` - matches if a string field sorts after a value at a given mapping path
* `Condition.stringGreaterThanEqualsJsonPath` - matches if a string field sorts after or equal to value at a given mapping path
* `Condition.stringGreaterThanEquals` - matches if a string field sorts after or equal to a given value
* `Condition.numberEquals` - matches if a numeric field has the given value
* `Condition.numberEqualsJsonPath` - matches if a numeric field has the value in a given mapping path
* `Condition.numberLessThan` - matches if a numeric field is less than the given value
* `Condition.numberLessThanJsonPath` - matches if a numeric field is less than the value at the given mapping path
* `Condition.numberLessThanEquals` - matches if a numeric field is less than or equal to the given value
* `Condition.numberLessThanEqualsJsonPath` - matches if a numeric field is less than or equal to the numeric value at given mapping path
* `Condition.numberGreaterThan` - matches if a numeric field is greater than the given value
* `Condition.numberGreaterThanJsonPath` - matches if a numeric field is greater than the value at a given mapping path
* `Condition.numberGreaterThanEquals` - matches if a numeric field is greater than or equal to the given value
* `Condition.numberGreaterThanEqualsJsonPath` - matches if a numeric field is greater than or equal to the value at a given mapping path
* `Condition.timestampEquals` - matches if a timestamp field is the same time as the given timestamp
* `Condition.timestampEqualsJsonPath` - matches if a timestamp field is the same time as the timestamp at a given mapping path
* `Condition.timestampLessThan` - matches if a timestamp field is before the given timestamp
* `Condition.timestampLessThanJsonPath` - matches if a timestamp field is before the timestamp at a given mapping path
* `Condition.timestampLessThanEquals` - matches if a timestamp field is before or equal to the given timestamp
* `Condition.timestampLessThanEqualsJsonPath` - matches if a timestamp field is before or equal to the timestamp at a given mapping path
* `Condition.timestampGreaterThan` - matches if a timestamp field is after the timestamp at a given mapping path
* `Condition.timestampGreaterThanJsonPath` - matches if a timestamp field is after the timestamp at a given mapping path
* `Condition.timestampGreaterThanEquals` - matches if a timestamp field is after or equal to the given timestamp
* `Condition.timestampGreaterThanEqualsJsonPath` - matches if a timestamp field is after or equal to the timestamp at a given mapping path
* `Condition.stringMatches` - matches if a field matches a string pattern that can contain a wild card (*) e.g: log-*.txt or *LATEST*. No other characters other than "*" have any special meaning - * can be escaped: \\*
### Parallel
A `Parallel` state executes one or more subworkflows in parallel. It can also
be used to catch and recover from errors in subworkflows.
```python
parallel = sfn.Parallel(self, "Do the work in parallel")
# Add branches to be executed in parallel
ship_item = sfn.Pass(self, "ShipItem")
send_invoice = sfn.Pass(self, "SendInvoice")
restock = sfn.Pass(self, "Restock")
parallel.branch(ship_item)
parallel.branch(send_invoice)
parallel.branch(restock)
# Retry the whole workflow if something goes wrong
parallel.add_retry(max_attempts=1)
# How to recover from errors
send_failure_notification = sfn.Pass(self, "SendFailureNotification")
parallel.add_catch(send_failure_notification)
# What to do in case everything succeeded
close_order = sfn.Pass(self, "CloseOrder")
parallel.next(close_order)
```
### Succeed
Reaching a `Succeed` state terminates the state machine execution with a
successful status.
```python
success = sfn.Succeed(self, "We did it!")
```
### Fail
Reaching a `Fail` state terminates the state machine execution with a
failure status. The fail state should report the reason for the failure.
Failures can be caught by encompassing `Parallel` states.
```python
success = sfn.Fail(self, "Fail",
error="WorkflowFailure",
cause="Something went wrong"
)
```
### Map
A `Map` state can be used to run a set of steps for each element of an input array.
A `Map` state will execute the same steps for multiple entries of an array in the state input.
While the `Parallel` state executes multiple branches of steps using the same input, a `Map` state will
execute the same steps for multiple entries of an array in the state input.
```python
map = sfn.Map(self, "Map State",
max_concurrency=1,
items_path=sfn.JsonPath.string_at("$.inputForMap")
)
map.iterator(sfn.Pass(self, "Pass State"))
```
### Custom State
It's possible that the high-level constructs for the states or `stepfunctions-tasks` do not have
the states or service integrations you are looking for. The primary reasons for this lack of
functionality are:
* A [service integration](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-service-integrations.html) is available through Amazon States Langauge, but not available as construct
classes in the CDK.
* The state or state properties are available through Step Functions, but are not configurable
through constructs
If a feature is not available, a `CustomState` can be used to supply any Amazon States Language
JSON-based object as the state definition.
[Code Snippets](https://docs.aws.amazon.com/step-functions/latest/dg/tutorial-code-snippet.html#tutorial-code-snippet-1) are available and can be plugged in as the state definition.
Custom states can be chained together with any of the other states to create your state machine
definition. You will also need to provide any permissions that are required to the `role` that
the State Machine uses.
The following example uses the `DynamoDB` service integration to insert data into a DynamoDB table.
```python
import aws_cdk.aws_dynamodb as dynamodb
# create a table
table = dynamodb.Table(self, "montable",
partition_key=dynamodb.Attribute(
name="id",
type=dynamodb.AttributeType.STRING
)
)
final_status = sfn.Pass(self, "final step")
# States language JSON to put an item into DynamoDB
# snippet generated from https://docs.aws.amazon.com/step-functions/latest/dg/tutorial-code-snippet.html#tutorial-code-snippet-1
state_json = {
"Type": "Task",
"Resource": "arn:aws:states:::dynamodb:putItem",
"Parameters": {
"TableName": table.table_name,
"Item": {
"id": {
"S": "MyEntry"
}
}
},
"ResultPath": null
}
# custom state which represents a task to insert data into DynamoDB
custom = sfn.CustomState(self, "my custom task",
state_json=state_json
)
chain = sfn.Chain.start(custom).next(final_status)
sm = sfn.StateMachine(self, "StateMachine",
definition=chain,
timeout=Duration.seconds(30)
)
# don't forget permissions. You need to assign them
table.grant_write_data(sm)
```
## Task Chaining
To make defining work flows as convenient (and readable in a top-to-bottom way)
as writing regular programs, it is possible to chain most methods invocations.
In particular, the `.next()` method can be repeated. The result of a series of
`.next()` calls is called a **Chain**, and can be used when defining the jump
targets of `Choice.on` or `Parallel.branch`:
```python
step1 = sfn.Pass(self, "Step1")
step2 = sfn.Pass(self, "Step2")
step3 = sfn.Pass(self, "Step3")
step4 = sfn.Pass(self, "Step4")
step5 = sfn.Pass(self, "Step5")
step6 = sfn.Pass(self, "Step6")
step7 = sfn.Pass(self, "Step7")
step8 = sfn.Pass(self, "Step8")
step9 = sfn.Pass(self, "Step9")
step10 = sfn.Pass(self, "Step10")
choice = sfn.Choice(self, "Choice")
condition1 = sfn.Condition.string_equals("$.status", "SUCCESS")
parallel = sfn.Parallel(self, "Parallel")
finish = sfn.Pass(self, "Finish")
definition = step1.next(step2).next(choice.when(condition1, step3.next(step4).next(step5)).otherwise(step6).afterwards()).next(parallel.branch(step7.next(step8)).branch(step9.next(step10))).next(finish)
sfn.StateMachine(self, "StateMachine",
definition=definition
)
```
If you don't like the visual look of starting a chain directly off the first
step, you can use `Chain.start`:
```python
step1 = sfn.Pass(self, "Step1")
step2 = sfn.Pass(self, "Step2")
step3 = sfn.Pass(self, "Step3")
definition = sfn.Chain.start(step1).next(step2).next(step3)
```
## State Machine Fragments
It is possible to define reusable (or abstracted) mini-state machines by
defining a construct that implements `IChainable`, which requires you to define
two fields:
* `startState: State`, representing the entry point into this state machine.
* `endStates: INextable[]`, representing the (one or more) states that outgoing
transitions will be added to if you chain onto the fragment.
Since states will be named after their construct IDs, you may need to prefix the
IDs of states if you plan to instantiate the same state machine fragment
multiples times (otherwise all states in every instantiation would have the same
name).
The class `StateMachineFragment` contains some helper functions (like
`prefixStates()`) to make it easier for you to do this. If you define your state
machine as a subclass of this, it will be convenient to use:
```python
from aws_cdk.core import Stack
from constructs import Construct
import aws_cdk.aws_stepfunctions as sfn
class MyJob(sfn.StateMachineFragment):
def __init__(self, parent, id, *, jobFlavor):
super().__init__(parent, id)
choice = sfn.Choice(self, "Choice").when(sfn.Condition.string_equals("$.branch", "left"), sfn.Pass(self, "Left Branch")).when(sfn.Condition.string_equals("$.branch", "right"), sfn.Pass(self, "Right Branch"))
# ...
self.start_state = choice
self.end_states = choice.afterwards().end_states
class MyStack(Stack):
def __init__(self, scope, id):
super().__init__(scope, id)
# Do 3 different variants of MyJob in parallel
parallel = sfn.Parallel(self, "All jobs").branch(MyJob(self, "Quick", job_flavor="quick").prefix_states()).branch(MyJob(self, "Medium", job_flavor="medium").prefix_states()).branch(MyJob(self, "Slow", job_flavor="slow").prefix_states())
sfn.StateMachine(self, "MyStateMachine",
definition=parallel
)
```
A few utility functions are available to parse state machine fragments.
* `State.findReachableStates`: Retrieve the list of states reachable from a given state.
* `State.findReachableEndStates`: Retrieve the list of end or terminal states reachable from a given state.
## Activity
**Activities** represent work that is done on some non-Lambda worker pool. The
Step Functions workflow will submit work to this Activity, and a worker pool
that you run yourself, probably on EC2, will pull jobs from the Activity and
submit the results of individual jobs back.
You need the ARN to do so, so if you use Activities be sure to pass the Activity
ARN into your worker pool:
```python
activity = sfn.Activity(self, "Activity")
# Read this CloudFormation Output from your application and use it to poll for work on
# the activity.
CfnOutput(self, "ActivityArn", value=activity.activity_arn)
```
### Activity-Level Permissions
Granting IAM permissions to an activity can be achieved by calling the `grant(principal, actions)` API:
```python
activity = sfn.Activity(self, "Activity")
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
activity.grant(role, "states:SendTaskSuccess")
```
This will grant the IAM principal the specified actions onto the activity.
## Metrics
`Task` object expose various metrics on the execution of that particular task. For example,
to create an alarm on a particular task failing:
```python
# task: sfn.Task
cloudwatch.Alarm(self, "TaskAlarm",
metric=task.metric_failed(),
threshold=1,
evaluation_periods=1
)
```
There are also metrics on the complete state machine:
```python
# state_machine: sfn.StateMachine
cloudwatch.Alarm(self, "StateMachineAlarm",
metric=state_machine.metric_failed(),
threshold=1,
evaluation_periods=1
)
```
And there are metrics on the capacity of all state machines in your account:
```python
cloudwatch.Alarm(self, "ThrottledAlarm",
metric=sfn.StateTransitionMetric.metric_throttled_events(),
threshold=10,
evaluation_periods=2
)
```
## Error names
Step Functions identifies errors in the Amazon States Language using case-sensitive strings, known as error names.
The Amazon States Language defines a set of built-in strings that name well-known errors, all beginning with the `States.` prefix.
* `States.ALL` - A wildcard that matches any known error name.
* `States.Runtime` - An execution failed due to some exception that could not be processed. Often these are caused by errors at runtime, such as attempting to apply InputPath or OutputPath on a null JSON payload. A `States.Runtime` error is not retriable, and will always cause the execution to fail. A retry or catch on `States.ALL` will NOT catch States.Runtime errors.
* `States.DataLimitExceeded` - A States.DataLimitExceeded exception will be thrown for the following:
* When the output of a connector is larger than payload size quota.
* When the output of a state is larger than payload size quota.
* When, after Parameters processing, the input of a state is larger than the payload size quota.
* See [the AWS documentation](https://docs.aws.amazon.com/step-functions/latest/dg/limits-overview.html) to learn more about AWS Step Functions Quotas.
* `States.HeartbeatTimeout` - A Task state failed to send a heartbeat for a period longer than the HeartbeatSeconds value.
* `States.Timeout` - A Task state either ran longer than the TimeoutSeconds value, or failed to send a heartbeat for a period longer than the HeartbeatSeconds value.
* `States.TaskFailed`- A Task state failed during the execution. When used in a retry or catch, `States.TaskFailed` acts as a wildcard that matches any known error name except for `States.Timeout`.
## Logging
Enable logging to CloudWatch by passing a logging configuration with a
destination LogGroup:
```python
import aws_cdk.aws_logs as logs
log_group = logs.LogGroup(self, "MyLogGroup")
sfn.StateMachine(self, "MyStateMachine",
definition=sfn.Chain.start(sfn.Pass(self, "Pass")),
logs=sfn.LogOptions(
destination=log_group,
level=sfn.LogLevel.ALL
)
)
```
## X-Ray tracing
Enable X-Ray tracing for StateMachine:
```python
sfn.StateMachine(self, "MyStateMachine",
definition=sfn.Chain.start(sfn.Pass(self, "Pass")),
tracing_enabled=True
)
```
See [the AWS documentation](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-xray-tracing.html)
to learn more about AWS Step Functions's X-Ray support.
## State Machine Permission Grants
IAM roles, users, or groups which need to be able to work with a State Machine should be granted IAM permissions.
Any object that implements the `IGrantable` interface (has an associated principal) can be granted permissions by calling:
* `stateMachine.grantStartExecution(principal)` - grants the principal the ability to execute the state machine
* `stateMachine.grantRead(principal)` - grants the principal read access
* `stateMachine.grantTaskResponse(principal)` - grants the principal the ability to send task tokens to the state machine
* `stateMachine.grantExecution(principal, actions)` - grants the principal execution-level permissions for the IAM actions specified
* `stateMachine.grant(principal, actions)` - grants the principal state-machine-level permissions for the IAM actions specified
### Start Execution Permission
Grant permission to start an execution of a state machine by calling the `grantStartExecution()` API.
```python
# definition: sfn.IChainable
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
state_machine = sfn.StateMachine(self, "StateMachine",
definition=definition
)
# Give role permission to start execution of state machine
state_machine.grant_start_execution(role)
```
The following permission is provided to a service principal by the `grantStartExecution()` API:
* `states:StartExecution` - to state machine
### Read Permissions
Grant `read` access to a state machine by calling the `grantRead()` API.
```python
# definition: sfn.IChainable
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
state_machine = sfn.StateMachine(self, "StateMachine",
definition=definition
)
# Give role read access to state machine
state_machine.grant_read(role)
```
The following read permissions are provided to a service principal by the `grantRead()` API:
* `states:ListExecutions` - to state machine
* `states:ListStateMachines` - to state machine
* `states:DescribeExecution` - to executions
* `states:DescribeStateMachineForExecution` - to executions
* `states:GetExecutionHistory` - to executions
* `states:ListActivities` - to `*`
* `states:DescribeStateMachine` - to `*`
* `states:DescribeActivity` - to `*`
### Task Response Permissions
Grant permission to allow task responses to a state machine by calling the `grantTaskResponse()` API:
```python
# definition: sfn.IChainable
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
state_machine = sfn.StateMachine(self, "StateMachine",
definition=definition
)
# Give role task response permissions to the state machine
state_machine.grant_task_response(role)
```
The following read permissions are provided to a service principal by the `grantRead()` API:
* `states:SendTaskSuccess` - to state machine
* `states:SendTaskFailure` - to state machine
* `states:SendTaskHeartbeat` - to state machine
### Execution-level Permissions
Grant execution-level permissions to a state machine by calling the `grantExecution()` API:
```python
# definition: sfn.IChainable
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
state_machine = sfn.StateMachine(self, "StateMachine",
definition=definition
)
# Give role permission to get execution history of ALL executions for the state machine
state_machine.grant_execution(role, "states:GetExecutionHistory")
```
### Custom Permissions
You can add any set of permissions to a state machine by calling the `grant()` API.
```python
# definition: sfn.IChainable
user = iam.User(self, "MyUser")
state_machine = sfn.StateMachine(self, "StateMachine",
definition=definition
)
# give user permission to send task success to the state machine
state_machine.grant(user, "states:SendTaskSuccess")
```
## Import
Any Step Functions state machine that has been created outside the stack can be imported
into your CDK stack.
State machines can be imported by their ARN via the `StateMachine.fromStateMachineArn()` API
```python
app = App()
stack = Stack(app, "MyStack")
sfn.StateMachine.from_state_machine_arn(stack, "ImportedStateMachine", "arn:aws:states:us-east-1:123456789012:stateMachine:StateMachine2E01A3A5-N5TJppzoevKQ")
```
Raw data
{
"_id": null,
"home_page": "https://github.com/aws/aws-cdk",
"name": "aws-cdk.aws-stepfunctions",
"maintainer": "",
"docs_url": null,
"requires_python": "~=3.7",
"maintainer_email": "",
"keywords": "",
"author": "Amazon Web Services",
"author_email": "",
"download_url": "https://files.pythonhosted.org/packages/16/b5/15672f07e881d26e2cebaddd73c20e99476deac7bb1f94975079a1ec7e60/aws-cdk.aws-stepfunctions-1.204.0.tar.gz",
"platform": null,
"description": "# AWS Step Functions Construct Library\n\n<!--BEGIN STABILITY BANNER-->---\n\n\n\n\n> AWS CDK v1 has reached End-of-Support on 2023-06-01.\n> This package is no longer being updated, and users should migrate to AWS CDK v2.\n>\n> For more information on how to migrate, see the [*Migrating to AWS CDK v2* guide](https://docs.aws.amazon.com/cdk/v2/guide/migrating-v2.html).\n\n---\n<!--END STABILITY BANNER-->\n\nThe `@aws-cdk/aws-stepfunctions` package contains constructs for building\nserverless workflows using objects. Use this in conjunction with the\n`@aws-cdk/aws-stepfunctions-tasks` package, which contains classes used\nto call other AWS services.\n\nDefining a workflow looks like this (for the [Step Functions Job Poller\nexample](https://docs.aws.amazon.com/step-functions/latest/dg/job-status-poller-sample.html)):\n\n## Example\n\n```python\nimport aws_cdk.aws_lambda as lambda_\n\n# submit_lambda: lambda.Function\n# get_status_lambda: lambda.Function\n\n\nsubmit_job = tasks.LambdaInvoke(self, \"Submit Job\",\n lambda_function=submit_lambda,\n # Lambda's result is in the attribute `Payload`\n output_path=\"$.Payload\"\n)\n\nwait_x = sfn.Wait(self, \"Wait X Seconds\",\n time=sfn.WaitTime.seconds_path(\"$.waitSeconds\")\n)\n\nget_status = tasks.LambdaInvoke(self, \"Get Job Status\",\n lambda_function=get_status_lambda,\n # Pass just the field named \"guid\" into the Lambda, put the\n # Lambda's result in a field called \"status\" in the response\n input_path=\"$.guid\",\n output_path=\"$.Payload\"\n)\n\njob_failed = sfn.Fail(self, \"Job Failed\",\n cause=\"AWS Batch Job Failed\",\n error=\"DescribeJob returned FAILED\"\n)\n\nfinal_status = tasks.LambdaInvoke(self, \"Get Final Job Status\",\n lambda_function=get_status_lambda,\n # Use \"guid\" field as input\n input_path=\"$.guid\",\n output_path=\"$.Payload\"\n)\n\ndefinition = submit_job.next(wait_x).next(get_status).next(sfn.Choice(self, \"Job Complete?\").when(sfn.Condition.string_equals(\"$.status\", \"FAILED\"), job_failed).when(sfn.Condition.string_equals(\"$.status\", \"SUCCEEDED\"), final_status).otherwise(wait_x))\n\nsfn.StateMachine(self, \"StateMachine\",\n definition=definition,\n timeout=Duration.minutes(5)\n)\n```\n\nYou can find more sample snippets and learn more about the service integrations\nin the `@aws-cdk/aws-stepfunctions-tasks` package.\n\n## State Machine\n\nA `stepfunctions.StateMachine` is a resource that takes a state machine\ndefinition. The definition is specified by its start state, and encompasses\nall states reachable from the start state:\n\n```python\nstart_state = sfn.Pass(self, \"StartState\")\n\nsfn.StateMachine(self, \"StateMachine\",\n definition=start_state\n)\n```\n\nState machines execute using an IAM Role, which will automatically have all\npermissions added that are required to make all state machine tasks execute\nproperly (for example, permissions to invoke any Lambda functions you add to\nyour workflow). A role will be created by default, but you can supply an\nexisting one as well.\n\n## Accessing State (the JsonPath class)\n\nEvery State Machine execution has [State Machine\nData](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-state-machine-data.html):\na JSON document containing keys and values that is fed into the state machine,\ngets modified as the state machine progresses, and finally is produced as output.\n\nYou can pass fragments of this State Machine Data into Tasks of the state machine.\nTo do so, use the static methods on the `JsonPath` class. For example, to pass\nthe value that's in the data key of `OrderId` to a Lambda function as you invoke\nit, use `JsonPath.stringAt('$.OrderId')`, like so:\n\n```python\nimport aws_cdk.aws_lambda as lambda_\n\n# order_fn: lambda.Function\n\n\nsubmit_job = tasks.LambdaInvoke(self, \"InvokeOrderProcessor\",\n lambda_function=order_fn,\n payload=sfn.TaskInput.from_object({\n \"OrderId\": sfn.JsonPath.string_at(\"$.OrderId\")\n })\n)\n```\n\nThe following methods are available:\n\n| Method | Purpose |\n|--------|---------|\n| `JsonPath.stringAt('$.Field')` | reference a field, return the type as a `string`. |\n| `JsonPath.listAt('$.Field')` | reference a field, return the type as a list of strings. |\n| `JsonPath.numberAt('$.Field')` | reference a field, return the type as a number. Use this for functions that expect a number argument. |\n| `JsonPath.objectAt('$.Field')` | reference a field, return the type as an `IResolvable`. Use this for functions that expect an object argument. |\n| `JsonPath.entirePayload` | reference the entire data object (equivalent to a path of `$`). |\n| `JsonPath.taskToken` | reference the [Task Token](https://docs.aws.amazon.com/step-functions/latest/dg/connect-to-resource.html#connect-wait-token), used for integration patterns that need to run for a long time. |\n\nYou can also call [intrinsic functions](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-intrinsic-functions.html) using the methods on `JsonPath`:\n\n| Method | Purpose |\n|--------|---------|\n| `JsonPath.array(JsonPath.stringAt('$.Field'), ...)` | make an array from other elements. |\n| `JsonPath.format('The value is {}.', JsonPath.stringAt('$.Value'))` | insert elements into a format string. |\n| `JsonPath.stringToJson(JsonPath.stringAt('$.ObjStr'))` | parse a JSON string to an object |\n| `JsonPath.jsonToString(JsonPath.objectAt('$.Obj'))` | stringify an object to a JSON string |\n\n## Amazon States Language\n\nThis library comes with a set of classes that model the [Amazon States\nLanguage](https://states-language.net/spec.html). The following State classes\nare supported:\n\n* [`Task`](#task)\n* [`Pass`](#pass)\n* [`Wait`](#wait)\n* [`Choice`](#choice)\n* [`Parallel`](#parallel)\n* [`Succeed`](#succeed)\n* [`Fail`](#fail)\n* [`Map`](#map)\n* [`Custom State`](#custom-state)\n\nAn arbitrary JSON object (specified at execution start) is passed from state to\nstate and transformed during the execution of the workflow. For more\ninformation, see the States Language spec.\n\n### Task\n\nA `Task` represents some work that needs to be done. The exact work to be\ndone is determine by a class that implements `IStepFunctionsTask`, a collection\nof which can be found in the `@aws-cdk/aws-stepfunctions-tasks` module.\n\nThe tasks in the `@aws-cdk/aws-stepfunctions-tasks` module support the\n[service integration pattern](https://docs.aws.amazon.com/step-functions/latest/dg/connect-to-resource.html) that integrates Step Functions with services\ndirectly in the Amazon States language.\n\n### Pass\n\nA `Pass` state passes its input to its output, without performing work.\nPass states are useful when constructing and debugging state machines.\n\nThe following example injects some fixed data into the state machine through\nthe `result` field. The `result` field will be added to the input and the result\nwill be passed as the state's output.\n\n```python\n# Makes the current JSON state { ..., \"subObject\": { \"hello\": \"world\" } }\npass = sfn.Pass(self, \"Add Hello World\",\n result=sfn.Result.from_object({\"hello\": \"world\"}),\n result_path=\"$.subObject\"\n)\n\n# Set the next state\nnext_state = sfn.Pass(self, \"NextState\")\npass.next(next_state)\n```\n\nThe `Pass` state also supports passing key-value pairs as input. Values can\nbe static, or selected from the input with a path.\n\nThe following example filters the `greeting` field from the state input\nand also injects a field called `otherData`.\n\n```python\npass = sfn.Pass(self, \"Filter input and inject data\",\n parameters={ # input to the pass state\n \"input\": sfn.JsonPath.string_at(\"$.input.greeting\"),\n \"other_data\": \"some-extra-stuff\"}\n)\n```\n\nThe object specified in `parameters` will be the input of the `Pass` state.\nSince neither `Result` nor `ResultPath` are supplied, the `Pass` state copies\nits input through to its output.\n\nLearn more about the [Pass state](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-pass-state.html)\n\n### Wait\n\nA `Wait` state waits for a given number of seconds, or until the current time\nhits a particular time. The time to wait may be taken from the execution's JSON\nstate.\n\n```python\n# Wait until it's the time mentioned in the the state object's \"triggerTime\"\n# field.\nwait = sfn.Wait(self, \"Wait For Trigger Time\",\n time=sfn.WaitTime.timestamp_path(\"$.triggerTime\")\n)\n\n# Set the next state\nstart_the_work = sfn.Pass(self, \"StartTheWork\")\nwait.next(start_the_work)\n```\n\n### Choice\n\nA `Choice` state can take a different path through the workflow based on the\nvalues in the execution's JSON state:\n\n```python\nchoice = sfn.Choice(self, \"Did it work?\")\n\n# Add conditions with .when()\nsuccess_state = sfn.Pass(self, \"SuccessState\")\nfailure_state = sfn.Pass(self, \"FailureState\")\nchoice.when(sfn.Condition.string_equals(\"$.status\", \"SUCCESS\"), success_state)\nchoice.when(sfn.Condition.number_greater_than(\"$.attempts\", 5), failure_state)\n\n# Use .otherwise() to indicate what should be done if none of the conditions match\ntry_again_state = sfn.Pass(self, \"TryAgainState\")\nchoice.otherwise(try_again_state)\n```\n\nIf you want to temporarily branch your workflow based on a condition, but have\nall branches come together and continuing as one (similar to how an `if ... then ... else` works in a programming language), use the `.afterwards()` method:\n\n```python\nchoice = sfn.Choice(self, \"What color is it?\")\nhandle_blue_item = sfn.Pass(self, \"HandleBlueItem\")\nhandle_red_item = sfn.Pass(self, \"HandleRedItem\")\nhandle_other_item_color = sfn.Pass(self, \"HanldeOtherItemColor\")\nchoice.when(sfn.Condition.string_equals(\"$.color\", \"BLUE\"), handle_blue_item)\nchoice.when(sfn.Condition.string_equals(\"$.color\", \"RED\"), handle_red_item)\nchoice.otherwise(handle_other_item_color)\n\n# Use .afterwards() to join all possible paths back together and continue\nship_the_item = sfn.Pass(self, \"ShipTheItem\")\nchoice.afterwards().next(ship_the_item)\n```\n\nIf your `Choice` doesn't have an `otherwise()` and none of the conditions match\nthe JSON state, a `NoChoiceMatched` error will be thrown. Wrap the state machine\nin a `Parallel` state if you want to catch and recover from this.\n\n#### Available Conditions\n\nsee [step function comparison operators](https://docs.aws.amazon.com/step-functions/latest/dg/amazon-states-language-choice-state.html#amazon-states-language-choice-state-rules)\n\n* `Condition.isPresent` - matches if a json path is present\n* `Condition.isNotPresent` - matches if a json path is not present\n* `Condition.isString` - matches if a json path contains a string\n* `Condition.isNotString` - matches if a json path is not a string\n* `Condition.isNumeric` - matches if a json path is numeric\n* `Condition.isNotNumeric` - matches if a json path is not numeric\n* `Condition.isBoolean` - matches if a json path is boolean\n* `Condition.isNotBoolean` - matches if a json path is not boolean\n* `Condition.isTimestamp` - matches if a json path is a timestamp\n* `Condition.isNotTimestamp` - matches if a json path is not a timestamp\n* `Condition.isNotNull` - matches if a json path is not null\n* `Condition.isNull` - matches if a json path is null\n* `Condition.booleanEquals` - matches if a boolean field has a given value\n* `Condition.booleanEqualsJsonPath` - matches if a boolean field equals a value in a given mapping path\n* `Condition.stringEqualsJsonPath` - matches if a string field equals a given mapping path\n* `Condition.stringEquals` - matches if a field equals a string value\n* `Condition.stringLessThan` - matches if a string field sorts before a given value\n* `Condition.stringLessThanJsonPath` - matches if a string field sorts before a value at given mapping path\n* `Condition.stringLessThanEquals` - matches if a string field sorts equal to or before a given value\n* `Condition.stringLessThanEqualsJsonPath` - matches if a string field sorts equal to or before a given mapping\n* `Condition.stringGreaterThan` - matches if a string field sorts after a given value\n* `Condition.stringGreaterThanJsonPath` - matches if a string field sorts after a value at a given mapping path\n* `Condition.stringGreaterThanEqualsJsonPath` - matches if a string field sorts after or equal to value at a given mapping path\n* `Condition.stringGreaterThanEquals` - matches if a string field sorts after or equal to a given value\n* `Condition.numberEquals` - matches if a numeric field has the given value\n* `Condition.numberEqualsJsonPath` - matches if a numeric field has the value in a given mapping path\n* `Condition.numberLessThan` - matches if a numeric field is less than the given value\n* `Condition.numberLessThanJsonPath` - matches if a numeric field is less than the value at the given mapping path\n* `Condition.numberLessThanEquals` - matches if a numeric field is less than or equal to the given value\n* `Condition.numberLessThanEqualsJsonPath` - matches if a numeric field is less than or equal to the numeric value at given mapping path\n* `Condition.numberGreaterThan` - matches if a numeric field is greater than the given value\n* `Condition.numberGreaterThanJsonPath` - matches if a numeric field is greater than the value at a given mapping path\n* `Condition.numberGreaterThanEquals` - matches if a numeric field is greater than or equal to the given value\n* `Condition.numberGreaterThanEqualsJsonPath` - matches if a numeric field is greater than or equal to the value at a given mapping path\n* `Condition.timestampEquals` - matches if a timestamp field is the same time as the given timestamp\n* `Condition.timestampEqualsJsonPath` - matches if a timestamp field is the same time as the timestamp at a given mapping path\n* `Condition.timestampLessThan` - matches if a timestamp field is before the given timestamp\n* `Condition.timestampLessThanJsonPath` - matches if a timestamp field is before the timestamp at a given mapping path\n* `Condition.timestampLessThanEquals` - matches if a timestamp field is before or equal to the given timestamp\n* `Condition.timestampLessThanEqualsJsonPath` - matches if a timestamp field is before or equal to the timestamp at a given mapping path\n* `Condition.timestampGreaterThan` - matches if a timestamp field is after the timestamp at a given mapping path\n* `Condition.timestampGreaterThanJsonPath` - matches if a timestamp field is after the timestamp at a given mapping path\n* `Condition.timestampGreaterThanEquals` - matches if a timestamp field is after or equal to the given timestamp\n* `Condition.timestampGreaterThanEqualsJsonPath` - matches if a timestamp field is after or equal to the timestamp at a given mapping path\n* `Condition.stringMatches` - matches if a field matches a string pattern that can contain a wild card (*) e.g: log-*.txt or *LATEST*. No other characters other than \"*\" have any special meaning - * can be escaped: \\\\*\n\n### Parallel\n\nA `Parallel` state executes one or more subworkflows in parallel. It can also\nbe used to catch and recover from errors in subworkflows.\n\n```python\nparallel = sfn.Parallel(self, \"Do the work in parallel\")\n\n# Add branches to be executed in parallel\nship_item = sfn.Pass(self, \"ShipItem\")\nsend_invoice = sfn.Pass(self, \"SendInvoice\")\nrestock = sfn.Pass(self, \"Restock\")\nparallel.branch(ship_item)\nparallel.branch(send_invoice)\nparallel.branch(restock)\n\n# Retry the whole workflow if something goes wrong\nparallel.add_retry(max_attempts=1)\n\n# How to recover from errors\nsend_failure_notification = sfn.Pass(self, \"SendFailureNotification\")\nparallel.add_catch(send_failure_notification)\n\n# What to do in case everything succeeded\nclose_order = sfn.Pass(self, \"CloseOrder\")\nparallel.next(close_order)\n```\n\n### Succeed\n\nReaching a `Succeed` state terminates the state machine execution with a\nsuccessful status.\n\n```python\nsuccess = sfn.Succeed(self, \"We did it!\")\n```\n\n### Fail\n\nReaching a `Fail` state terminates the state machine execution with a\nfailure status. The fail state should report the reason for the failure.\nFailures can be caught by encompassing `Parallel` states.\n\n```python\nsuccess = sfn.Fail(self, \"Fail\",\n error=\"WorkflowFailure\",\n cause=\"Something went wrong\"\n)\n```\n\n### Map\n\nA `Map` state can be used to run a set of steps for each element of an input array.\nA `Map` state will execute the same steps for multiple entries of an array in the state input.\n\nWhile the `Parallel` state executes multiple branches of steps using the same input, a `Map` state will\nexecute the same steps for multiple entries of an array in the state input.\n\n```python\nmap = sfn.Map(self, \"Map State\",\n max_concurrency=1,\n items_path=sfn.JsonPath.string_at(\"$.inputForMap\")\n)\nmap.iterator(sfn.Pass(self, \"Pass State\"))\n```\n\n### Custom State\n\nIt's possible that the high-level constructs for the states or `stepfunctions-tasks` do not have\nthe states or service integrations you are looking for. The primary reasons for this lack of\nfunctionality are:\n\n* A [service integration](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-service-integrations.html) is available through Amazon States Langauge, but not available as construct\n classes in the CDK.\n* The state or state properties are available through Step Functions, but are not configurable\n through constructs\n\nIf a feature is not available, a `CustomState` can be used to supply any Amazon States Language\nJSON-based object as the state definition.\n\n[Code Snippets](https://docs.aws.amazon.com/step-functions/latest/dg/tutorial-code-snippet.html#tutorial-code-snippet-1) are available and can be plugged in as the state definition.\n\nCustom states can be chained together with any of the other states to create your state machine\ndefinition. You will also need to provide any permissions that are required to the `role` that\nthe State Machine uses.\n\nThe following example uses the `DynamoDB` service integration to insert data into a DynamoDB table.\n\n```python\nimport aws_cdk.aws_dynamodb as dynamodb\n\n\n# create a table\ntable = dynamodb.Table(self, \"montable\",\n partition_key=dynamodb.Attribute(\n name=\"id\",\n type=dynamodb.AttributeType.STRING\n )\n)\n\nfinal_status = sfn.Pass(self, \"final step\")\n\n# States language JSON to put an item into DynamoDB\n# snippet generated from https://docs.aws.amazon.com/step-functions/latest/dg/tutorial-code-snippet.html#tutorial-code-snippet-1\nstate_json = {\n \"Type\": \"Task\",\n \"Resource\": \"arn:aws:states:::dynamodb:putItem\",\n \"Parameters\": {\n \"TableName\": table.table_name,\n \"Item\": {\n \"id\": {\n \"S\": \"MyEntry\"\n }\n }\n },\n \"ResultPath\": null\n}\n\n# custom state which represents a task to insert data into DynamoDB\ncustom = sfn.CustomState(self, \"my custom task\",\n state_json=state_json\n)\n\nchain = sfn.Chain.start(custom).next(final_status)\n\nsm = sfn.StateMachine(self, \"StateMachine\",\n definition=chain,\n timeout=Duration.seconds(30)\n)\n\n# don't forget permissions. You need to assign them\ntable.grant_write_data(sm)\n```\n\n## Task Chaining\n\nTo make defining work flows as convenient (and readable in a top-to-bottom way)\nas writing regular programs, it is possible to chain most methods invocations.\nIn particular, the `.next()` method can be repeated. The result of a series of\n`.next()` calls is called a **Chain**, and can be used when defining the jump\ntargets of `Choice.on` or `Parallel.branch`:\n\n```python\nstep1 = sfn.Pass(self, \"Step1\")\nstep2 = sfn.Pass(self, \"Step2\")\nstep3 = sfn.Pass(self, \"Step3\")\nstep4 = sfn.Pass(self, \"Step4\")\nstep5 = sfn.Pass(self, \"Step5\")\nstep6 = sfn.Pass(self, \"Step6\")\nstep7 = sfn.Pass(self, \"Step7\")\nstep8 = sfn.Pass(self, \"Step8\")\nstep9 = sfn.Pass(self, \"Step9\")\nstep10 = sfn.Pass(self, \"Step10\")\nchoice = sfn.Choice(self, \"Choice\")\ncondition1 = sfn.Condition.string_equals(\"$.status\", \"SUCCESS\")\nparallel = sfn.Parallel(self, \"Parallel\")\nfinish = sfn.Pass(self, \"Finish\")\n\ndefinition = step1.next(step2).next(choice.when(condition1, step3.next(step4).next(step5)).otherwise(step6).afterwards()).next(parallel.branch(step7.next(step8)).branch(step9.next(step10))).next(finish)\n\nsfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n```\n\nIf you don't like the visual look of starting a chain directly off the first\nstep, you can use `Chain.start`:\n\n```python\nstep1 = sfn.Pass(self, \"Step1\")\nstep2 = sfn.Pass(self, \"Step2\")\nstep3 = sfn.Pass(self, \"Step3\")\n\ndefinition = sfn.Chain.start(step1).next(step2).next(step3)\n```\n\n## State Machine Fragments\n\nIt is possible to define reusable (or abstracted) mini-state machines by\ndefining a construct that implements `IChainable`, which requires you to define\ntwo fields:\n\n* `startState: State`, representing the entry point into this state machine.\n* `endStates: INextable[]`, representing the (one or more) states that outgoing\n transitions will be added to if you chain onto the fragment.\n\nSince states will be named after their construct IDs, you may need to prefix the\nIDs of states if you plan to instantiate the same state machine fragment\nmultiples times (otherwise all states in every instantiation would have the same\nname).\n\nThe class `StateMachineFragment` contains some helper functions (like\n`prefixStates()`) to make it easier for you to do this. If you define your state\nmachine as a subclass of this, it will be convenient to use:\n\n```python\nfrom aws_cdk.core import Stack\nfrom constructs import Construct\nimport aws_cdk.aws_stepfunctions as sfn\n\nclass MyJob(sfn.StateMachineFragment):\n\n def __init__(self, parent, id, *, jobFlavor):\n super().__init__(parent, id)\n\n choice = sfn.Choice(self, \"Choice\").when(sfn.Condition.string_equals(\"$.branch\", \"left\"), sfn.Pass(self, \"Left Branch\")).when(sfn.Condition.string_equals(\"$.branch\", \"right\"), sfn.Pass(self, \"Right Branch\"))\n\n # ...\n\n self.start_state = choice\n self.end_states = choice.afterwards().end_states\n\nclass MyStack(Stack):\n def __init__(self, scope, id):\n super().__init__(scope, id)\n # Do 3 different variants of MyJob in parallel\n parallel = sfn.Parallel(self, \"All jobs\").branch(MyJob(self, \"Quick\", job_flavor=\"quick\").prefix_states()).branch(MyJob(self, \"Medium\", job_flavor=\"medium\").prefix_states()).branch(MyJob(self, \"Slow\", job_flavor=\"slow\").prefix_states())\n\n sfn.StateMachine(self, \"MyStateMachine\",\n definition=parallel\n )\n```\n\nA few utility functions are available to parse state machine fragments.\n\n* `State.findReachableStates`: Retrieve the list of states reachable from a given state.\n* `State.findReachableEndStates`: Retrieve the list of end or terminal states reachable from a given state.\n\n## Activity\n\n**Activities** represent work that is done on some non-Lambda worker pool. The\nStep Functions workflow will submit work to this Activity, and a worker pool\nthat you run yourself, probably on EC2, will pull jobs from the Activity and\nsubmit the results of individual jobs back.\n\nYou need the ARN to do so, so if you use Activities be sure to pass the Activity\nARN into your worker pool:\n\n```python\nactivity = sfn.Activity(self, \"Activity\")\n\n# Read this CloudFormation Output from your application and use it to poll for work on\n# the activity.\nCfnOutput(self, \"ActivityArn\", value=activity.activity_arn)\n```\n\n### Activity-Level Permissions\n\nGranting IAM permissions to an activity can be achieved by calling the `grant(principal, actions)` API:\n\n```python\nactivity = sfn.Activity(self, \"Activity\")\n\nrole = iam.Role(self, \"Role\",\n assumed_by=iam.ServicePrincipal(\"lambda.amazonaws.com\")\n)\n\nactivity.grant(role, \"states:SendTaskSuccess\")\n```\n\nThis will grant the IAM principal the specified actions onto the activity.\n\n## Metrics\n\n`Task` object expose various metrics on the execution of that particular task. For example,\nto create an alarm on a particular task failing:\n\n```python\n# task: sfn.Task\n\ncloudwatch.Alarm(self, \"TaskAlarm\",\n metric=task.metric_failed(),\n threshold=1,\n evaluation_periods=1\n)\n```\n\nThere are also metrics on the complete state machine:\n\n```python\n# state_machine: sfn.StateMachine\n\ncloudwatch.Alarm(self, \"StateMachineAlarm\",\n metric=state_machine.metric_failed(),\n threshold=1,\n evaluation_periods=1\n)\n```\n\nAnd there are metrics on the capacity of all state machines in your account:\n\n```python\ncloudwatch.Alarm(self, \"ThrottledAlarm\",\n metric=sfn.StateTransitionMetric.metric_throttled_events(),\n threshold=10,\n evaluation_periods=2\n)\n```\n\n## Error names\n\nStep Functions identifies errors in the Amazon States Language using case-sensitive strings, known as error names.\nThe Amazon States Language defines a set of built-in strings that name well-known errors, all beginning with the `States.` prefix.\n\n* `States.ALL` - A wildcard that matches any known error name.\n* `States.Runtime` - An execution failed due to some exception that could not be processed. Often these are caused by errors at runtime, such as attempting to apply InputPath or OutputPath on a null JSON payload. A `States.Runtime` error is not retriable, and will always cause the execution to fail. A retry or catch on `States.ALL` will NOT catch States.Runtime errors.\n* `States.DataLimitExceeded` - A States.DataLimitExceeded exception will be thrown for the following:\n\n * When the output of a connector is larger than payload size quota.\n * When the output of a state is larger than payload size quota.\n * When, after Parameters processing, the input of a state is larger than the payload size quota.\n * See [the AWS documentation](https://docs.aws.amazon.com/step-functions/latest/dg/limits-overview.html) to learn more about AWS Step Functions Quotas.\n* `States.HeartbeatTimeout` - A Task state failed to send a heartbeat for a period longer than the HeartbeatSeconds value.\n* `States.Timeout` - A Task state either ran longer than the TimeoutSeconds value, or failed to send a heartbeat for a period longer than the HeartbeatSeconds value.\n* `States.TaskFailed`- A Task state failed during the execution. When used in a retry or catch, `States.TaskFailed` acts as a wildcard that matches any known error name except for `States.Timeout`.\n\n## Logging\n\nEnable logging to CloudWatch by passing a logging configuration with a\ndestination LogGroup:\n\n```python\nimport aws_cdk.aws_logs as logs\n\n\nlog_group = logs.LogGroup(self, \"MyLogGroup\")\n\nsfn.StateMachine(self, \"MyStateMachine\",\n definition=sfn.Chain.start(sfn.Pass(self, \"Pass\")),\n logs=sfn.LogOptions(\n destination=log_group,\n level=sfn.LogLevel.ALL\n )\n)\n```\n\n## X-Ray tracing\n\nEnable X-Ray tracing for StateMachine:\n\n```python\nsfn.StateMachine(self, \"MyStateMachine\",\n definition=sfn.Chain.start(sfn.Pass(self, \"Pass\")),\n tracing_enabled=True\n)\n```\n\nSee [the AWS documentation](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-xray-tracing.html)\nto learn more about AWS Step Functions's X-Ray support.\n\n## State Machine Permission Grants\n\nIAM roles, users, or groups which need to be able to work with a State Machine should be granted IAM permissions.\n\nAny object that implements the `IGrantable` interface (has an associated principal) can be granted permissions by calling:\n\n* `stateMachine.grantStartExecution(principal)` - grants the principal the ability to execute the state machine\n* `stateMachine.grantRead(principal)` - grants the principal read access\n* `stateMachine.grantTaskResponse(principal)` - grants the principal the ability to send task tokens to the state machine\n* `stateMachine.grantExecution(principal, actions)` - grants the principal execution-level permissions for the IAM actions specified\n* `stateMachine.grant(principal, actions)` - grants the principal state-machine-level permissions for the IAM actions specified\n\n### Start Execution Permission\n\nGrant permission to start an execution of a state machine by calling the `grantStartExecution()` API.\n\n```python\n# definition: sfn.IChainable\nrole = iam.Role(self, \"Role\",\n assumed_by=iam.ServicePrincipal(\"lambda.amazonaws.com\")\n)\nstate_machine = sfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n\n# Give role permission to start execution of state machine\nstate_machine.grant_start_execution(role)\n```\n\nThe following permission is provided to a service principal by the `grantStartExecution()` API:\n\n* `states:StartExecution` - to state machine\n\n### Read Permissions\n\nGrant `read` access to a state machine by calling the `grantRead()` API.\n\n```python\n# definition: sfn.IChainable\nrole = iam.Role(self, \"Role\",\n assumed_by=iam.ServicePrincipal(\"lambda.amazonaws.com\")\n)\nstate_machine = sfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n\n# Give role read access to state machine\nstate_machine.grant_read(role)\n```\n\nThe following read permissions are provided to a service principal by the `grantRead()` API:\n\n* `states:ListExecutions` - to state machine\n* `states:ListStateMachines` - to state machine\n* `states:DescribeExecution` - to executions\n* `states:DescribeStateMachineForExecution` - to executions\n* `states:GetExecutionHistory` - to executions\n* `states:ListActivities` - to `*`\n* `states:DescribeStateMachine` - to `*`\n* `states:DescribeActivity` - to `*`\n\n### Task Response Permissions\n\nGrant permission to allow task responses to a state machine by calling the `grantTaskResponse()` API:\n\n```python\n# definition: sfn.IChainable\nrole = iam.Role(self, \"Role\",\n assumed_by=iam.ServicePrincipal(\"lambda.amazonaws.com\")\n)\nstate_machine = sfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n\n# Give role task response permissions to the state machine\nstate_machine.grant_task_response(role)\n```\n\nThe following read permissions are provided to a service principal by the `grantRead()` API:\n\n* `states:SendTaskSuccess` - to state machine\n* `states:SendTaskFailure` - to state machine\n* `states:SendTaskHeartbeat` - to state machine\n\n### Execution-level Permissions\n\nGrant execution-level permissions to a state machine by calling the `grantExecution()` API:\n\n```python\n# definition: sfn.IChainable\nrole = iam.Role(self, \"Role\",\n assumed_by=iam.ServicePrincipal(\"lambda.amazonaws.com\")\n)\nstate_machine = sfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n\n# Give role permission to get execution history of ALL executions for the state machine\nstate_machine.grant_execution(role, \"states:GetExecutionHistory\")\n```\n\n### Custom Permissions\n\nYou can add any set of permissions to a state machine by calling the `grant()` API.\n\n```python\n# definition: sfn.IChainable\nuser = iam.User(self, \"MyUser\")\nstate_machine = sfn.StateMachine(self, \"StateMachine\",\n definition=definition\n)\n\n# give user permission to send task success to the state machine\nstate_machine.grant(user, \"states:SendTaskSuccess\")\n```\n\n## Import\n\nAny Step Functions state machine that has been created outside the stack can be imported\ninto your CDK stack.\n\nState machines can be imported by their ARN via the `StateMachine.fromStateMachineArn()` API\n\n```python\napp = App()\nstack = Stack(app, \"MyStack\")\nsfn.StateMachine.from_state_machine_arn(stack, \"ImportedStateMachine\", \"arn:aws:states:us-east-1:123456789012:stateMachine:StateMachine2E01A3A5-N5TJppzoevKQ\")\n```\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "The CDK Construct Library for AWS::StepFunctions",
"version": "1.204.0",
"project_urls": {
"Homepage": "https://github.com/aws/aws-cdk",
"Source": "https://github.com/aws/aws-cdk.git"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "7f34dfe026664b2e7a96ec0291b87fb292db3ffe273fbd09ab47dfc2ea821acd",
"md5": "b949823eb560b64ce59e572069a02103",
"sha256": "9836cb35e0a19525b59815135c6bfa56350eddda78d0f09d73878f8fe1531133"
},
"downloads": -1,
"filename": "aws_cdk.aws_stepfunctions-1.204.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b949823eb560b64ce59e572069a02103",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "~=3.7",
"size": 365591,
"upload_time": "2023-06-19T21:01:40",
"upload_time_iso_8601": "2023-06-19T21:01:40.688531Z",
"url": "https://files.pythonhosted.org/packages/7f/34/dfe026664b2e7a96ec0291b87fb292db3ffe273fbd09ab47dfc2ea821acd/aws_cdk.aws_stepfunctions-1.204.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "16b515672f07e881d26e2cebaddd73c20e99476deac7bb1f94975079a1ec7e60",
"md5": "fc39ec41805596c81f547d520cd974ff",
"sha256": "bab48ffaaef6f33512e5408908b245cfa55709d50e30c8ab939fb3693b5d35a4"
},
"downloads": -1,
"filename": "aws-cdk.aws-stepfunctions-1.204.0.tar.gz",
"has_sig": false,
"md5_digest": "fc39ec41805596c81f547d520cd974ff",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "~=3.7",
"size": 373421,
"upload_time": "2023-06-19T21:07:48",
"upload_time_iso_8601": "2023-06-19T21:07:48.243743Z",
"url": "https://files.pythonhosted.org/packages/16/b5/15672f07e881d26e2cebaddd73c20e99476deac7bb1f94975079a1ec7e60/aws-cdk.aws-stepfunctions-1.204.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-06-19 21:07:48",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "aws",
"github_project": "aws-cdk",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aws-cdk.aws-stepfunctions"
}
JSON
