from __future__ import annotations __copyright__ = "Copyright (C) 2017 Andreas Kloeckner" __license__ = """ Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ from typing import TYPE_CHECKING, Any, Text # noqa from django import http from django.core.exceptions import PermissionDenied from django.shortcuts import get_object_or_404 from course.auth import APIError, with_course_api_auth from course.constants import participation_permission as pperm from course.models import FlowSession # {{{ mypy if TYPE_CHECKING: from course.auth import APIContext # }}} def flow_session_to_json(sess: FlowSession) -> Any: last_activity = sess.last_activity() return { "id": sess.id, "participation_username": ( sess.participation.user.username if sess.participation is not None else None), "participation_institutional_id": ( sess.participation.user.institutional_id if sess.participation is not None else None), "active_git_commit_sha": sess.active_git_commit_sha, "flow_id": sess.flow_id, "start_time": sess.start_time.isoformat(), "completion_time": sess.completion_time, "last_activity_time": ( last_activity.isoformat() if last_activity is not None else None), "page_count": sess.page_count, "in_progress": sess.in_progress, "access_rules_tag": sess.access_rules_tag, "expiration_mode": sess.expiration_mode, "points": sess.points, "max_points": sess.max_points, "result_comment": sess.result_comment, } @with_course_api_auth("Token") def get_flow_sessions( api_ctx: APIContext, course_identifier: str) -> http.HttpResponse: if not api_ctx.has_permission(pperm.view_gradebook): raise PermissionDenied("token role does not have required permissions") try: flow_id = api_ctx.request.GET["flow_id"] except KeyError: raise APIError("must specify flow_id GET parameter") sessions = FlowSession.objects.filter( course=api_ctx.course, flow_id=flow_id) result = [flow_session_to_json(sess) for sess in sessions] return http.JsonResponse(result, safe=False) @with_course_api_auth("Token") def get_flow_session_content( api_ctx: APIContext, course_identifier: str) -> http.HttpResponse: if not api_ctx.has_permission(pperm.view_gradebook): raise PermissionDenied("token role does not have required permissions") try: session_id_str = api_ctx.request.GET["flow_session_id"] except KeyError: raise APIError("must specify flow_id GET parameter") session_id = int(session_id_str) flow_session = get_object_or_404(FlowSession, id=session_id) if flow_session.course != api_ctx.course: raise PermissionDenied( "session's course does not match auth context") from course.content import get_course_repo from course.flow import adjust_flow_session_page_data, assemble_answer_visits with get_course_repo(api_ctx.course) as repo: from course.utils import FlowContext, instantiate_flow_page_with_ctx fctx = FlowContext(repo, api_ctx.course, flow_session.flow_id) adjust_flow_session_page_data(repo, flow_session, api_ctx.course.identifier, fctx.flow_desc) from course.flow import get_all_page_data all_page_data = get_all_page_data(flow_session) answer_visits = assemble_answer_visits(flow_session) pages = [] for i, page_data in enumerate(all_page_data): page = instantiate_flow_page_with_ctx(fctx, page_data) assert i == page_data.page_ordinal page_data_json = { "ordinal": i, "page_type": page_data.page_type, "group_id": page_data.group_id, "page_id": page_data.page_id, "page_data": page_data.data, "title": page_data.title, "bookmarked": page_data.bookmarked, } answer_json = None grade_json = None visit = answer_visits[i] if visit is not None: from course.page.base import PageContext pctx = PageContext(api_ctx.course, repo, fctx.course_commit_sha, flow_session) norm_bytes_answer_tup = page.normalized_bytes_answer( pctx, page_data.data, visit.answer) # norm_answer needs to be JSON-encodable norm_answer: Any = None if norm_bytes_answer_tup is not None: answer_file_ext, norm_bytes_answer = norm_bytes_answer_tup if answer_file_ext in [".txt", ".py"]: norm_answer = norm_bytes_answer.decode("utf-8") elif answer_file_ext == ".json": import json norm_answer = json.loads(norm_bytes_answer) else: from base64 import b64encode norm_answer = [answer_file_ext, b64encode(norm_bytes_answer).decode("utf-8")] answer_json = { "visit_time": visit.visit_time.isoformat(), "remote_address": repr(visit.remote_address), "user": ( visit.user.username if visit.user is not None else None), "impersonated_by": ( visit.impersonated_by.username if visit.impersonated_by is not None else None), "is_synthetic_visit": visit.is_synthetic, "answer_data": visit.answer, "answer": norm_answer, } grade = visit.get_most_recent_grade() if grade is not None: grade_json = { "grader": (grade.grader.username if grade.grader is not None else None), "grade_time": grade.grade_time.isoformat(), "graded_at_git_commit_sha": ( grade.graded_at_git_commit_sha), "max_points": grade.max_points, "correctness": grade.correctness, "feedback": grade.feedback} pages.append({ "page": page_data_json, "answer": answer_json, "grade": grade_json, }) result = { "session": flow_session_to_json(flow_session), "pages": pages, } return http.JsonResponse(result, safe=False) # vim: foldmethod=marker