mirror of
https://github.com/sartography/cr-connect-workflow.git
synced 2025-02-23 21:28:32 +00:00
da7cae51b8
Improving the study_info script documentation to provide detailed examples of values returned based on arguments. Making the tests a little more targetted and less subject to breaking through better mocks. Allow all tests to pass even when ther protocol builder mock isn't running locally. Removing the duplication of reference files in tests and static, as this seems silly to me at the moment.
241 lines
10 KiB
Python
241 lines
10 KiB
Python
import json
|
|
import os
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
from xml.etree import ElementTree
|
|
|
|
from pandas import ExcelFile
|
|
|
|
from crc import session
|
|
from crc.api.common import ApiError
|
|
from crc.models.file import FileType, FileDataModel, FileModel, LookupFileModel, LookupDataModel
|
|
from crc.models.workflow import WorkflowSpecModel
|
|
from crc.services.workflow_processor import WorkflowProcessor
|
|
import hashlib
|
|
|
|
|
|
class FileService(object):
|
|
"""Provides consistent management and rules for storing, retrieving and processing files."""
|
|
DOCUMENT_LIST = "irb_documents.xlsx"
|
|
INVESTIGATOR_LIST = "investigators.xlsx"
|
|
|
|
@staticmethod
|
|
def add_workflow_spec_file(workflow_spec: WorkflowSpecModel,
|
|
name, content_type, binary_data, primary=False, is_status=False):
|
|
"""Create a new file and associate it with a workflow spec."""
|
|
file_model = FileModel(
|
|
workflow_spec_id=workflow_spec.id,
|
|
name=name,
|
|
primary=primary,
|
|
is_status=is_status,
|
|
)
|
|
|
|
return FileService.update_file(file_model, binary_data, content_type)
|
|
|
|
@staticmethod
|
|
def is_allowed_document(code):
|
|
data_model = FileService.get_reference_file_data(FileService.DOCUMENT_LIST)
|
|
xls = ExcelFile(data_model.data)
|
|
df = xls.parse(xls.sheet_names[0])
|
|
return code in df['code'].values
|
|
|
|
@staticmethod
|
|
def add_form_field_file(study_id, workflow_id, task_id, form_field_key, name, content_type, binary_data):
|
|
"""Create a new file and associate it with a user task form field within a workflow.
|
|
Please note that the form_field_key MUST be a known file in the irb_documents.xslx reference document."""
|
|
if not FileService.is_allowed_document(form_field_key):
|
|
raise ApiError("invalid_form_field_key",
|
|
"When uploading files, the form field id must match a known document in the "
|
|
"irb_docunents.xslx reference file. This code is not found in that file '%s'" % form_field_key)
|
|
|
|
file_model = FileModel(
|
|
study_id=study_id,
|
|
workflow_id=workflow_id,
|
|
task_id=task_id,
|
|
name=name,
|
|
form_field_key=form_field_key,
|
|
irb_doc_code=form_field_key
|
|
)
|
|
return FileService.update_file(file_model, binary_data, content_type)
|
|
|
|
@staticmethod
|
|
def get_reference_data(reference_file_name, index_column, int_columns=[]):
|
|
""" Opens a reference file (assumes that it is xls file) and returns the data as a
|
|
dictionary, each row keyed on the given index_column name. If there are columns
|
|
that should be represented as integers, pass these as an array of int_columns, lest
|
|
you get '1.0' rather than '1' """
|
|
data_model = FileService.get_reference_file_data(reference_file_name)
|
|
xls = ExcelFile(data_model.data)
|
|
df = xls.parse(xls.sheet_names[0])
|
|
for c in int_columns:
|
|
df[c] = df[c].fillna(0)
|
|
df = df.astype({c: 'Int64'})
|
|
df = df.fillna('')
|
|
df = df.applymap(str)
|
|
df = df.set_index(index_column)
|
|
return json.loads(df.to_json(orient='index'))
|
|
|
|
@staticmethod
|
|
def add_task_file(study_id, workflow_id, workflow_spec_id, task_id, name, content_type, binary_data,
|
|
irb_doc_code=None):
|
|
"""Create a new file and associate it with an executing task within a workflow."""
|
|
file_model = FileModel(
|
|
study_id=study_id,
|
|
workflow_id=workflow_id,
|
|
workflow_spec_id=workflow_spec_id,
|
|
task_id=task_id,
|
|
name=name,
|
|
irb_doc_code=irb_doc_code
|
|
)
|
|
return FileService.update_file(file_model, binary_data, content_type)
|
|
|
|
@staticmethod
|
|
def add_reference_file(name, content_type, binary_data):
|
|
"""Create a file with the given name, but not associated with a spec or workflow.
|
|
Only one file with the given reference name can exist."""
|
|
file_model = session.query(FileModel). \
|
|
filter(FileModel.is_reference == True). \
|
|
filter(FileModel.name == name).first()
|
|
if not file_model:
|
|
file_model = FileModel(
|
|
name=name,
|
|
is_reference=True
|
|
)
|
|
return FileService.update_file(file_model, binary_data, content_type)
|
|
|
|
@staticmethod
|
|
def get_extension(file_name):
|
|
basename, file_extension = os.path.splitext(file_name)
|
|
return file_extension.lower().strip()[1:]
|
|
|
|
@staticmethod
|
|
def update_file(file_model, binary_data, content_type):
|
|
|
|
file_data_model = session.query(FileDataModel). \
|
|
filter_by(file_model_id=file_model.id,
|
|
version=file_model.latest_version
|
|
).with_for_update().first()
|
|
md5_checksum = UUID(hashlib.md5(binary_data).hexdigest())
|
|
if (file_data_model is not None and md5_checksum == file_data_model.md5_hash):
|
|
# This file does not need to be updated, it's the same file.
|
|
return file_model
|
|
|
|
# Verify the extension
|
|
file_extension = FileService.get_extension(file_model.name)
|
|
if file_extension not in FileType._member_names_:
|
|
raise ApiError('unknown_extension',
|
|
'The file you provided does not have an accepted extension:' +
|
|
file_extension, status_code=404)
|
|
else:
|
|
file_model.type = FileType[file_extension]
|
|
file_model.content_type = content_type
|
|
|
|
if file_data_model is None:
|
|
version = 1
|
|
else:
|
|
version = file_data_model.version + 1
|
|
|
|
# If this is a BPMN, extract the process id.
|
|
if file_model.type == FileType.bpmn:
|
|
bpmn: ElementTree.Element = ElementTree.fromstring(binary_data)
|
|
file_model.primary_process_id = WorkflowProcessor.get_process_id(bpmn)
|
|
|
|
file_model.latest_version = version
|
|
file_data_model = FileDataModel(data=binary_data, file_model=file_model, version=version,
|
|
md5_hash=md5_checksum, last_updated=datetime.now())
|
|
|
|
session.add_all([file_model, file_data_model])
|
|
session.commit()
|
|
session.flush() # Assure the id is set on the model before returning it.
|
|
return file_model
|
|
|
|
@staticmethod
|
|
def get_files(workflow_spec_id=None,
|
|
study_id=None, workflow_id=None, task_id=None, form_field_key=None,
|
|
name=None, is_reference=False, irb_doc_code=None):
|
|
query = session.query(FileModel).filter_by(is_reference=is_reference)
|
|
if workflow_spec_id:
|
|
query = query.filter_by(workflow_spec_id=workflow_spec_id)
|
|
if study_id:
|
|
query = query.filter_by(study_id=study_id)
|
|
if workflow_id:
|
|
query = query.filter_by(workflow_id=workflow_id)
|
|
if task_id:
|
|
query = query.filter_by(task_id=str(task_id))
|
|
if form_field_key:
|
|
query = query.filter_by(form_field_key=form_field_key)
|
|
if name:
|
|
query = query.filter_by(name=name)
|
|
if irb_doc_code:
|
|
query = query.filter_by(irb_doc_code=irb_doc_code)
|
|
|
|
results = query.all()
|
|
return results
|
|
|
|
@staticmethod
|
|
def get_file_data(file_id, file_model=None):
|
|
"""Returns the file_data that is associated with the file model id, if an actual file_model
|
|
is provided, uses that rather than looking it up again."""
|
|
if file_model is None:
|
|
file_model = session.query(FileModel).filter(FileModel.id == file_id).first()
|
|
return session.query(FileDataModel) \
|
|
.filter(FileDataModel.file_model_id == file_id) \
|
|
.filter(FileDataModel.version == file_model.latest_version) \
|
|
.first()
|
|
|
|
@staticmethod
|
|
def get_reference_file_data(file_name):
|
|
file_model = session.query(FileModel). \
|
|
filter(FileModel.is_reference == True). \
|
|
filter(FileModel.name == file_name).first()
|
|
if not file_model:
|
|
raise ApiError("file_not_found", "There is no reference file with the name '%s'" % file_name)
|
|
return FileService.get_file_data(file_model.id, file_model)
|
|
|
|
@staticmethod
|
|
def get_workflow_file_data(workflow, file_name):
|
|
"""Given a SPIFF Workflow Model, tracks down a file with the given name in the database and returns it's data"""
|
|
workflow_spec_model = FileService.find_spec_model_in_db(workflow)
|
|
|
|
if workflow_spec_model is None:
|
|
raise ApiError(code="workflow_model_error",
|
|
message="Something is wrong. I can't find the workflow you are using.")
|
|
|
|
file_data_model = session.query(FileDataModel) \
|
|
.join(FileModel) \
|
|
.filter(FileModel.name == file_name) \
|
|
.filter(FileModel.workflow_spec_id == workflow_spec_model.id).first()
|
|
|
|
if file_data_model is None:
|
|
raise ApiError(code="file_missing",
|
|
message="Can not find a file called '%s' within workflow specification '%s'"
|
|
% (file_name, workflow_spec_model.id))
|
|
|
|
return file_data_model
|
|
|
|
@staticmethod
|
|
def find_spec_model_in_db(workflow):
|
|
""" Search for the workflow """
|
|
# When the workflow spec model is created, we record the primary process id,
|
|
# then we can look it up. As there is the potential for sub-workflows, we
|
|
# may need to travel up to locate the primary process.
|
|
spec = workflow.spec
|
|
workflow_model = session.query(WorkflowSpecModel).join(FileModel). \
|
|
filter(FileModel.primary_process_id == spec.name).first()
|
|
if workflow_model is None and workflow != workflow.outer_workflow:
|
|
return FileService.find_spec_model_in_db(workflow.outer_workflow)
|
|
|
|
return workflow_model
|
|
|
|
@staticmethod
|
|
def delete_file(file_id):
|
|
data_models = session.query(FileDataModel).filter_by(file_model_id=file_id).all()
|
|
for dm in data_models:
|
|
lookup_files = session.query(LookupFileModel).filter_by(file_data_model_id=dm.id).all()
|
|
for lf in lookup_files:
|
|
session.query(LookupDataModel).filter_by(lookup_file_model_id=lf.id).delete()
|
|
session.query(LookupFileModel).filter_by(id=lf.id).delete()
|
|
session.query(FileDataModel).filter_by(file_model_id=file_id).delete()
|
|
session.query(FileModel).filter_by(id=file_id).delete()
|
|
session.commit()
|