ran fix docstrings on src w/ burnettk

This commit is contained in:
jasquat 2022-05-31 09:37:12 -04:00
parent 50dbac1d8b
commit 2ee0799db5
5 changed files with 26 additions and 16 deletions

View File

@ -1,4 +1,4 @@
"""Default."""
from os import environ
BPMN_SPEC_ABSOLUTE_DIR = environ.get('BPMN_SPEC_ABSOLUTE_DIR', default="")
BPMN_SPEC_ABSOLUTE_DIR = environ.get('BPMN_SPEC_ABSOLUTE_DIR', default="")

View File

@ -0,0 +1 @@
"""__init__."""

View File

@ -1,61 +1,67 @@
"""APIs for dealing with process groups, process models, and process instances."""
from typing import Any
import connexion
import requests
from flask import Blueprint, g, render_template
from flask import Blueprint, render_template
from flask_bpmn.models.db import db
from flask import request
from spiff_workflow_webapp.models.user import UserModel
from spiff_workflow_webapp.services.process_instance_processor import ProcessInstanceProcessor
from spiff_workflow_webapp.services.process_instance_service import ProcessInstanceService
from spiff_workflow_webapp.services.user_service import UserService
from spiff_workflow_webapp.services.spec_file_service import SpecFileService
from spiff_workflow_webapp.services.process_model_service import ProcessModelService
admin_blueprint = Blueprint("admin", __name__, template_folder='templates', static_folder='static')
@admin_blueprint.route("/index", methods=["GET"])
def hello_world():
"""Hello_world."""
return render_template('index.html')
@admin_blueprint.route("/view/<process_model_id>/<file_id>", methods=["GET"])
def view_bpmn(process_model_id, file_id):
"""View_bpmn."""
process_model = ProcessModelService().get_spec(process_model_id)
files = SpecFileService.get_files(process_model)
SpecFileService.get_files(process_model)
bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name)
return render_template('view.html', bpmn_xml=bpmn_xml.decode("utf-8") )
return render_template('view.html', bpmn_xml=bpmn_xml.decode("utf-8"))
@admin_blueprint.route("/run/<process_model_id>", methods=["GET"])
def run_bpmn(process_model_id):
user = find_or_create_user('Mr. Test') # Fixme - sheesh!
"""Run_bpmn."""
user = find_or_create_user('Mr. Test') # Fixme - sheesh!
process_instance = ProcessInstanceService.create_process_instance(process_model_id, user)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps()
result = processor.get_data()
process_model = ProcessModelService().get_spec(process_model_id)
files = SpecFileService.get_files(process_model)
SpecFileService.get_files(process_model)
bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name)
return render_template('view.html', bpmn_xml=bpmn_xml.decode("utf-8"), result=result,
process_model_id=process_model_id)
@admin_blueprint.route("/edit/<process_model_id>", methods=["GET"])
def edit_bpmn(process_model_id):
"""Edit_bpmn."""
process_model = ProcessModelService().get_spec(process_model_id)
files = SpecFileService.get_files(process_model)
SpecFileService.get_files(process_model)
bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name)
return render_template('edit.html', bpmn_xml=bpmn_xml.decode("utf-8"),
process_model_id=process_model_id)
@admin_blueprint.route("/save/<process_model_id>", methods=["POST"])
def save_bpmn(process_model_id):
"""Save_bpmn."""
process_model = ProcessModelService().get_spec(process_model_id)
primary_file = SpecFileService.get_files(process_model, process_model.primary_file_name)
SpecFileService.get_files(process_model, process_model.primary_file_name)
SpecFileService.update_file(process_model, process_model.primary_file_name, request.get_data())
bpmn_xml = SpecFileService.get_data(process_model, process_model.primary_file_name)
return render_template('edit.html', bpmn_xml=bpmn_xml.decode("utf-8"),
@ -64,14 +70,16 @@ def save_bpmn(process_model_id):
@admin_blueprint.route("/process_models", methods=["GET"])
def listProcessModels():
"""ListProcessModels."""
models = ProcessModelService().get_specs()
return render_template('process_models.html', models=models)
def find_or_create_user(username: str = "test_user1") -> Any:
"""Find_or_create_user."""
user = UserModel.query.filter_by(username=username).first()
if user is None:
user = UserModel(username=username)
db.session.add(user)
db.session.commit()
return user

View File

@ -29,7 +29,6 @@ from spiff_workflow_webapp.models.task_event import TaskEventModel, TaskAction
from spiff_workflow_webapp.models.user import UserModelSchema
from spiff_workflow_webapp.models.process_model import ProcessModelInfo
from spiff_workflow_webapp.models.process_instance import ProcessInstanceModel, ProcessInstanceStatus
from spiff_workflow_webapp.scripts.script import Script
from spiff_workflow_webapp.services.spec_file_service import SpecFileService
# from crc.services.user_file_service import UserFileService
from spiff_workflow_webapp.services.user_service import UserService
@ -47,7 +46,6 @@ class CustomBpmnScriptEngine(PythonScriptEngine):
"""Evaluate."""
return self._evaluate(expression, task.data, task)
def _evaluate(self, expression, context, task=None, external_methods=None):
"""Evaluate the given expression, within the context of the given task and return the result."""
try:
@ -207,10 +205,12 @@ class ProcessInstanceProcessor(object):
@staticmethod
def __get_bpmn_workflow(process_instance_model: ProcessInstanceModel, spec: WorkflowSpec = None, validate_only=False):
"""__get_bpmn_workflow."""
if process_instance_model.bpmn_workflow_json:
version = ProcessInstanceProcessor._serializer.get_version(process_instance_model.bpmn_workflow_json)
if(version == ProcessInstanceProcessor.SERIALIZER_VERSION):
bpmn_workflow = ProcessInstanceProcessor._serializer.deserialize_json(process_instance_model.bpmn_workflow_json)
bpmn_workflow = ProcessInstanceProcessor._serializer.deserialize_json(
process_instance_model.bpmn_workflow_json)
else:
bpmn_workflow = ProcessInstanceProcessor.\
_old_serializer.deserialize_workflow(process_instance_model.bpmn_workflow_json,

View File

@ -74,6 +74,7 @@ class ProcessInstanceService():
@staticmethod
def update_navigation(navigation: List[NavItem], processor: ProcessInstanceProcessor):
"""Update_navigation."""
# Recursive function to walk down through children, and clean up descriptions, and statuses
for nav_item in navigation:
spiff_task = processor.bpmn_workflow.get_task(nav_item.task_id)
@ -116,4 +117,4 @@ class ProcessInstanceService():
current_app.logger.error("missing_form_data", missing_form_error, exc_info=True)
return {}
else:
return {}
return {}