Spaces:
Sleeping
Sleeping
File size: 7,292 Bytes
5854e13 bacb197 5854e13 050fe0d 5854e13 050fe0d 5854e13 050fe0d 5854e13 bacb197 5854e13 bacb197 5854e13 bacb197 5854e13 bacb197 5854e13 bacb197 5854e13 8beb233 5854e13 8beb233 5854e13 8beb233 5854e13 8beb233 5854e13 8beb233 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | from bson import ObjectId
from datetime import datetime
from db import get_workflows_collection
from models.department import Department
class Workflow:
def __init__(self, title, description, data_requirements=None, raw_forms=None, form_fields=None,
department_id=None, _id=None, created_at=None, updated_at=None,
markdown_template=None, template_name=None):
self.title = title
self.description = description
self.data_requirements = data_requirements or [] # List of (String, String) for fields and descriptions
self.raw_forms = raw_forms or [] # List of file paths
self.form_fields = form_fields or [] # List of (Number, String) for positions and fields
self.department_id = department_id
self._id = _id
self.created_at = created_at or datetime.now()
self.updated_at = updated_at or datetime.now()
self.markdown_template = markdown_template # The markdown content as a string
self.template_name = template_name # The name of the uploaded markdown file
def to_dict(self):
"""Convert instance to dictionary"""
workflow_dict = {
"title": self.title,
"description": self.description,
"data_requirements": self.data_requirements,
"raw_forms": self.raw_forms,
"form_fields": self.form_fields,
"department_id": str(self.department_id) if self.department_id else None,
"created_at": self.created_at,
"updated_at": self.updated_at,
"markdown_template": self.markdown_template,
"template_name": self.template_name
}
if self._id:
workflow_dict["_id"] = str(self._id)
return workflow_dict
@classmethod
def from_dict(cls, workflow_dict):
"""Create instance from dictionary"""
if "_id" in workflow_dict and workflow_dict["_id"]:
workflow_dict["_id"] = ObjectId(workflow_dict["_id"]) if isinstance(workflow_dict["_id"], str) else workflow_dict["_id"]
if "department_id" in workflow_dict and workflow_dict["department_id"]:
workflow_dict["department_id"] = ObjectId(workflow_dict["department_id"]) if isinstance(workflow_dict["department_id"], str) else workflow_dict["department_id"]
return cls(**workflow_dict)
def save(self):
"""Save workflow to database"""
workflows_collection = get_workflows_collection()
workflow_dict = self.to_dict()
if self._id:
# Update existing workflow
workflow_dict["updated_at"] = datetime.now()
if "_id" in workflow_dict:
del workflow_dict["_id"]
result = workflows_collection.update_one(
{"_id": ObjectId(self._id)},
{"$set": workflow_dict}
)
success = result.modified_count > 0
else:
# Insert new workflow
workflow_dict["created_at"] = datetime.now()
workflow_dict["updated_at"] = datetime.now()
result = workflows_collection.insert_one(workflow_dict)
self._id = result.inserted_id
success = result.acknowledged
# Add workflow to department if insert was successful
if success and self.department_id:
department = Department.find_by_id(self.department_id)
if department:
department.add_workflow(self._id)
return success
@classmethod
def find_by_id(cls, workflow_id):
"""Find workflow by ID"""
workflows_collection = get_workflows_collection()
workflow_data = workflows_collection.find_one({"_id": ObjectId(workflow_id)})
if workflow_data:
return cls.from_dict(workflow_data)
return None
@classmethod
def find_by_title(cls, title, department_id=None):
"""Find workflow by title, optionally filtered by department"""
workflows_collection = get_workflows_collection()
query = {"title": title}
if department_id:
query["department_id"] = ObjectId(department_id)
workflow_data = workflows_collection.find_one(query)
if workflow_data:
return cls.from_dict(workflow_data)
return None
@classmethod
def find_by_department(cls, department_id):
"""Find all workflows for a department"""
workflows_collection = get_workflows_collection()
workflows_data = workflows_collection.find({"department_id": ObjectId(department_id)})
return [cls.from_dict(workflow_data) for workflow_data in workflows_data]
@classmethod
def get_all(cls):
"""Get all workflows"""
workflows_collection = get_workflows_collection()
workflows_data = workflows_collection.find()
return [cls.from_dict(workflow_data) for workflow_data in workflows_data]
def delete(self):
"""Delete workflow from database"""
if not self._id:
return False
workflows_collection = get_workflows_collection()
result = workflows_collection.delete_one({"_id": ObjectId(self._id)})
return result.deleted_count > 0
def add_form(self, template_data):
"""Add a form template to workflow - for backward compatibility"""
if isinstance(template_data, dict) and 'markdown_template' in template_data:
# Extract the markdown content and store directly
self.markdown_template = template_data['markdown_template']
self.template_name = template_data.get('filename', '')
return self.save()
# For backward compatibility, if it's just a string
elif isinstance(template_data, str):
self.raw_forms.append(template_data)
return self.save()
return False
def remove_form(self, form_path):
"""Remove a form from workflow by path (legacy method)"""
if form_path in self.raw_forms:
self.raw_forms.remove(form_path)
return self.save()
return True
def remove_form_by_index(self, index):
"""Remove a form from workflow by index (legacy method)"""
if 0 <= index < len(self.raw_forms):
self.raw_forms.pop(index)
return self.save()
return False
def add_data_requirement(self, field, description):
"""Add a data requirement (field and description)"""
requirement = (field, description)
if requirement not in self.data_requirements:
self.data_requirements.append(requirement)
return self.save()
return True
def add_form_field(self, position, field_name):
"""Add a form field (position and field name)"""
form_field = (position, field_name)
if form_field not in self.form_fields:
self.form_fields.append(form_field)
return self.save()
return True
def clear_template(self):
"""Clear the markdown template"""
self.markdown_template = None
self.template_name = None
return self.save() |