| Name | pystepfunction JSON |
| Version |
1.1.131
JSON |
| download |
| home_page | |
| Summary | Create AWS Step Functions asl.json with Python |
| upload_time | 2023-10-03 14:42:05 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.10 |
| license | |
| keywords |
aws
stepfunctions
step functions
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 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
```python
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()
```
Raw data
{
"_id": null,
"home_page": "",
"name": "pystepfunction",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "",
"keywords": "aws,stepfunctions,step functions",
"author": "",
"author_email": "Chris Hopkinson <therealchrishopkinson@hotmail.com>",
"download_url": "https://files.pythonhosted.org/packages/39/e1/83daa455335500999975397ff22d4fc87c2ba7428a7249514544c3e1fe99/pystepfunction-1.1.131.tar.gz",
"platform": null,
"description": "# pystepfunctions\n\nCreate AWS Stepfunction asl.json files (state machine definition files) using python. \nPre-made tasks that are easy to use. \nTest dataflow through the stepfunction using the same python code. \nVisualise the stepfunction using pyvis for easier dubugging without the use of the AWS console. \n\n## Docs\n\nhttps://mrhopko.github.io/pystepfunction/pystepfunction/\n\n## Installation\n```\npip3 install pystepfunction\n```\n\n## Complete example\n```python\nfrom logging import getLogger\nfrom pystepfunction.tasks import (\n LambdaTask,\n GlueTask,\n PassTask,\n SucceedTask,\n FailTask,\n)\nfrom pystepfunction.branch import Branch\nfrom pystepfunction.viz import BranchViz\nfrom pystepfunction.state import StateMachine\nfrom pystepfunction.errors import ERROR_STATE_ALL\n\nlogger = getLogger(__name__)\n\n# create a lambda task\nlambda_task = (\n LambdaTask(\"lambda_task\", function_arn=\"aws::my-lambda-arn\")\n .with_retry(error_equals=[ERROR_STATE_ALL], interval_seconds=1, max_attempts=3)\n .with_catcher(\n error=[ERROR_STATE_ALL],\n task=FailTask(\n \"lambda_task_fail\", cause=\"lambda task failed\", error=\"MyLambdaError\"\n ),\n )\n .with_resource_result({\"Payload\": {\"Result\": \"LambdaResult\"}})\n .with_output(\n result_path=\"$.LambdaTaskResult\",\n result_selector={\"SelectThis.$\": \"$.Payload.Result\"},\n )\n)\n\n# create a glue task\nglue_task = (\n GlueTask(\n \"glue_task\",\n job_name=\"my-glue-job-name\",\n job_args={\"job_input_arg.$\": \"$.LambdaTaskResult.SelectThis\"},\n )\n .with_catcher(\n error=[ERROR_STATE_ALL],\n task=FailTask(\n \"glue_task_fail\", cause=\"glue task failed\", error=\"MyGlueError\"\n ),\n )\n .with_resource_result({\"JobResult\": \"GlueResult\"})\n .with_output(result_path=\"$.GlueTaskResult\")\n)\n\n# chain them together and create a branch\nlambda_task = lambda_task >> glue_task >> SucceedTask(\"succeeded\")\nbranch = Branch(comment=\"Lambda and Glue\", start_task=lambda_task)\n\n# view the asl\nprint(branch)\n\n# asl as a dict\nasl = branch.to_asl()\n\n# write the asl to a file\nbranch.write_asl(\"my_asl_file.asl.json\")\n\n# visualise the asl\nBranchViz(branch).show()\n\n# create a state machine\nsm = StateMachine(state={\"Input1\": \"Input1Value\"}, logger=logger)\nsm.apply_branch(branch)\nsm.show_logs()\n```\n",
"bugtrack_url": null,
"license": "",
"summary": "Create AWS Step Functions asl.json with Python",
"version": "1.1.131",
"project_urls": {
"Docs": "https://mrhopko.github.io/pystepfunction/pystepfunction",
"Homepage": "https://github.com/mrhopko/pystepfunction"
},
"split_keywords": [
"aws",
"stepfunctions",
"step functions"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "fe95ac83a32ba34b1500cd798ca2cdb91e9f253290f7024554772fe0f1d40a5a",
"md5": "b4a0d231e99d94f4ddb84719a212698f",
"sha256": "6a3c009a1dd740f220f3199826c6c3ca627b7797fd5eeda8761c63a1289cdab3"
},
"downloads": -1,
"filename": "pystepfunction-1.1.131-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b4a0d231e99d94f4ddb84719a212698f",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 20827,
"upload_time": "2023-10-03T14:42:02",
"upload_time_iso_8601": "2023-10-03T14:42:02.717198Z",
"url": "https://files.pythonhosted.org/packages/fe/95/ac83a32ba34b1500cd798ca2cdb91e9f253290f7024554772fe0f1d40a5a/pystepfunction-1.1.131-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "39e183daa455335500999975397ff22d4fc87c2ba7428a7249514544c3e1fe99",
"md5": "2a303eb57132c9472774b60a20854322",
"sha256": "87e0260473c4efadf9980439e268c6d58476e8eb9dc43ba6ddb549c1fabdb4c1"
},
"downloads": -1,
"filename": "pystepfunction-1.1.131.tar.gz",
"has_sig": false,
"md5_digest": "2a303eb57132c9472774b60a20854322",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 21501,
"upload_time": "2023-10-03T14:42:05",
"upload_time_iso_8601": "2023-10-03T14:42:05.354713Z",
"url": "https://files.pythonhosted.org/packages/39/e1/83daa455335500999975397ff22d4fc87c2ba7428a7249514544c3e1fe99/pystepfunction-1.1.131.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-10-03 14:42:05",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "mrhopko",
"github_project": "pystepfunction",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pystepfunction"
}
