sartography/spiffworkflow-backend
1
0
Fork 0
mirror of https://github.com/sartography/spiffworkflow-backend.git synced 2025-02-23 12:58:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add the ability to call custom built-in functions from within a Python script. With the first real script being a "get_env()" which will return the current environment (ie, 'testing', 'development' etc...)

Browse Source
This commit is contained in:
Dan 2022-09-30 16:50:45 -04:00
parent ce1e605fb5
commit ac999a8e65
7 changed files with 99 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/spiffworkflow_backend/scripts/fact_service.py Normal file
View File

@ -0,0 +1,28 @@
import requests
from spiffworkflow_backend.scripts.script import Script
class FactService(Script):
def get_description(self):
return """Just your basic class that can pull in data from a few api endpoints and
do a basic task."""
def run(self, **kwargs):
if "type" not in kwargs:
raise Exception("Please specify a 'type' of fact as a keyword argument.")
else:
fact = kwargs["type"]
if fact == "cat":
details = "The cat in the hat" # self.get_cat()
elif fact == "norris":
details = "Chuck Norris doesn’t read books. He stares them down until he gets the information he wants." # self.get_norris()
elif fact == "buzzword":
details = "Move the Needle." # self.get_buzzword()
else:
details = "unknown fact type."
#self.add_data_to_task(task, details)
return details

13
src/spiffworkflow_backend/scripts/get_env.py Normal file
View File

@ -0,0 +1,13 @@
import requests
from flask import current_app
from spiffworkflow_backend.scripts.script import Script
from SpiffWorkflow import Task as SpiffTask # type: ignore
class GetEnv(Script):
def get_description(self):
return """Returns the current environment - ie TESTING, STAGING, PRODUCTION"""
def run(self, task: SpiffTask, environment_identifier: str, **kwargs):
return environment_identifier

92
src/spiffworkflow_backend/scripts/script.py
View File

@ -7,6 +7,7 @@ import pkgutil
from abc import abstractmethod
from typing import Any
from typing import Callable
from SpiffWorkflow import Task as SpiffTask # type: ignore
from flask_bpmn.api.api_error import ApiError
@ -27,32 +28,19 @@ class Script:
raise ApiError("invalid_script", "This script does not supply a description.")
@abstractmethod
def do_task(
self, task: Task, workflow_id: int, *args: list[Any], **kwargs: dict[Any, Any]
def run(
self, task: Task, environment_identifier: str, *args: list[Any], **kwargs: dict[Any, Any]
) -> None:
"""Do_task."""
"""run."""
raise ApiError(
"invalid_script",
"This is an internal error. The script you are trying to execute '%s' "
% self.__class__.__name__
+ "does not properly implement the do_task function.",
)
@abstractmethod
def do_task_validate_only(
self, task: Task, workflow_id: int, *args: list[Any], **kwargs: dict[Any, Any]
) -> None:
"""Do_task_validate_only."""
raise ApiError(
"invalid_script",
"This is an internal error. The script you are trying to execute '%s' "
% self.__class__.__name__
+ "does must provide a validate_only option that mimics the do_task, "
+ "but does not make external calls or database updates.",
+ "does not properly implement the run function.",
)
@staticmethod
def generate_augmented_list(task: Task, workflow_id: int) -> dict[str, Callable]:
def generate_augmented_list(task: SpiffTask, environment_identifier: str) -> dict[str, Callable]:
"""This makes a dictionary of lambda functions that are closed over the class instance that they represent.
This is passed into PythonScriptParser as a list of helper functions that are
@ -64,7 +52,7 @@ class Script:
"""
def make_closure(
subclass: type[Script], task: Task, workflow_id: int
subclass: type[Script], task: Task, environment_identifier: str
) -> Callable:
"""Yes - this is black magic.
@ -76,8 +64,8 @@ class Script:
that we created.
"""
instance = subclass()
return lambda *ar, **kw: subclass.do_task(
instance, task, workflow_id, *ar, **kw
return lambda *ar, **kw: subclass.run(
instance, task=task, environment_identifier=environment_identifier, *ar, **kw
)
execlist = {}
@ -85,41 +73,9 @@ class Script:
for x in range(len(subclasses)):
subclass = subclasses[x]
execlist[subclass.__module__.split(".")[-1]] = make_closure(
subclass, task, workflow_id
)
subclass, task=task, environment_identifier=environment_identifier)
return execlist
@staticmethod
def generate_augmented_validate_list(
task: Task, workflow_id: int
) -> dict[str, Callable]:
"""This makes a dictionary of lambda functions that are closed over the class instance that they represent.
This is passed into PythonScriptParser as a list of helper functions that are
available for running. In general, they maintain the do_task call structure that they had, but
they always return a value rather than updating the task data.
We may be able to remove the task for each of these calls if we are not using it other than potentially
updating the task data.
"""
def make_closure_validate(
subclass: type[Script], task: Task, workflow_id: int
) -> Callable:
"""Make_closure_validate."""
instance = subclass()
return lambda *a, **b: subclass.do_task_validate_only(
instance, task, workflow_id, *a, **b
)
execlist = {}
subclasses = Script.get_all_subclasses()
for x in range(len(subclasses)):
subclass = subclasses[x]
execlist[subclass.__module__.split(".")[-1]] = make_closure_validate(
subclass, task, workflow_id
)
return execlist
@classmethod
def get_all_subclasses(cls) -> list[type[Script]]:
@ -145,30 +101,4 @@ class Script:
all_subclasses.append(subclass)
all_subclasses.extend(Script._get_all_subclasses(subclass))
return all_subclasses
def add_data_to_task(self, task: Task, data: Any) -> None:
"""Add_data_to_task."""
key = self.__class__.__name__
if task.data is None:
task.data = {}
if key in task.data:
task.data[key].update(data)
else:
task.data[key] = data
class ScriptValidationError:
"""ScriptValidationError."""
def __init__(self, code: str, message: str):
"""__init__."""
self.code = code
self.message = message
@classmethod
def from_api_error(cls, api_error: ApiError) -> ScriptValidationError:
"""From_api_error."""
return cls(api_error.code, api_error.message)
return all_subclasses

17
src/spiffworkflow_backend/services/process_instance_processor.py
View File

@ -68,6 +68,7 @@ from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.models.task_event import TaskAction
from spiffworkflow_backend.models.task_event import TaskEventModel
from spiffworkflow_backend.models.user import UserModelSchema
from spiffworkflow_backend.scripts.script import Script
from spiffworkflow_backend.services.file_system_service import FileSystemService
from spiffworkflow_backend.services.process_model_service import ProcessModelService
from spiffworkflow_backend.services.service_task_service import ServiceTaskService
@ -86,6 +87,10 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
scripts directory available for execution.
"""
def __get_augment_methods(self, task):
return Script.generate_augmented_list(task, current_app.env)
def evaluate(self, task: SpiffTask, expression: str) -> Any:
"""Evaluate."""
return self._evaluate(expression, task.data, task)
@ -97,10 +102,15 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
task: Optional[SpiffTask] = None,
external_methods: Optional[Dict[str, Any]] = None,
) -> Any:
methods = self.__get_augment_methods(task)
if external_methods:
methods.update(external_methods)
"""Evaluate the given expression, within the context of the given task and return the result."""
try:
return super()._evaluate(
expression, context, external_methods=external_methods
expression, context, external_methods=methods
)
except Exception as exception:
if task is None:
@ -120,7 +130,10 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
) -> None:
"""Execute."""
try:
super().execute(task, script, external_methods)
methods = self.__get_augment_methods(task)
if external_methods:
methods.update(external_methods)
super().execute(task, script, methods)
except WorkflowException as e:
raise e
except Exception as e:

5
tests/data/simple_script/simple_script.bpmn
View File

@ -1,5 +1,5 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_1kbzkan" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="4.2.0">
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_1kbzkan" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="5.0.0">
<bpmn:process id="Process_SimpleScript" name="Simple Script" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_1k9q28c</bpmn:outgoing>
@ -15,7 +15,8 @@
<bpmn:outgoing>Flow_1fviiob</bpmn:outgoing>
<bpmn:script>a = 1
b = 2
c = a + b</bpmn:script>
c = a + b
norris=fact_service(type='norris')</bpmn:script>
</bpmn:scriptTask>
<bpmn:manualTask id="Activity_DisplayData" name="Display Data">
<bpmn:documentation>## Display Data

19
tests/spiffworkflow_backend/unit/test_environment_var_script.py Normal file
View File

@ -0,0 +1,19 @@
from flask import Flask
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
class TestEnvironmentVarScript(BaseTest):
# it's not totally obvious we want to keep this test/file
def test_script_engine_can_use_custom_scripts(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_script_engine_takes_data_and_returns_expected_results."""
with app.app_context():
script_engine = ProcessInstanceProcessor._script_engine
result = script_engine._evaluate("get_env()", {})
assert result == "testing"

10
tests/spiffworkflow_backend/unit/test_process_instance_processor.py
View File

@ -42,3 +42,13 @@ def test_get_bpmn_process_instance_from_process_model_can_acccess_tasks_from_sub
# tasks = bpmn_process_instance.get_tasks(TaskState.ANY_MASK)
# task_ids = [t.task_spec.name for t in tasks]
# print(f"task_ids: {task_ids}")
# it's not totally obvious we want to keep this test/file
def test_script_engine_can_use_custom_scripts(
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_script_engine_takes_data_and_returns_expected_results."""
script_engine = ProcessInstanceProcessor._script_engine
result = script_engine._evaluate("fact_service(type='norris')", {})
assert result == "Chuck Norris doesn’t read books. He stares them down until he gets the information he wants."