in microservices/assessment_service/src/routes/assessment.py [0:0]
def get_assessments(skip: int = Query(0, ge=0, le=2000),
limit: int = Query(10, ge=1, le=100),
is_archived: Optional[bool] = False,
is_autogradable: Optional[bool] = False,
performance_indicators: Union[List[str], None] = \
Query(default=None),
sort_by: Optional[str] = "created_time",
sort_order: Optional[Literal["ascending", "descending"]] =
"descending",
search: Optional[str] = None):
"""
The get assessments endpoint will return an array assessments
from firestore
### Args:
- skip (int): Number of objects to be skipped
- limit (int): Size of assessment array to be returned
- sort_by (str): Data Model Fields name
- sort_order (str): ascending/descending
- search_query (str): search string
### Raises:
- ValidationError: If request validation fails
- Exception: 500 Internal Server Error if something went wrong
### Returns:
- AllAssessmentsModelResponse: Array of Assessment Object
"""
try:
collection_manager = Assessment.collection
fetch_length = skip + limit
collection_manager = collection_manager.filter(
"is_autogradable", "==", is_autogradable)
collection_manager = collection_manager.filter(
"is_archived", "==", is_archived)
data = collection_manager.order(f"{sort_by}").fetch() \
if sort_order == "ascending" else collection_manager.order(
f"-{sort_by}").fetch()
assessments = [i.get_fields(reformat_datetime=True) for i in data]
filtered_assessments = [] if search else assessments
if search and len(search):
for assessment in assessments:
if search.lower() in assessment["name"].lower():
filtered_assessments.append(assessment)
elif search.lower() in assessment["type"].lower():
filtered_assessments.append(assessment)
elif assessment.get("references", {}) and \
search.lower() in assessment["references"].get("competencies", []):
filtered_assessments.append(assessment)
elif assessment.get("references", {}) and \
search.lower() in assessment["references"].get("skills", []):
filtered_assessments.append(assessment)
filtered_assessments = [CollectionHandler.loads_field_data_from_collection(
i) for i in filtered_assessments]
final_filtered_assessments = []
for assessment in filtered_assessments:
modified_competencies = []
modified_skills = []
performance_indicator_overlap = not bool(performance_indicators)
if assessment.get("references", {}) and assessment.get("references", \
{}).get("skills", []):
for skill in assessment["references"]["skills"]:
if isinstance(skill, dict):
modified_skill = {"uuid": skill.get("uuid"),
"name": skill.get("name")}
modified_skills.append(modified_skill)
if performance_indicators and\
skill.get("uuid") in performance_indicators:
performance_indicator_overlap = True
assessment["references"]["skills"] = modified_skills
if assessment.get("references", {}) and assessment.get("references", \
{}).get("competencies", []) and \
performance_indicator_overlap:
for competency in assessment["references"]["competencies"]:
if isinstance(competency, dict):
modified_competency = {
"uuid": competency.get("uuid"),
"name": competency.get("name")
}
modified_competencies.append(modified_competency)
assessment["references"]["competencies"] = modified_competencies
if performance_indicator_overlap:
final_filtered_assessments.append(assessment)
count = 10000
response = {
"records": final_filtered_assessments[skip:fetch_length],
"total_count": count
}
return {
"success": True,
"message": "Successfully fetched the assessments",
"data": response
}
except ValidationError as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise BadRequest(str(e)) from e
except ResourceNotFoundException as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise ResourceNotFound(str(e)) from e
except Exception as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise InternalServerError(str(e)) from e