New paste Repaste Download
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)
Filename: None. Size: 3kb. View raw, , hex, or download this file.

This paste expires on 2024-04-03 06:59:38.135351. Pasted through web.