Source code for codegrade._api.submission

"""The endpoints for submission objects.

SPDX-License-Identifier: AGPL-3.0-only OR BSD-3-Clause-Clear
"""

import os
import typing as t

import cg_request_args as rqa
from cg_maybe import Maybe, Nothing

from .. import parsers, utils

if t.TYPE_CHECKING or os.getenv("CG_EAGERIMPORT", False):
    from .. import client
    from ..models.create_proxy_submission_data import CreateProxySubmissionData
    from ..models.extended_work import ExtendedWork
    from ..models.feedback_with_replies import FeedbackWithReplies
    from ..models.feedback_without_replies import FeedbackWithoutReplies
    from ..models.grade_history import GradeHistory
    from ..models.mirror_file_result import MirrorFileResult
    from ..models.patch_grader_submission_data import PatchGraderSubmissionData
    from ..models.patch_rubric_result_submission_data import (
        PatchRubricResultSubmissionData,
    )
    from ..models.patch_submission_data import PatchSubmissionData
    from ..models.proxy import Proxy
    from ..models.root_file_trees_json import RootFileTreesJSON
    from ..models.work_rubric_result_as_json import WorkRubricResultAsJSON


_ClientT = t.TypeVar("_ClientT", bound="client._BaseClient")


[docs]class SubmissionService(t.Generic[_ClientT]): __slots__ = ("__client",) def __init__(self, client: _ClientT) -> None: self.__client = client
[docs] def create_proxy( self: "SubmissionService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "CreateProxySubmissionData"], *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "Proxy": """Create a proxy to view the files of the given submission through. This allows you to view files of a submission without authentication for a limited time. :param json_body: The body of the request. See :class:`.CreateProxySubmissionData` for information about the possible fields. You can provide this data as a :class:`.CreateProxySubmissionData` or as a dictionary. :param submission_id: The submission for which the proxy should be created. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The created proxy. """ url = "/api/v1/submissions/{submissionId}/proxy".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.post( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.proxy import Proxy return parsers.JsonResponseParser( parsers.ParserFor.make(Proxy) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def get( self: "SubmissionService[client.AuthenticatedClient]", *, submission_id: "int", type: "t.Literal['default', 'feedback', 'zip']" = "default", owner: "t.Literal['auto', 'student', 'teacher']" = "auto", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Union[ExtendedWork, MirrorFileResult]": """Get the given submission (also called work) by id. :param submission_id: The id of the submission :param type: If passed this cause you not to receive a submission object. What you will receive will depend on the value passed. If you pass `zip` you will receive a zip file with all the files of the submission. If you pass `feedback` you will receive a text file with a textual representation of all the feedback given on this submission. :param owner: This query parameter is only used when `type=='zip'`. It will determine which revision is used to generate the zip file. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The requested submission, or one of the other types as requested by the `type` query parameter. """ url = "/api/v1/submissions/{submissionId}".format( submissionId=submission_id ) params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "type": utils.to_dict(type), "owner": utils.to_dict(owner), } with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.extended_work import ExtendedWork from ..models.mirror_file_result import MirrorFileResult return parsers.JsonResponseParser( parsers.make_union( parsers.ParserFor.make(ExtendedWork), parsers.ParserFor.make(MirrorFileResult), ) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def delete( self, *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "None": """Delete a submission and all its files. :param submission_id: The submission to delete. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: Nothing """ url = "/api/v1/submissions/{submissionId}".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.delete(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 204): return parsers.ConstantlyParser(None).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def patch( self: "SubmissionService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchSubmissionData"], *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "ExtendedWork": """Update the given submission with new values. :param json_body: The body of the request. See :class:`.PatchSubmissionData` for information about the possible fields. You can provide this data as a :class:`.PatchSubmissionData` or as a dictionary. :param submission_id: The id of the submission. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The updated submission. """ url = "/api/v1/submissions/{submissionId}".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.patch( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.extended_work import ExtendedWork return parsers.JsonResponseParser( parsers.ParserFor.make(ExtendedWork) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def delete_grader( self, *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "None": """Change the assigned grader of the given submission. :param submission_id: The id of the submission. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: Empty response and a 204 status. """ url = "/api/v1/submissions/{submissionId}/grader".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.delete(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 204): return parsers.ConstantlyParser(None).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def patch_grader( self: "SubmissionService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchGraderSubmissionData"], *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "None": """Change the assigned grader of the given submission. :param json_body: The body of the request. See :class:`.PatchGraderSubmissionData` for information about the possible fields. You can provide this data as a :class:`.PatchGraderSubmissionData` or as a dictionary. :param submission_id: The id of the submission. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: Empty response and a 204 status. """ url = "/api/v1/submissions/{submissionId}/grader".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.patch( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 204): return parsers.ConstantlyParser(None).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def get_feedback( self: "SubmissionService[client.AuthenticatedClient]", *, submission_id: "int", with_replies: "bool" = False, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Union[FeedbackWithReplies, FeedbackWithoutReplies]": """Get all feedback for a submission :param submission_id: The submission of which you want to get the feedback. :param with_replies: Do you want to include replies in with your comments? Starting with version "O.1" the default value will change to `True`. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The feedback of this submission. """ url = "/api/v1/submissions/{submissionId}/feedbacks/".format( submissionId=submission_id ) params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "with_replies": utils.to_dict(with_replies), } with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.feedback_with_replies import FeedbackWithReplies from ..models.feedback_without_replies import ( FeedbackWithoutReplies, ) return parsers.JsonResponseParser( parsers.make_union( parsers.ParserFor.make(FeedbackWithReplies), parsers.ParserFor.make(FeedbackWithoutReplies), ) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def get_grade_history( self: "SubmissionService[client.AuthenticatedClient]", *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "t.Sequence[GradeHistory]": """Get the grade history for the given submission. :param submission_id: The submission for which you want to get the grade history. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: All the `GradeHistory` objects, which describe the history of this grade. """ url = "/api/v1/submissions/{submissionId}/grade_history/".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.grade_history import GradeHistory return parsers.JsonResponseParser( rqa.List(parsers.ParserFor.make(GradeHistory)) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def get_root_file_trees( self: "SubmissionService[client.AuthenticatedClient]", *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "RootFileTreesJSON": """Get all the file trees of a submission. :param submission_id: The id of the submission of which you want to get the file trees. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The student and teacher file tree, from the base/root directory of the submission. """ url = "/api/v1/submissions/{submissionId}/root_file_trees/".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.root_file_trees_json import RootFileTreesJSON return parsers.JsonResponseParser( parsers.ParserFor.make(RootFileTreesJSON) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def get_rubric_result( self: "SubmissionService[client.AuthenticatedClient]", *, submission_id: "int", extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "WorkRubricResultAsJSON": """Get the full rubric result of the given submission (work). :param submission_id: The id of the submission :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The rubric result of the given submission, which also contains the rubric. """ url = "/api/v1/submissions/{submissionId}/rubrics/".format( submissionId=submission_id ) params = extra_parameters or {} with self.__client as client: resp = client.http.get(url=url, params=params) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.work_rubric_result_as_json import ( WorkRubricResultAsJSON, ) return parsers.JsonResponseParser( parsers.ParserFor.make(WorkRubricResultAsJSON) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )
[docs] def patch_rubric_result( self: "SubmissionService[client.AuthenticatedClient]", json_body: t.Union[dict, list, "PatchRubricResultSubmissionData"], *, submission_id: "int", copy_locked_items: "bool" = False, extra_parameters: t.Optional[ t.Mapping[str, t.Union[str, bool, int, float]] ] = None, ) -> "ExtendedWork": """Select the given rubric items for the given submission. :param json_body: The body of the request. See :class:`.PatchRubricResultSubmissionData` for information about the possible fields. You can provide this data as a :class:`.PatchRubricResultSubmissionData` or as a dictionary. :param submission_id: The submission to unselect the item for. :param copy_locked_items: Should we maintain the selected items in locked rubric rows. :param extra_parameters: The extra query parameters you might want to add. By default no extra query parameters are added. :returns: The work of which you updated the rubric items. """ url = "/api/v1/submissions/{submissionId}/rubricitems/".format( submissionId=submission_id ) params: t.Dict[str, t.Any] = { **(extra_parameters or {}), "copy_locked_items": utils.to_dict(copy_locked_items), } with self.__client as client: resp = client.http.patch( url=url, json=utils.to_dict(json_body), params=params ) utils.log_warnings(resp) if utils.response_code_matches(resp.status_code, 200): from ..models.extended_work import ExtendedWork return parsers.JsonResponseParser( parsers.ParserFor.make(ExtendedWork) ).try_parse(resp) from ..models.any_error import AnyError raise utils.get_error( resp, ( ( (400, 409, 401, 403, 404, 429, 500), utils.unpack_union(AnyError), ), ), )