- Notifications
You must be signed in to change notification settings - Fork8
Python SDK for Serverless Workflow
License
serverlessworkflow/sdk-python
Folders and files
| Name | Name | Last commit message | Last commit date | |
|---|---|---|---|---|
Repository files navigation
Provides the Python API/SPI for theServerless Workflow Specification
With the SDK you can:
- Programmatically build workflow definitions
- Parse workflow JSON and YAML definitions
- Validate workflow definitions
Current sdk version conforms to theServerless Workflow specification v0.8.
Python 3 required
pipenv required
pip install pipenv
pipenv install --devpipenv run pip install 'setuptools==70.3.0'pipenv shellpython setup.py pytest workflow = Workflow( , 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 thetest_workflow.py file
swf_content = """id: greetingname: Greeting Workflowversion: '1.0'description: Greet SomeonespecVersion: '0.8'start: Greetstates:- name: Greet type: operation actions: - functionRef: refName: greetingFunction arguments: name: ${ .person.name } actionDataFilter: results: ${ .greeting } end: truefunctions:- name: greetingFunction operation: file://myapis/greetingapis.json#greeting""" workflow = Workflow.from_source(swf_content)You can see a full example in thetest_workflow.py file
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 thetest_workflow.py file
workflow = Workflow(id_="greeting", name="Greeting Workflow", description="Greet Someone", version='1.0', specVersion='0.8', start="Greet", states=[], functions=[])WorkflowValidator(Workflow(workflow)).validate()Thevalidate method will raise an exception if the provided workflow does not complaint specification.
You can see a full example in thetest_workflow_validator file
To generate the workflow graph diagram:
fromserverlessworkflow.sdk.workflowimportWorkflowfromserverlessworkflow.sdk.state_machine_helperimportStateMachineHelperdefmain():subflows= []withopen("tests/examples/graph.json")asf:workflow=Workflow.from_source(f.read())withopen("tests/examples/advertise-listing.json")asf:subflows.append(Workflow.from_source(f.read()))withopen("tests/examples/second-subgraph.json")asf:subflows.append(Workflow.from_source(f.read()))machine_helper=StateMachineHelper(workflow=workflow,get_actions=True,subflows=subflows)machine_helper.draw('diagram.svg')if__name__=="__main__":main()
TheStateMachineHelper can be set withget_actions asFalse and the produced diagram will not represent the actions inside each state (it will only create a diagram with the states and their transitions). Moreover, the developer may not give anysubflows, and they simply will not be generated.As for thedraw method, the developer can also specifygraph_engine='mermaid'. In that case, the method will not generate a figure, but rather the Mermaid code that can be executed, for instance, in theMermaid Live Editor.
It is also possible to only generate the workflow state machine. An example on how to do so can be analyzed in thestate_machine_helper source code.
About
Python SDK for Serverless Workflow
Topics
Resources
License
Code of conduct
Uh oh!
There was an error while loading.Please reload this page.
Stars
Watchers
Forks
Packages0
Uh oh!
There was an error while loading.Please reload this page.
Contributors6
Uh oh!
There was an error while loading.Please reload this page.