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-1.0.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 71b54a7ff40a0a239f3a2d511f14c5a18b34388bb631dba6f2c0843c77cb4e26 |
|
MD5 | 2be586ff755b1cb66454e552ace1eabb |
|
BLAKE2b-256 | e23a5505a1b8a54cf5d71a996088b3f8b5fd140a819f1617235b9d133f86b078 |
Close
Hashes for serverlessworkflow.sdk-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 47678d90b756cafd3e9db0810f3cb7412d2ccd6a75b96a15fbfa3e113600c2be |
|
MD5 | b4d397e653fe1fd96bd9a3464640c4aa |
|
BLAKE2b-256 | 89d5130832f6b0fa872ba966759a330c763bf70682f0453a52a93c94570b73d7 |