Skip to main content

Python object model for the Amazon States Language

Project description


Python Object Model for Amazon State Language

This package provides an object model for creating Step Functions


Simple Hello World Example

import py_asl

def hello_world():
    state_machine = py_asl.StateMachine(Comment="A simple minimal example of the States language", StartAt="Hello World")
    state = py_asl.TaskState("Hello World", Resource="arn:aws:lambda:us-east-1:123456789012:function:HelloWorld", End=True)
    hw = state_machine.dumps(indent=2)
    return hw


Example of creating a Parallel Task

import py_asl

def parallel_states():
    state1 = py_asl.TaskState("Hello World", Resource="arn:aws:lambda:us-east-1:123456789012:function:HelloWorld", End=True)
    state2 = py_asl.TaskState("Goodbye World", Resource="arn:aws:lambda:us-east-1:123456789012:function:GoodbyeWorld", End=True)
    parallel_state = state_model.ParallelState("Hello Goodbye", Branches=[state1, state2], Next="Foo")
    ps = parallel_state.dumps(indent=2)
    return ps


Arrays and Generated Templates

Template variables that where place holders for arrays were tricky, so I implemented a workaround which is probably not suitable for all cases...Let's say that you have to deploy a Step Function that has a task that invokes an ECS/Fargate task and you need to specify the VPC configuration. In this case you will need to supply parameters that include a list of subnets and a list of Security Groups.

Your code might look like this:

import py_asl

Parameters = {"Cluster": "${cluster_arn}",
              "TaskDefinition": "${task_definition_arn}",
              "LaunchType": "FARGATE",
              "NetworkConfiguration": {"AwsvpcConfiguration": {
                                                               "SecurityGroups": "[${security_groups}]",
                                                               "Subnets": "[${subnets}]"

task = py_asl.TaskState("Run Fargate",

state_machine = py_asl.StateMachine(Comment="Step Function to Test Invoking ECS/Fargate Task",
                                    StartAt="Run Fargate",

if __name__ == '__main__':

NOTE: The arrays for the security groups will be transformed into:

  "Comment": "Step Function to Test Invoking ECS/Fargate Task",
  "StartAt": "Run Fargate",
  "States": {
    "Run Fargate": {
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "End": true,
      "Parameters": {
        "Cluster": "${cluster_arn}",
        "TaskDefinition": "${task_definition_arn}",
        "LaunchType": "FARGATE",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "SecurityGroups": ${security_groups},
            "Subnets": ${subnets}
      "Type": "Task"

It is expected that your Terraform code supplies and array. For example,

variable "security_groups" {
    type = "list"
    default = ["ab-12345", "bc-56788"]

variable "subnets" {
    type = "list"
    default = ["itsy", "bitsy"]

When replacing values in the template you can use:

data "template_file" "step_function" {
    template = "${file("${path.module}/step-function.json.tpl")}"
    vars = {
        security_groups = "${jsonencode(var.cluster_security_groups)}"
        subnets = "${jsonencode(var.cluster_subnets)}"

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

py_asl-0.1.4.tar.gz (4.5 kB view hashes)

Uploaded source

Built Distribution

py_asl-0.1.4-py2.py3-none-any.whl (4.7 kB view hashes)

Uploaded py2 py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page