in microservices/lti/src/routes/line_item.py [0:0]
def create_score_for_line_item(context_id: str,
line_item_id: str,
input_score: BasicScoreModel,
token: auth_scheme = Depends()):
"""The create score for line item endpoint will add a score for a line item
to the firestore.
### Args:
input_score: `BasicScoreModel`
Input line item to be inserted
### Raises:
ResourceNotFoundException:
If the line item does not exist <br/>
Exception:
Internal Server Error. Raised if something went wrong
### Returns:
Result Data: `ResultResponseModel`
"""
try:
line_item = LineItem.find_by_id(line_item_id)
if line_item.contextId != context_id:
raise ResourceNotFoundException(
f"Line item with id {line_item_id} in {context_id} not found")
input_score_dict = {**input_score.dict()}
if input_score_dict.get("scoreGiven") is not None:
if input_score_dict.get("scoreMaximum") is not None:
if input_score_dict["scoreGiven"] > input_score_dict["scoreMaximum"]:
raise ValidationError(
"Score maximum should not be greater than the given score")
else:
raise ValidationError("Score maximum should not be a null value")
input_score_dict["lineItemId"] = line_item_id
new_score = Score()
new_score = new_score.from_dict(input_score_dict)
new_score.save()
line_item_url = f"{LTI_ISSUER_DOMAIN}/lti/api/v1/{context_id}/line_items/{line_item_id}"
input_result_dict = {
"userId": input_score_dict["userId"],
"resultScore": input_score_dict["scoreGiven"],
"resultMaximum": input_score_dict["scoreMaximum"],
"comment": input_score_dict["comment"],
"scoreOf": line_item_id,
"lineItemId": line_item_id,
"isGradeSyncCompleted": False
}
user_id = input_result_dict["userId"]
result = Result.collection.filter("scoreOf", "==", line_item_id).filter(
"userId", "==", input_score_dict["userId"]).get()
if result:
result_fields = result.get_fields()
for key, value in input_result_dict.items():
result_fields[key] = value
for key, value in result_fields.items():
setattr(result, key, value)
result.update()
result_fields = result.get_fields(reformat_datetime=True)
result_fields["scoreOf"] = line_item_url
result_fields[
"id"] = f"{LTI_ISSUER_DOMAIN}/lti/api/v1/{context_id}/line_items/{line_item_id}/results/{result.id}"
else:
new_result = Result()
new_result = new_result.from_dict(input_result_dict)
new_result.save()
result_fields = new_result.get_fields(reformat_datetime=True)
result_fields["scoreOf"] = line_item_url
result_fields[
"id"] = f"{LTI_ISSUER_DOMAIN}/lti/api/v1/{context_id}/line_items/{line_item_id}/results/{new_result.id}"
# Passing grades back to the LMS using shim service API
if line_item.resourceLinkId:
# check if user is present in the context
section_user_mapping_status_url = f"http://classroom-shim/classroom-shim/api/v1/context/{context_id}/user/{user_id}"
check_section_user_mapping_req = requests.get(
url=section_user_mapping_status_url,
headers={"Authorization": f"Bearer {auth_client.get_id_token()}"},
timeout=60)
lti_content_item = LTIContentItem.find_by_id(line_item.resourceLinkId)
tool = Tool.find_by_id(lti_content_item.tool_id)
user_exception = UserGradeException.find_by_user_and_tool_id(
user_id=user_id, tool_id=lti_content_item.tool_id)
if user_exception and user_exception.allow_exception is True:
error_message = f"Skipped grade passback due to manual exception for user with id {user_id} and context {context_id}. API request payload --- {str(input_score)} ---"
Logger.error(error_message)
else:
if check_section_user_mapping_req.status_code == 200:
input_grade_dict = {
"user_id": user_id,
"comment": input_result_dict["comment"],
"lti_content_item_id": line_item.resourceLinkId,
"maximum_grade": input_result_dict["resultMaximum"],
"assigned_grade": None,
"draft_grade": None,
"validate_title": tool.validate_title_for_grade_sync,
"line_item_title": line_item.label
}
if input_score_dict["gradingProgress"] in [
"Pending", "PendingManual"
]:
input_grade_dict["draft_grade"] = input_result_dict["resultScore"]
if input_score_dict["gradingProgress"] == "FullyGraded":
input_grade_dict["assigned_grade"] = input_result_dict[
"resultScore"]
if input_grade_dict.get(
"draft_grade") is not None or input_grade_dict.get(
"assigned_grade") is not None:
Logger.info(
f"Processing grade passback for context/{context_id}/line_items/{line_item_id} with api request payload --- {str(input_score)} ---"
)
gpb_resp = grade_pass_back(input_grade_dict, user_id, line_item_id)
if gpb_resp:
result = Result.collection.filter(
"scoreOf", "==",
line_item_id).filter("userId", "==",
input_score_dict["userId"]).get()
result.isGradeSyncCompleted = True
result.update()
else:
error_message = f"Failed to passback grade. User with id {user_id} not found in context {context_id}. API request payload --- {str(input_score)} ---"
Logger.error(error_message)
raise Exception(error_message)
else:
Logger.error(
f"Content item id not found for given line item {line_item_id} to trigger grade passback"
)
return result_fields
except InvalidTokenError as e:
Logger.error(e)
raise Unauthenticated(str(e)) from e
except ResourceNotFoundException as e:
Logger.error(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