cr-connect-workflow/example_data.py

83 lines
3.6 KiB
Python
Raw Normal View History

import glob
import os
from crc import app, db, session
from crc.models.file import CONTENT_TYPES
from crc.models.workflow import WorkflowSpecModel
from crc.services.document_service import DocumentService
from crc.services.reference_file_service import ReferenceFileService
from crc.services.spec_file_service import SpecFileService
from crc.services.study_service import StudyService
class ExampleDataLoader:
@staticmethod
def clean_db():
session.flush() # Clear out any transactions before deleting it all to avoid spurious errors.
engine = session.bind.engine
connection = engine.connect()
for table in reversed(db.metadata.sorted_tables):
if engine.dialect.has_table(connection, table):
session.execute(table.delete())
session.commit()
session.flush()
2021-10-05 17:36:53 +00:00
def create_spec(self, id, display_name="", description="", filepath=None, master_spec=False,
category_id=None, display_order=None, from_tests=False, standalone=False, library=False):
"""Assumes that a directory exists in static/bpmn with the same name as the given id.
further assumes that the [id].bpmn is the primary file for the workflow.
returns an array of data models to be added to the database."""
global file
spec = WorkflowSpecModel(id=id,
display_name=display_name,
description=description,
is_master_spec=master_spec,
2020-04-10 15:13:43 +00:00
category_id=category_id,
display_order=display_order,
standalone=standalone,
library=library)
db.session.add(spec)
db.session.commit()
if not filepath and not from_tests:
filepath = os.path.join(app.root_path, 'static', 'bpmn', id, "*.*")
if not filepath and from_tests:
filepath = os.path.join(app.root_path, '..', 'tests', 'data', id, "*.*")
files = glob.glob(filepath)
for file_path in files:
if os.path.isdir(file_path):
continue # Don't try to process sub directories
noise, file_extension = os.path.splitext(file_path)
filename = os.path.basename(file_path)
is_primary = filename.lower() == id + '.bpmn'
2020-05-05 20:15:38 +00:00
file = None
try:
file = open(file_path, 'rb')
data = file.read()
content_type = CONTENT_TYPES[file_extension[1:]]
SpecFileService.add_file(workflow_spec=spec, file_name=filename, binary_data=data)
if is_primary:
SpecFileService.set_primary_bpmn(spec, filename)
except IsADirectoryError as de:
# Ignore sub directories
pass
finally:
if file:
file.close()
return spec
def load_reference_documents(self):
file_path = os.path.join(app.root_path, 'static', 'reference', 'documents.xlsx')
file = open(file_path, "rb")
ReferenceFileService.add_reference_file(DocumentService.DOCUMENT_LIST,
file.read())
file.close()
file_path = os.path.join(app.root_path, 'static', 'reference', 'investigators.xlsx')
file = open(file_path, "rb")
ReferenceFileService.add_reference_file(StudyService.INVESTIGATOR_LIST,
file.read())
file.close()