{
"cells": [
{
"cell_type": "markdown",
"source": [
"# Logic Flow\n",
"\n",
"The purpose of StepFunction is to orchestrate many tasks working together, and transition from one to another conditionally. Overall, there are four generic flow patterns, and all other patterns are just combination of them:\n",
"\n",
"- **Simple straight chain**\n",
"- **Parallel execute sub-workflow**\n",
"- **Execute the same sub-workflow many times with different parameters**\n",
"- **Choice branch based on condition**\n",
"\n",
"In this tutorial, we would like to use 'Pass' state to simulate the 'Task' state, so we can focus on the logic flow orchestration."
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%% md\n"
}
}
},
{
"cell_type": "code",
"execution_count": 31,
"outputs": [],
"source": [
"import aws_stepfunction as sfn\n",
"from rich import print"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "markdown",
"source": [
"## Simple Straight Chain\n",
"\n",
"Let's create the long-chain workflow in ``aws_stepfunction``.\n",
"\n",
""
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%% md\n"
}
}
},
{
"cell_type": "code",
"execution_count": 18,
"outputs": [],
"source": [
"workflow = sfn.Workflow()\n",
"\n",
"lambda_1 = sfn.Pass(id=\"Lambda 1\")\n",
"lambda_2 = sfn.Pass(id=\"Lambda 2\")\n",
"lambda_3 = sfn.Pass(id=\"Lambda 3\")"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "code",
"execution_count": 32,
"outputs": [
{
"data": {
"text/plain": "\u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Lambda 1'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m,\n \u001B[32m'Result'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'action'\u001B[0m: \u001B[32m'run lambda job'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Next'\u001B[0m: \u001B[32m'Choice-by-Input Parameter Handling'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Choice-by-Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Choice'\u001B[0m,\n \u001B[32m'Choices'\u001B[0m: \u001B[1m[\u001B[0m\n \u001B[1m{\u001B[0m\u001B[32m'Variable'\u001B[0m: \u001B[32m'$.action'\u001B[0m, \u001B[32m'StringEquals'\u001B[0m: \u001B[32m'run lambda job'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Run Lambda Job'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[1m{\u001B[0m\u001B[32m'Variable'\u001B[0m: \u001B[32m'$.action'\u001B[0m, \u001B[32m'StringEquals'\u001B[0m: \u001B[32m'run glue job'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Run Glue Job'\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m]\u001B[0m,\n \u001B[32m'Default'\u001B[0m: \u001B[32m'Fail'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Run Lambda Job'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Send Notification'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Run Glue Job'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Send Notification'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Fail'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Fail'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Send Notification'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Lambda 1'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Wait-after-Lambda 1'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Wait-after-Lambda 1'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Wait'\u001B[0m, \u001B[32m'Seconds'\u001B[0m: \u001B[1;36m3\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Lambda 2'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Lambda 2'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Wait-077b428'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Wait-077b428'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Wait'\u001B[0m, \u001B[32m'Seconds'\u001B[0m: \u001B[1;36m3\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Lambda 3'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Lambda 3'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n\u001B[1m}\u001B[0m\n",
"text/html": "{\n 'StartAt': 'Lambda 1',\n 'States': {\n 'Input Parameter Handling': {\n 'Type': 'Pass',\n 'Result': {'action': 'run lambda job'},\n 'Next': 'Choice-by-Input Parameter Handling'\n },\n 'Choice-by-Input Parameter Handling': {\n 'Type': 'Choice',\n 'Choices': [\n {'Variable': '$.action', 'StringEquals': 'run lambda job', 'Next': 'Run Lambda Job'},\n {'Variable': '$.action', 'StringEquals': 'run glue job', 'Next': 'Run Glue Job'}\n ],\n 'Default': 'Fail'\n },\n 'Run Lambda Job': {'Type': 'Pass', 'Next': 'Send Notification'},\n 'Run Glue Job': {'Type': 'Pass', 'Next': 'Send Notification'},\n 'Fail': {'Type': 'Fail'},\n 'Send Notification': {'Type': 'Pass', 'End': True},\n 'Lambda 1': {'Type': 'Pass', 'Next': 'Wait-after-Lambda 1'},\n 'Wait-after-Lambda 1': {'Type': 'Wait', 'Seconds': 3, 'Next': 'Lambda 2'},\n 'Lambda 2': {'Type': 'Pass', 'Next': 'Wait-077b428'},\n 'Wait-077b428': {'Type': 'Wait', 'Seconds': 3, 'Next': 'Lambda 3'},\n 'Lambda 3': {'Type': 'Pass', 'End': True}\n }\n}\n\n"
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"# use bracket to enclose your logic flow\n",
"# so you don't need to add annoying \"\\\" at the end of each line\n",
"(\n",
" workflow.start_from(lambda_1)\n",
" .wait(seconds=3) # you don't have to explicitly define a 'Wait' State\n",
" .next_then(lambda_2)\n",
" .next_then(sfn.Wait(seconds=3)) # you still can if you want\n",
" .next_then(lambda_3)\n",
" .end()\n",
")\n",
"\n",
"print(workflow.serialize())"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "markdown",
"source": [
"## Parallel execute sub-workflow\n",
"\n",
"From one point, we execute multiple sub-workflow concurrently and then merge back and continue.\n",
"\n",
""
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%% md\n"
}
}
},
{
"cell_type": "code",
"execution_count": 36,
"outputs": [],
"source": [
"workflow = sfn.Workflow()\n",
"\n",
"input_parameter_handling = sfn.Pass(id=\"Input Parameter Handling\")\n",
"subflow1_step1 = sfn.Pass(id=\"Subflow 1 - Step 1\")\n",
"subflow1_step2 = sfn.Pass(id=\"Subflow 1 - Step 2\")\n",
"subflow2_step1 = sfn.Pass(id=\"Subflow 2 - Step 1\")\n",
"subflow2_step2 = sfn.Pass(id=\"Subflow 2 - Step 2\")\n",
"clean_up = sfn.Pass(id=\"Clean Up\")"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "code",
"execution_count": 37,
"outputs": [
{
"data": {
"text/plain": "\u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Input Parameter Handling'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Parallel-after-Input Parameter Handling'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Parallel-after-Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Parallel'\u001B[0m,\n \u001B[32m'Branches'\u001B[0m: \u001B[1m[\u001B[0m\n \u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Subflow 1 - Step 1'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Subflow 1 - Step 1'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Subflow 1 - Step 2'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Subflow 1 - Step 2'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Subflow 2 - Step 1'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Subflow 2 - Step 1'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Subflow 2 - Step 2'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Subflow 2 - Step 2'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n \u001B[1m]\u001B[0m,\n \u001B[32m'Next'\u001B[0m: \u001B[32m'Clean Up'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Clean Up'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n\u001B[1m}\u001B[0m\n",
"text/html": "{\n 'StartAt': 'Input Parameter Handling',\n 'States': {\n 'Input Parameter Handling': {'Type': 'Pass', 'Next': 'Parallel-after-Input Parameter Handling'},\n 'Parallel-after-Input Parameter Handling': {\n 'Type': 'Parallel',\n 'Branches': [\n {\n 'StartAt': 'Subflow 1 - Step 1',\n 'States': {\n 'Subflow 1 - Step 1': {'Type': 'Pass', 'Next': 'Subflow 1 - Step 2'},\n 'Subflow 1 - Step 2': {'Type': 'Pass', 'End': True}\n }\n },\n {\n 'StartAt': 'Subflow 2 - Step 1',\n 'States': {\n 'Subflow 2 - Step 1': {'Type': 'Pass', 'Next': 'Subflow 2 - Step 2'},\n 'Subflow 2 - Step 2': {'Type': 'Pass', 'End': True}\n }\n }\n ],\n 'Next': 'Clean Up'\n },\n 'Clean Up': {'Type': 'Pass', 'End': True}\n }\n}\n\n"
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"(\n",
" workflow.start_from(input_parameter_handling)\n",
" .parallel([\n",
" ( # create a subflow in parallel\n",
" workflow.subflow_from(subflow1_step1)\n",
" .next_then(subflow1_step2)\n",
" .end() # each subflow in parallel has to end\n",
" ),\n",
" (\n",
" workflow.subflow_from(subflow2_step1)\n",
" .next_then(subflow2_step2)\n",
" .end()\n",
" ),\n",
" ])\n",
" .next_then(clean_up)\n",
" .end()\n",
")\n",
"\n",
"print(workflow.serialize())"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "markdown",
"source": [
"## Execute the same sub-workflow many times with different parameters\n",
"\n",
"We execute multiple sub-workflow with same logic but different input arguments. For example, we may have a data processing pipeline that can handle different data sources.\n",
"\n",
""
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%% md\n"
}
}
},
{
"cell_type": "code",
"execution_count": 40,
"outputs": [],
"source": [
"workflow = sfn.Workflow()\n",
"\n",
"input_parameter_handling = sfn.Pass(\n",
" id=\"Input Parameter Handling\",\n",
" result={\"dataset_name_list\": [\"customers\", \"sales\", \"orders\"]}\n",
")\n",
"read_data = sfn.Pass(id=\"Read Data\")\n",
"process_data = sfn.Pass(id=\"Process Data\")\n",
"write_data = sfn.Pass(id=\"Write Data\")\n",
"clean_up = sfn.Pass(id=\"Clean Up\")"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "code",
"execution_count": 41,
"outputs": [
{
"data": {
"text/plain": "\u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Input Parameter Handling'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m,\n \u001B[32m'Result'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'dataset_name_list'\u001B[0m: \u001B[1m[\u001B[0m\u001B[32m'customers'\u001B[0m, \u001B[32m'sales'\u001B[0m, \u001B[32m'orders'\u001B[0m\u001B[1m]\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Next'\u001B[0m: \u001B[32m'Map-after-Input Parameter Handling'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Map-after-Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Map'\u001B[0m,\n \u001B[32m'Iterator'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Read Data'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Read Data'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Process Data'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Process Data'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Write Data'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Write Data'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'ItemsPath'\u001B[0m: \u001B[32m'$.dataset_name_list'\u001B[0m,\n \u001B[32m'Next'\u001B[0m: \u001B[32m'Clean Up'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Clean Up'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n\u001B[1m}\u001B[0m\n",
"text/html": "{\n 'StartAt': 'Input Parameter Handling',\n 'States': {\n 'Input Parameter Handling': {\n 'Type': 'Pass',\n 'Result': {'dataset_name_list': ['customers', 'sales', 'orders']},\n 'Next': 'Map-after-Input Parameter Handling'\n },\n 'Map-after-Input Parameter Handling': {\n 'Type': 'Map',\n 'Iterator': {\n 'StartAt': 'Read Data',\n 'States': {\n 'Read Data': {'Type': 'Pass', 'Next': 'Process Data'},\n 'Process Data': {'Type': 'Pass', 'Next': 'Write Data'},\n 'Write Data': {'Type': 'Pass', 'End': True}\n }\n },\n 'ItemsPath': '$.dataset_name_list',\n 'Next': 'Clean Up'\n },\n 'Clean Up': {'Type': 'Pass', 'End': True}\n }\n}\n\n"
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"(\n",
" workflow.start_from(input_parameter_handling)\n",
" .map(\n",
" ( # create a subflow for mapping\n",
" workflow.subflow_from(read_data)\n",
" .next_then(process_data)\n",
" .next_then(write_data)\n",
" .end()\n",
" ),\n",
" items_path=\"$.dataset_name_list\", # each item in the list is the input argument in the subflow\n",
" )\n",
" .next_then(clean_up)\n",
" .end()\n",
")\n",
"\n",
"print(workflow.serialize())"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "markdown",
"source": [
"## Choice branch based on condition\n",
"\n",
"We jump to different branch based on the condition test. Each branch is just another start with a state, NOT a sub-workflow.\n",
"\n",
""
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%% md\n"
}
}
},
{
"cell_type": "code",
"execution_count": 45,
"outputs": [],
"source": [
"workflow = sfn.Workflow()\n",
"\n",
"input_parameter_handling = sfn.Pass(\n",
" id=\"Input Parameter Handling\",\n",
" result={\"action\": \"run lambda job\"},\n",
")\n",
"run_lambda_job = sfn.Pass(id=\"Run Lambda Job\")\n",
"run_glue_job = sfn.Pass(id=\"Run Glue Job\")\n",
"fail = sfn.Fail(id=\"Fail\")\n",
"send_notification = sfn.Pass(id=\"Send Notification\")"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "code",
"execution_count": 46,
"outputs": [
{
"data": {
"text/plain": "\u001B[1m{\u001B[0m\n \u001B[32m'StartAt'\u001B[0m: \u001B[32m'Input Parameter Handling'\u001B[0m,\n \u001B[32m'States'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m,\n \u001B[32m'Result'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'action'\u001B[0m: \u001B[32m'run lambda job'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Next'\u001B[0m: \u001B[32m'Choice-by-Input Parameter Handling'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Choice-by-Input Parameter Handling'\u001B[0m: \u001B[1m{\u001B[0m\n \u001B[32m'Type'\u001B[0m: \u001B[32m'Choice'\u001B[0m,\n \u001B[32m'Choices'\u001B[0m: \u001B[1m[\u001B[0m\n \u001B[1m{\u001B[0m\u001B[32m'Variable'\u001B[0m: \u001B[32m'$.action'\u001B[0m, \u001B[32m'StringEquals'\u001B[0m: \u001B[32m'run lambda job'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Run Lambda Job'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[1m{\u001B[0m\u001B[32m'Variable'\u001B[0m: \u001B[32m'$.action'\u001B[0m, \u001B[32m'StringEquals'\u001B[0m: \u001B[32m'run glue job'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Run Glue Job'\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m]\u001B[0m,\n \u001B[32m'Default'\u001B[0m: \u001B[32m'Fail'\u001B[0m\n \u001B[1m}\u001B[0m,\n \u001B[32m'Run Lambda Job'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Send Notification'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Run Glue Job'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'Next'\u001B[0m: \u001B[32m'Send Notification'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Fail'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Fail'\u001B[0m\u001B[1m}\u001B[0m,\n \u001B[32m'Send Notification'\u001B[0m: \u001B[1m{\u001B[0m\u001B[32m'Type'\u001B[0m: \u001B[32m'Pass'\u001B[0m, \u001B[32m'End'\u001B[0m: \u001B[3;92mTrue\u001B[0m\u001B[1m}\u001B[0m\n \u001B[1m}\u001B[0m\n\u001B[1m}\u001B[0m\n",
"text/html": "{\n 'StartAt': 'Input Parameter Handling',\n 'States': {\n 'Input Parameter Handling': {\n 'Type': 'Pass',\n 'Result': {'action': 'run lambda job'},\n 'Next': 'Choice-by-Input Parameter Handling'\n },\n 'Choice-by-Input Parameter Handling': {\n 'Type': 'Choice',\n 'Choices': [\n {'Variable': '$.action', 'StringEquals': 'run lambda job', 'Next': 'Run Lambda Job'},\n {'Variable': '$.action', 'StringEquals': 'run glue job', 'Next': 'Run Glue Job'}\n ],\n 'Default': 'Fail'\n },\n 'Run Lambda Job': {'Type': 'Pass', 'Next': 'Send Notification'},\n 'Run Glue Job': {'Type': 'Pass', 'Next': 'Send Notification'},\n 'Fail': {'Type': 'Fail'},\n 'Send Notification': {'Type': 'Pass', 'End': True}\n }\n}\n\n"
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"(\n",
" workflow.start_from(input_parameter_handling)\n",
" .choice(\n",
" [\n",
" # as a human, I cannot remember all the data test expression\n",
" # the IDE will automatically tell me the correct syntax\n",
" sfn.Var(\"$.action\").string_equals(\"run lambda job\").next_then(run_lambda_job),\n",
" sfn.Var(\"$.action\").string_equals(\"run glue job\").next_then(run_glue_job),\n",
" ],\n",
" default=fail, # set the default state\n",
" )\n",
")\n",
"\n",
"workflow.continue_from(run_lambda_job).next_then(send_notification)\n",
"workflow.continue_from(run_glue_job).next_then(send_notification).end()\n",
"\n",
"print(workflow.serialize())"
],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
},
{
"cell_type": "code",
"execution_count": 44,
"outputs": [],
"source": [],
"metadata": {
"collapsed": false,
"pycharm": {
"name": "#%%\n"
}
}
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 2
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython2",
"version": "2.7.6"
}
},
"nbformat": 4,
"nbformat_minor": 0
}