mirror of
https://github.com/status-im/spiff-arena.git
synced 2025-01-29 11:15:00 +00:00
195c609d58
Removing (very nearly, except for script unit tests) all the XML Parsing we were doing, see related PR on SpiffWorkflow Moved the Custom Parser into its own file to solve some circular import issues
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""Example_data."""
|
|
import glob
|
|
import os
|
|
from typing import Optional
|
|
|
|
from flask import current_app
|
|
from spiffworkflow_backend.models.process_model import ProcessModelInfo
|
|
from spiffworkflow_backend.services.process_model_service import ProcessModelService
|
|
from spiffworkflow_backend.services.spec_file_service import SpecFileService
|
|
|
|
|
|
class ExampleDataLoader:
|
|
"""ExampleDataLoader."""
|
|
|
|
@staticmethod
|
|
def create_spec(
|
|
process_model_id: str,
|
|
display_name: str = "",
|
|
description: str = "",
|
|
display_order: int = 0,
|
|
bpmn_file_name: Optional[str] = None,
|
|
process_model_source_directory: Optional[str] = None,
|
|
) -> ProcessModelInfo:
|
|
"""Assumes that process_model_source_directory exists in static/bpmn and contains bpmn_file_name.
|
|
|
|
further assumes that bpmn_file_name is the primary file for the process model.
|
|
|
|
if bpmn_file_name is None we load all files in process_model_source_directory,
|
|
otherwise, we only load bpmn_file_name
|
|
"""
|
|
|
|
if process_model_source_directory is None:
|
|
raise Exception("You must include `process_model_source_directory`.")
|
|
|
|
spec = ProcessModelInfo(
|
|
id=process_model_id,
|
|
display_name=display_name,
|
|
description=description,
|
|
display_order=display_order,
|
|
is_review=False,
|
|
)
|
|
workflow_spec_service = ProcessModelService()
|
|
workflow_spec_service.add_spec(spec)
|
|
|
|
bpmn_file_name_with_extension = bpmn_file_name
|
|
if not bpmn_file_name_with_extension:
|
|
bpmn_file_name_with_extension = process_model_id
|
|
|
|
if not bpmn_file_name_with_extension.endswith(".bpmn"):
|
|
bpmn_file_name_with_extension += ".bpmn"
|
|
|
|
process_model_source_directory_to_use = process_model_source_directory
|
|
if not process_model_source_directory_to_use:
|
|
process_model_source_directory_to_use = process_model_id
|
|
|
|
file_name_matcher = "*.*"
|
|
if bpmn_file_name:
|
|
file_name_matcher = bpmn_file_name_with_extension
|
|
|
|
# we need instance_path here for nox tests
|
|
file_glob = os.path.join(
|
|
current_app.instance_path,
|
|
"..",
|
|
"..",
|
|
"tests",
|
|
"data",
|
|
process_model_source_directory_to_use,
|
|
file_name_matcher,
|
|
)
|
|
|
|
files = glob.glob(file_glob)
|
|
for file_path in files:
|
|
if os.path.isdir(file_path):
|
|
continue # Don't try to process sub directories
|
|
|
|
filename = os.path.basename(file_path)
|
|
# since there are multiple bpmn files in a test data directory, ensure we set the correct one as the primary
|
|
is_primary = filename.lower() == bpmn_file_name_with_extension
|
|
file = None
|
|
try:
|
|
file = open(file_path, "rb")
|
|
data = file.read()
|
|
file_info = SpecFileService.add_file(
|
|
process_model_info=spec, file_name=filename, binary_data=data
|
|
)
|
|
if is_primary:
|
|
references = SpecFileService.get_references_for_file(file_info, spec)
|
|
spec.primary_process_id = references[0].id
|
|
spec.primary_file_name = filename
|
|
ProcessModelService().save_process_model(spec)
|
|
finally:
|
|
if file:
|
|
file.close()
|
|
return spec
|