Sceptre state machine definition resolver

final_cfn_template.json

{
    "Outputs": {
        "Arn": {
            "Value": {
                "Ref": "sfnname"
            }
        }
    },
    "Parameters": {
        "RoleArn": {
            "Type": "String"
        },
        "StateMachineName": {
            "Type": "String"
        }
    },
    "Resources": {
        "sfnname": {
            "Properties": {
                "DefinitionString": "{\"Comment\": \"Invoke Lambda every minute\", \"StartAt\": \"Iterator\", \"States\": {\"Iterator\": {\"Type\": \"Task\", \"Resource\": \"arn:aws:lambda:us-east-1:123456789012:function:FUNCTION_NAME\", \"ResultPath\": \"$.iterator\", \"Next\": \"Done\", \"Retry\": [{\"ErrorEquals\": [\"InvalidInstanceId\"], \"IntervalSeconds\": 60, \"MaxAttempts\": 10, \"BackoffRate\": 1}]}, \"Done\": {\"Type\": \"Pass\", \"End\": true}}}",
                "RoleArn": {
                    "Ref": "RoleArn"
                },
                "StateMachineName": {
                    "Ref": "StateMachineName"
                }
            },
            "Type": "AWS::StepFunctions::StateMachine"
        }
    }
}

stack_config.yaml

state_machines:
  sfn1:
    Comment: Invoke Lambda every minute
    StartAt: Iterator
    States:
      Iterator:
        Type: Task
        Resource: "arn:aws:lambda:us-east-1:123456789012:function:FUNCTION_NAME"
        ResultPath: "$.iterator"
        Next: Done
        Retry:
        - ErrorEquals:
          - InvalidInstanceId
          IntervalSeconds: 60
          MaxAttempts: 10
          BackoffRate: 1
      Done:
        Type: Pass
        End: true

template_path: state_machine.py
sceptre_user_data:
  StateMachineName: sfn-name
  RoleArn: "arn:aws:iam::123456789012:role/role-name"
  DefinitionString: !state_machine_json sfn1

stm_json.py

import json
import logging
from sceptre.resolvers import Resolver, ResolvableProperty
from functools import reduce


class StateMachineJson(Resolver):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger(f"{self.logger.name}.{self.__class__.__name__}")

    def resolve(self):
        if 'state_machines' not in self.stack_config:
            raise KeyError('Stack config must contain "state_machines:" section.')
        values = recursive_get(self.stack_config['state_machines'], *self.argument.split('.'))
        resolved_values = ResolvableProperty('state_machine_json').resolve_values(values)
        return json.dumps(resolved_values)


def recursive_get(d: dict, *keys):
    """
    Recursively iterate nested dict and get value

    Parameters
    ----------
    d
        dict to be iterated through
    keys
        Keys to iterate through.

    Returns
    -------
        Key value or empty dict if not found.

    Examples
    --------
    >>> d = {'a':{'b':'c'}}
    >>> recursive_get(d, 'a')
    {'b': 'c'}
    >>> recursive_get(d, 'a', 'b')
    'c'
    >>> recursive_get(d, 'a', 'c')
    {}

    """
    return reduce(lambda c, key: c.get(key, {}), keys, d)