def approved_exp_keyword_search()

in microservices/prior_learning_assessment/src/routes/approved_experience.py [0:0]

• 75 lines of code
• 33 McCabe index (conditional complexity)

def approved_exp_keyword_search(
    keyword: str,
    skip: int = Query(0, ge=0, le=2000),
    limit: int = Query(6, ge=1, le=100),
    type: Union[List[str], None] = Query(default=None),
    student_type: Union[List[str], None] = Query(default=None),
    class_level: Union[List[str], None] = Query(default=None),
    status: Union[List[str], None] = Query(default=None),
    organization: Union[List[str], None] = Query(default=None),
    credits_range_lower_limit: Optional[int] = None,
    credits_range_upper_limit: Optional[int] = None,
    sort_by:
      Optional[Literal["total_credit", "experience_name"]] = "total_credit",
    sort_order: Optional[Literal["ascending", "descending"]] = "descending",
  ):
  """
  Search for approved experience title and description
  using the provided keyword
  Args:
    keyword (str): Required text for search
    skip (int): Number of objects to be skipped
    limit (int): Size of skill array to be returned
    type (list): type to filter approved_experiences
    student_type (list): student_type to filter approved_experiences
    class_level (list): class_level to filter approved_experiences
    status (list): status to filter approved_experiences
    organizations (list): organizations to filter approved_experiences
    credits_range_lower_limit (int):
        lower limit of credits to filter approved_experiences
    credits_range_upper_limit (int):
        upper limit of credits to filter approved_experiences
    sort_by (str): field on which sorting would be performed. Values can be
        total_credit or experience_name. Default is total_credit.
    sort_order (str): sorting order for the sort_by field. Default is
        descending.
  Returns:
    AllApprovedExperienceResponseModel : Array of matched Keywords Object
  """
  try:
    keyword = re.escape(keyword)
    collection_manager = ApprovedExperience.collection
    if type is not None:
      collection_manager = collection_manager.filter("type", "in", type)
    if student_type is not None:
      collection_manager = collection_manager.filter("student_type", "in",
                                                     student_type)
    if class_level is not None:
      collection_manager = collection_manager.filter("class_level", "in",
                                                     class_level)
    if status is not None:
      collection_manager = collection_manager.filter("status", "in", status)

    ae_docs = list(collection_manager.fetch())
    ae_dicts = [doc.get_fields(reformat_datetime=True) for doc in ae_docs]
    if organization is not None:
      ae_dicts = [doc for doc in ae_dicts \
        if doc.get("organization", None) in organization]
    if credits_range_lower_limit is not None:
      ae_dicts = [doc for doc in ae_dicts \
        if doc.get("credits_range", {}).get("lower_limit", None) and \
          doc["credits_range"]["lower_limit"] >= credits_range_lower_limit]
    if credits_range_upper_limit is not None:
      ae_dicts = [doc for doc in ae_dicts \
        if doc.get("credits_range", {}).get("upper_limit", None) and \
          doc["credits_range"]["upper_limit"] <= credits_range_upper_limit]
    search_result = []
    for doc in ae_dicts:
      if doc.get("description", None) and \
          len(re.findall(keyword.lower(), doc["description"].lower())) > 0:
        search_result.append(doc)
      elif doc.get("title", None) and \
          len(re.findall(keyword.lower(), doc["title"].lower())) > 0:
        search_result.append(doc)

    if sort_by == "total_credit" and sort_order == "descending":
      search_result = sorted(search_result,
        key=lambda d: d["credits_range"]["upper_limit"], reverse=True)
    elif sort_by == "total_credit" and sort_order == "ascending":
      search_result = sorted(search_result,
        key=lambda d: d["credits_range"]["upper_limit"])
    elif sort_by == "experience_name" and sort_order == "descending":
      search_result = sorted(search_result,
        key=lambda d: d["title"], reverse=True)
    elif sort_by == "experience_name" and sort_order == "ascending":
      search_result = sorted(search_result, key=lambda d: d["title"])

    count = len(search_result)
    response = {"records": search_result[skip:skip+limit], "total_count": count}

    return {
      "success": True,
      "message": "Successfully fetched the Approved Experiences",
      "data": response
    }
  except ValidationError as e:
    raise BadRequest(str(e)) from e
  except ResourceNotFoundException as e:
    raise ResourceNotFound(str(e)) from e
  except Exception as e:
    Logger.error(e)
    Logger.error(traceback.print_exc())
    raise InternalServerError(str(e)) from e