def search_pla_record()

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

• 89 lines of code
• 45 McCabe index (conditional complexity)

def search_pla_record(
  skip: int = Query(0, ge=0, le=2000),
  limit: int = Query(6, ge=1, le=100),
  keyword: Optional[str] = None,
  start_date: Optional[str] = None,
  end_date: Optional[str] = None,
  status: Union[List[str], None] = Query(default=None),
  assessor: Union[List[str], None] = Query(default=None),
  sort_by:
      Optional[Literal["title", "date_last_edited", "ID Number"]] = "title",
  sort_order: Optional[Literal["ascending", "descending"]] = "ascending",
  ):
  """
  Search for pla records/sessions title and description using the provided
  keyword. If keyword is not passed, it will return all the records.
  Args:
    skip (int): Number of objects to be skipped
    limit (int): Size of skill array to be returned
    keyword (str): Text for search
    status (list): status to filter pla_records
    assessor_name (list): assessors to filter pla_records
    start_date (str): lower limit og save date to filter pla_records
    end_date (str): lower limit og save date to filter pla_records
    sort_by (str): field on which sorting would be performed. Values can be
        title or date_last_edited ot ID Number. Default is title.
    sort_order (str): sorting order for the sort_by field. Default is
        ascending.
  Returns:
    PLARecordSearchModelResponse : Array of matched Keywords Object along
    with the count of the total search results
  """
  try:
    collection_manager = PLARecord.collection.filter("type", "==", "saved")
    if assessor is not None:
      collection_manager = collection_manager.filter("assessor_name", "in",
                                                     assessor)
    pla_docs = list(collection_manager.fetch())
    pla_dicts = [doc.get_fields(reformat_datetime=True) for doc in pla_docs]
    if start_date is not None:
      pla_dicts = [doc for doc in pla_dicts \
        if doc.get("last_modified_time", None) and \
          doc.get("last_modified_time", None) >= start_date]
    if end_date is not None:
      pla_dicts = [doc for doc in pla_dicts \
        if doc.get("last_modified_time", None) and \
          doc.get("last_modified_time", None) <= end_date]
    if status is not None:
      pla_dicts_status = []
      pla_dicts_flagged = []
      pla_dicts_archived = []
      if "Archived" in status:
        pla_dicts_archived = [doc for doc in pla_dicts \
                              if doc.get("is_archived")]
        status.remove("Archived")
      if "Flagged" in status:
        pla_dicts_flagged = [doc for doc in pla_dicts if doc.get("is_flagged") \
                             if not doc.get("is_archived")]
        status.remove("Flagged")
      if status is not None:
        pla_dicts_status = [doc for doc in pla_dicts \
                            if doc.get("status", None) in status
                            if not doc.get("is_archived")
                            if not doc.get("is_flagged")]
      pla_dicts = pla_dicts_flagged + pla_dicts_archived + pla_dicts_status

    if keyword is not None:
      search_result = []
      keyword = re.escape(keyword)
      for doc in pla_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)
    else:
      search_result = pla_dicts

    if sort_by == "title" and sort_order == "descending":
      search_result = sorted(search_result,
        key=lambda d: d["title"], reverse=True)
    elif sort_by == "title" and sort_order == "ascending":
      search_result = sorted(search_result,
        key=lambda d: d["title"])
    elif sort_by == "ID Number" and sort_order == "descending":
      search_result = sorted(search_result,
        key=lambda d: d["id_number"], reverse=True)
    elif sort_by == "ID Number" and sort_order == "ascending":
      search_result = sorted(search_result, key=lambda d: d["id_number"])
    elif sort_by == "date_last_edited" and sort_order == "descending":
      search_result = sorted(search_result, key=\
                             lambda d: d["last_modified_time"],reverse=True)
    elif sort_by == "date_last_edited" and sort_order == "ascending":
      search_result = sorted(search_result, key=\
                             lambda d: d["last_modified_time"])

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

    return {
      "success": True,
      "message": "Successfully fetched the PLA Records",
      "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