mirror of
https://github.com/sartography/spiff-arena.git
synced 2025-01-09 17:15:49 +00:00
d740518e75
1. It's not just processes, it contains the list of all DMN Decisions as well. 2. It is closely linked to the SpecReference object that can be generated by looking through all the Spec files to find the processes and decisions they contain. 3. It is a cache of information, the file system is the source of truth. Seems likely we will cache more things in the future -- so setting things up this way made sense.
96 lines
3.4 KiB
Python
96 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].identifier
|
|
spec.primary_file_name = filename
|
|
ProcessModelService().save_process_model(spec)
|
|
finally:
|
|
if file:
|
|
file.close()
|
|
return spec
|