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.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | cbf02f7b1513b7812cdbfb5b4e82ec646196019b8b369eb22a9c317b846ebbb5 |
|
MD5 | 729ce7c86a86eae552e455743d30d988 |
|
BLAKE2b-256 | 4978aefc9f1427661315633abb676220706922b831dc6a3c15b74725e98f2890 |
Close
Hashes for serverlessworkflow.sdk-0.1.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 6cd484ff9e6765d3114f12cfac33a901fb6ff37dc3c233554a87d2d21ce49a79 |
|
MD5 | da2e42fa2171020fa9b73d49647e0289 |
|
BLAKE2b-256 | f3897ee618fe2267e6fc82bf79a391dcd4bcadb9f6bcb9dc486b5f0b8f832d18 |