Issue/462 Submit Multiple Files to assignment

canvasapi/assignment.py

@@ -1,3 +1,5 @@
+from typing import List
+
 from canvasapi.canvas_object import CanvasObject
 from canvasapi.exceptions import CanvasException, RequiredFieldMissing
 from canvasapi.paginated_list import PaginatedList
@@ -22,6 +24,71 @@ def __init__(self, requester, attributes):
     def __str__(self):
         return "{} ({})".format(self.name, self.id)
 
+    def bulk_submit(self, submission, files=None, **kwargs):
+        """
+        Makes a submission of multiple files for an assigment.
+
+        :calls: `POST /api/v1/courses/:course_id/assignments/:assignment_id/submissions \
+        <https://canvas.instructure.com/doc/api/submissions.html#method.submissions.create>`_
+
+        :param submission: The attributes of the submission.
+        :type submission: dict
+        :param files: A list of files to upload with the submission. (Optional,
+            defaults to `None`. Submission type must be `online_upload`)
+        :type files: a list of files or str
+
+        :rtype: :class:`canvasapi.submission.Submission`
+        """
+        if isinstance(submission, dict) and "submission_type" in submission:
+            kwargs["submission"] = submission
+        else:
+            raise RequiredFieldMissing(
+                "Dictionary with key 'submission_type' is required."
+            )
+
+        if files:
+            if submission.get("submission_type") != "online_upload":
+                raise ValueError(
+                    "To upload files, 'submission['submission_type']' must be 'online_upload'."
+                )
+            file_ids = self.bulk_upload(files, **kwargs)
+            kwargs["submission"]["file_ids"] = file_ids
+
+        response = self._requester.request(
+            "POST",
+            "courses/{}/assignments/{}/submissions".format(self.course_id, self.id),
+            _kwargs=combine_kwargs(**kwargs),
+        )
+        response_json = response.json()
+        response_json.update(course_id=self.course_id)
+
+        return Submission(self._requester, response_json)
+
+    def bulk_upload(self, files: List[FileOrPathLike], user="self", **kwargs):
+        """
+        Upload multiples files to a submission.
+
+        :param files: A list of files or path of files to upload
+        :type files: list of FileLike
+        param user: The object or ID of the related user, or 'self' for the
+            current user. Defaults to 'self'.
+        :type user: :class:`canvasapi.user.User`, int, or str
+
+        :returns: A list of file ids corresponding to files uploaded \
+            to the assigment.
+        :rtype: list
+        """
+        file_ids = []
+        for file in files:
+            upload_response = self.upload_to_submission(file, user, **kwargs)
+
+            if upload_response[0]:
+                file_ids.append(upload_response[1]["id"])
+            else:
+                raise CanvasException(f"File {file} upload failed.")
+
+        return file_ids
+
     def create_override(self, **kwargs):
         """
         Create an override for this assignment.
@@ -443,7 +510,7 @@ def upload_to_submission(self, file: FileOrPathLike, user="self", **kwargs):
                 self.course_id, self.id, user_id
             ),
             file,
-            **kwargs
+            **kwargs,
         ).start()
 
 

tests/test_assignment.py

@@ -41,6 +41,141 @@ def test__init__overrides(self, m):
         self.assertEqual(len(assignment.overrides), 1)
         self.assertIsInstance(assignment.overrides[0], AssignmentOverride)
 
+    # bulk_submit()
+    def test_bulk_submit(self, m):
+        register_uris({"assignment": ["submit"]}, m)
+
+        sub_type = "online_upload"
+        sub_dict = {"submission_type": sub_type}
+        submission = self.assignment.bulk_submit(sub_dict)
+
+        self.assertIsInstance(submission, Submission)
+        self.assertTrue(hasattr(submission, "submission_type"))
+        self.assertEqual(submission.submission_type, sub_type)
+
+    def test_bulk_submit_fail(self, m):
+        with self.assertRaises(RequiredFieldMissing):
+            self.assignment.bulk_submit({})
+
+    def test_bulk_submit_file(self, m):
+        register_uris({"assignment": ["submit", "upload", "upload_final"]}, m)
+
+        filename = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        try:
+            with open(filename, "w+") as file:
+                sub_type = "online_upload"
+                sub_dict = {"submission_type": sub_type}
+                submission = self.assignment.bulk_submit(sub_dict, file)
+
+            self.assertIsInstance(submission, Submission)
+            self.assertTrue(hasattr(submission, "submission_type"))
+            self.assertEqual(submission.submission_type, sub_type)
+        finally:
+            cleanup_file(filename)
+
+    def test_bulk_submit_multiple_files(self, m):
+        register_uris({"assignment": ["submit", "upload", "upload_final"]}, m)
+
+        filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        try:
+            with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2:
+                files = [f1, f2]
+
+                sub_type = "online_upload"
+                sub_dict = {"submission_type": sub_type}
+                submission = self.assignment.bulk_submit(sub_dict, files)
+
+            self.assertIsInstance(submission, Submission)
+            self.assertTrue(hasattr(submission, "submission_type"))
+            self.assertEqual(submission.submission_type, sub_type)
+        finally:
+            cleanup_file(filename_one)
+            cleanup_file(filename_two)
+
+    def test_bulk_submit_wrong_type(self, m):
+        filename = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        sub_type = "online_text_entry"
+        sub_dict = {"submission_type": sub_type}
+
+        with self.assertRaises(ValueError):
+            self.assignment.bulk_submit(sub_dict, filename)
+
+    def test_bulk_submit_upload_failure(self, m):
+        register_uris({"assignment": ["submit", "upload", "upload_fail"]}, m)
+
+        filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        try:
+            with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2:
+                files = [f1, f2]
+
+                sub_type = "online_upload"
+                sub_dict = {"submission_type": sub_type}
+                with self.assertRaises(CanvasException):
+                    self.assignment.bulk_submit(sub_dict, files)
+        finally:
+            cleanup_file(filename_one)
+            cleanup_file(filename_two)
+
+    # bulk_upload()
+    def test_bulk_upload_self(self, m):
+        register_uris({"assignment": ["upload", "upload_final"]}, m)
+
+        filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        try:
+            with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2:
+                files = [f1, f2]
+
+                file_ids = self.assignment.bulk_upload(files)
+
+            self.assertIsNotNone(file_ids)
+            self.assertIsInstance(file_ids, list)
+        finally:
+            cleanup_file(filename_one)
+            cleanup_file(filename_two)
+
+    def test_bulk_upload_failure(self, m):
+        register_uris({"assignment": ["upload", "upload_fail"]}, m)
+
+        filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        try:
+            with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2:
+                files = [f1, f2]
+
+                with self.assertRaises(CanvasException):
+                    self.assignment.bulk_upload(files)
+        finally:
+            cleanup_file(filename_one)
+            cleanup_file(filename_two)
+
+    def test_bulk_upload_user(self, m):
+        register_uris({"assignment": ["upload_by_id", "upload_final"]}, m)
+
+        filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex)
+        filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex)
+
+        user_id = 1
+
+        try:
+            with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2:
+                files = [f1, f2]
+
+                file_ids = self.assignment.bulk_upload(files, user_id)
+
+            self.assertIsNotNone(file_ids)
+            self.assertIsInstance(file_ids, list)
+        finally:
+            cleanup_file(filename_one)
+            cleanup_file(filename_two)
+
     # create_override()
     def test_create_override(self, m):
         register_uris({"assignment": ["create_override"]}, m)