pystepfunctions

Create AWS Stepfunction asl.json files (state machine definition files) using python.
Pre-made tasks that are easy to use.
Test dataflow through the stepfunction using the same python code.
Visualise the stepfunction using pyvis for easier dubugging without the use of the AWS console.

Docs

https://mrhopko.github.io/pystepfunction/pystepfunction/

Installation

pip3 install pystepfunction

Complete example

from logging import getLogger
from pystepfunction.tasks import (
    LambdaTask,
    GlueTask,
    PassTask,
    SucceedTask,
    FailTask,
)
from pystepfunction.branch import Branch
from pystepfunction.viz import BranchViz
from pystepfunction.state import StateMachine
from pystepfunction.errors import ERROR_STATE_ALL

logger = getLogger(__name__)

# create a lambda task
lambda_task = (
    LambdaTask("lambda_task", function_arn="aws::my-lambda-arn")
    .with_retry(error_equals=[ERROR_STATE_ALL], interval_seconds=1, max_attempts=3)
    .with_catcher(
        error=[ERROR_STATE_ALL],
        task=FailTask(
            "lambda_task_fail", cause="lambda task failed", error="MyLambdaError"
        ),
    )
    .with_resource_result({"Payload": {"Result": "LambdaResult"}})
    .with_output(
        result_path="$.LambdaTaskResult",
        result_selector={"SelectThis.$": "$.Payload.Result"},
    )
)

# create a glue task
glue_task = (
    GlueTask(
        "glue_task",
        job_name="my-glue-job-name",
        job_args={"job_input_arg.$": "$.LambdaTaskResult.SelectThis"},
    )
    .with_catcher(
        error=[ERROR_STATE_ALL],
        task=FailTask(
            "glue_task_fail", cause="glue task failed", error="MyGlueError"
        ),
    )
    .with_resource_result({"JobResult": "GlueResult"})
    .with_output(result_path="$.GlueTaskResult")
)

# chain them together and create a branch
lambda_task = lambda_task >> glue_task >> SucceedTask("succeeded")
branch = Branch(comment="Lambda and Glue", start_task=lambda_task)

# view the asl
print(branch)

# asl as a dict
asl = branch.to_asl()

# write the asl to a file
branch.write_asl("my_asl_file.asl.json")

# visualise the asl
BranchViz(branch).show()

# create a state machine
sm = StateMachine(state={"Input1": "Input1Value"}, logger=logger)
sm.apply_branch(branch)
sm.show_logs()

Metadata

Release files for pystepfunction 1.1.131

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for pystepfunction 1.1.131
File	Size	Uploaded
pystepfunction-1.1.131.tar.gz	21.5 kB	Oct 3, 2023	Details

Built distribution (wheel)

Table of built distributions (wheels) for pystepfunction 1.1.131
File	Interpreter	ABI	Platform	Reset
pystepfunction-1.1.131-py3-none-any.whl 20.8 kB Oct 3, 2023	Python 3	none	any	Details

Total release size: 42.3 kB

Release files / pystepfunction-1.1.131.tar.gz

Download URL	pystepfunction-1.1.131.tar.gz
Size	21.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`87e0260473c4efadf9980439e268c6d58476e8eb9dc43ba6ddb549c1fabdb4c1`
BLAKE2b-256 checksum How to use checksums	`39e183daa455335500999975397ff22d4fc87c2ba7428a7249514544c3e1fe99`
Upload date	Oct 3, 2023
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/4.0.2 CPython/3.11.6`

Release files / pystepfunction-1.1.131-py3-none-any.whl

Download URL	pystepfunction-1.1.131-py3-none-any.whl
Size	20.8 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`6a3c009a1dd740f220f3199826c6c3ca627b7797fd5eeda8761c63a1289cdab3`
BLAKE2b-256 checksum How to use checksums	`fe95ac83a32ba34b1500cd798ca2cdb91e9f253290f7024554772fe0f1d40a5a`
Upload date	Oct 3, 2023
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/4.0.2 CPython/3.11.6`