Serverless Workflow Specification - Python SDK
Project description
Serverless Workflow Specification - Python SDK
Provides the Python API/SPI for the Serverless Workflow Specification
With the SDK you can:
- Programmatically build workflow definitions
- Parse workflow JSON and YAML definitions
- Validate workflow definitions
Status
Current sdk version conforms to the Serverless Workflow specification v0.8.
Install dependencies and run test
-
Python 3 required
-
pipenv required
pip install pipenv
pipenv install --dev
pipenv shell
python setup.py pytest
Programmatically build workflow definitions
workflow = Workflow(
id="greeting",
name="Greeting Workflow",
description="Greet Someone",
version='1.0',
specVersion='0.8',
start="Greet",
states=[
OperationState(
name="Greet",
type="operation",
actions=[
Action(
functionRef=FunctionRef(
refName="greetingFunction",
arguments={
"name": "${ .person.name }"
}
),
actionDataFilter=ActionDataFilter(
results="${ .greeting }"
)
)
],
end=True
)
],
functions=[
Function(name="greetingFunction",
operation="file://myapis/greetingapis.json#greeting")
]
)
You can see a full example in the test_workflow.py file
Parse workflow JSON and YAML definitions
Convert from JSON or YAML source
swf_content = """id: greeting
name: Greeting Workflow
version: '1.0'
description: Greet Someone
specVersion: '0.8'
start: Greet
states:
- name: Greet
type: operation
actions:
- functionRef:
refName: greetingFunction
arguments:
name: ${ .person.name }
actionDataFilter:
results: ${ .greeting }
end: true
functions:
- name: greetingFunction
operation: file://myapis/greetingapis.json#greeting
"""
workflow = Workflow.from_source(swf_content)
You can see a full example in the test_workflow.py file
Parse workflow to JSON / YAML
workflow = Workflow(id_="greeting",
name="Greeting Workflow",
description="Greet Someone",
version='1.0',
specVersion='0.8',
start="Greet",
states=[],
functions=[]
)
print(workflow.to_json())
print(workflow.to_yaml())
You can see a full example in the test_workflow.py file
Validate workflow definitions
workflow = Workflow(id_="greeting",
name="Greeting Workflow",
description="Greet Someone",
version='1.0',
specVersion='0.8',
start="Greet",
states=[],
functions=[]
)
WorkflowValidator(Workflow(workflow)).validate()
The validate
method will raise an exception if the provided workflow does not complaint specification.
You can see a full example in the test_workflow_validator file
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for serverlessworkflow.sdk-0.1.1.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | ac01058f2003be4d709f4882fa35a7ae17730faa6c30b8ebbdb736bb1f5bbef5 |
|
MD5 | eb064cd19e0f59b548a836d3cde8fe74 |
|
BLAKE2b-256 | 9e18bb1843089dddfa657b36e701b27a578f60e2d55e435b07dc102536f74996 |
Close
Hashes for serverlessworkflow.sdk-0.1.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5d7644e1c40cd04ab00f01de70a72d356624238b0ee29318289c3c62e35b4ab6 |
|
MD5 | f788ebfbc2a87f57bae92f1b99e8a35b |
|
BLAKE2b-256 | 34400d729863825801f82c8cc4501c1d4eb810676c6d55b143f5ddc3a1c6f6a2 |