in microservices/assessment_service/src/routes/assessment.py [0:0]
def create_human_graded_assessment(input_assessment:HumanGradedAssessmentModel):
"""
The create assessment endpoint will add the assessment to the
firestore if it does not exist. If the assessment exist then it will
update the assessment.
Additionally, for these human graded assessments it creates the rubrics
and rubric criterion and established the parent child relationship
### Args:
- input_assessment (HumanGradedAssessmentModel): input human graded
assessment to be inserted
### Raises:
- ResourceNotFoundException: If the assessment does not exist
- Exception: 500 Internal Server Error if something went wrong
### Returns:
- AssessmentModel: Created Assessment Object
"""
try:
input_assessment_dict = {**input_assessment.dict()}
assessment_type = input_assessment_dict.get("type", None)
if assessment_type is None:
raise ValidationError("Assessment `type` field is mandatory")
resource_paths = input_assessment_dict.get("resource_paths")
author_id = input_assessment_dict.get("author_id")
rubrics_list = input_assessment_dict.get("child_nodes", {})\
.get("rubrics", [])
if assessment_type == "project" and not rubrics_list:
raise ValidationError(
"Assessment of type `project` should have rubric criteria")
performance_indicators = []
if rubrics_list:
# Creation of rubric_criteria
# Parse the dictionary to create the rubric criterion(leaf_node)
for rubric_index in range(0,len(rubrics_list)):
rubric = rubrics_list[rubric_index]
# Validating the schema of rubric data model
RubricModel(**rubric)
rubric_criteria_list = rubric["child_nodes"]["rubric_criteria"]
for index in range(0,len(rubric_criteria_list)):
rc_data = {**rubric_criteria_list[index]}
performance_indicators.extend(rc_data.get("performance_indicators",
[]))
rubric_criteria_create_req =\
BasicRubricCriterionModel(**rubric_criteria_list[index])
rubric_create_response = \
create_rubric_criterion(rubric_criteria_create_req)
#Replacing the rubric dictionary with child_nodes uuid
rubric["child_nodes"]["rubric_criteria"][index] =\
rubric_create_response["data"]["uuid"]
rubric_create_req = RubricModel(**rubric)
rubric_create_response = create_rubric(rubric_create_req)
rubrics_list[rubric_index] = rubric_create_response["data"]["uuid"]
# Updating the input_assessment_request with child_node uuids
input_assessment_dict["child_nodes"]["rubrics"] = rubrics_list
ParentChildNodesHandler.validate_parent_child_nodes_references(
input_assessment_dict)
if assessment_type == "project":
input_assessment_dict["max_attempts"] = 3
else:
input_assessment_dict["max_attempts"] = 1
new_assessment = Assessment()
new_assessment = new_assessment.from_dict(input_assessment_dict)
new_assessment.uuid = ""
# Validate if the skill ID exists in Firestore DB
for skill_id in performance_indicators:
_ = Skill.find_by_uuid(skill_id)
references = new_assessment.references
references["skills"] = performance_indicators
new_assessment.references = references
new_assessment.save()
new_assessment.uuid = new_assessment.id
if resource_paths and author_id:
updated_resource_paths = attach_files_to_assessment(author_id,
new_assessment.id, resource_paths, CONTENT_SERVING_BUCKET)
new_assessment.resource_paths = updated_resource_paths
new_assessment.update()
assessment_fields = new_assessment.get_fields(reformat_datetime=True)
ParentChildNodesHandler.update_child_references(
assessment_fields, Assessment, operation="add")
ParentChildNodesHandler.update_parent_references(
assessment_fields, Assessment, operation="add")
return {
"success": True,
"message": "Successfully created the assessment",
"data": assessment_fields
}
except ResourceNotFoundException as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise ResourceNotFound(str(e)) from e
except ValidationError as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise BadRequest(str(e), data=e.data) from e
except Exception as e:
Logger.error(e)
Logger.error(traceback.print_exc())
raise InternalServerError(str(e)) from e