| from enum import Enum
|
| from fastapi import Query, HTTPException
|
| from pymongo import DESCENDING
|
|
|
| class WorkflowStatus(str, Enum):
|
| Pending = "Pending"
|
| Accepted = "Accepted"
|
| Rejected = "Rejected"
|
| Draft = "Draft"
|
| Approved = "Approved"
|
|
|
| @router_workflow.get("/workflow")
|
| async def get_workflows(
|
| user: Users = Depends(get_current_user),
|
| paginate_params: tuple = Depends(paginate_query_params),
|
| search: str = Query(None, description="workflow_name"),
|
| start_date: date = Query(None, description="Start date"),
|
| end_date: date = Query(None, description="End date"),
|
| status: WorkflowStatus = Query(None, description="Status filter"),
|
| check_user_status: Users = Depends(check_user_status)
|
| ):
|
| skip, page_size = paginate_params
|
| filter_dict = {}
|
|
|
| if search:
|
| search = search.strip()
|
| filter_dict["$or"] = [
|
| {"workflow_name": {"$regex": f".*{re.escape(search)}.*", "$options": "i"}},
|
| {"forms.name": {"$regex": f".*{re.escape(search)}.*", "$options": "i"}},
|
| {"forms.elements.itemData.name": {"$regex": f".*{re.escape(search)}.*", "$options": "i"}},
|
| {"created_by": {"$regex": f".*{re.escape(search)}.*", "$options": "i"}},
|
| ]
|
|
|
| if start_date:
|
| start_datetime = datetime.combine(start_date, datetime.min.time())
|
| filter_dict["created_on"] = {"$gte": start_datetime}
|
|
|
| if end_date:
|
| end_datetime = datetime.combine(end_date, datetime.max.time())
|
| if "created_on" not in filter_dict:
|
| filter_dict["created_on"] = {}
|
| filter_dict["created_on"]["$lte"] = end_datetime
|
|
|
| if status:
|
| filter_dict["forms.status"] = status.value
|
|
|
| u_id = str(user["_id"]) # Assuming user id is stored as string in the database
|
|
|
| if verify_permissions(user, [Codes.REVIEW_WORKFLOW, Codes.VIEW_WORKFLOW_LIST]):
|
| print("User has permission to view workflow list")
|
| workflows_list = db[WORKFLOW_COLLECTION].find(filter_dict).sort("_id", DESCENDING).skip(skip).limit(page_size)
|
| filtered_workflows = filter_workflows_by_user(workflows_list, u_id)
|
| elif verify_permissions(user, [Codes.ALL, Codes.VIEW_WORKFLOW_LIST], False):
|
| workflows_list = db[WORKFLOW_COLLECTION].find(filter_dict).sort("_id", DESCENDING).skip(skip).limit(page_size)
|
| filtered_workflows = list_workflow_serial(workflows_list)
|
| else:
|
| raise HTTPException(status_code=403, detail="User does not have permission to access this resource")
|
|
|
| total_count = db[WORKFLOW_COLLECTION].count_documents(filter_dict)
|
|
|
| return {"total_count": total_count, "workflows": filtered_workflows}
|
|
|
| def filter_workflows_by_user(workflows_list, user_id):
|
| filtered_workflows = []
|
| for workflow in workflows_list:
|
| filtered_forms = []
|
| for form in workflow["forms"]:
|
| if str(form.get("assign_user")) == user_id:
|
| filtered_forms.append(form)
|
| if filtered_forms:
|
| workflow["forms"] = filtered_forms
|
| filtered_workflows.append(workflow)
|
| return list_workflow_serial(filtered_workflows)
|