feat: add submission attachment api

pyodk/_endpoints/submission_attachments.py

@@ -0,0 +1,163 @@
+import logging
+from os import PathLike
+
+from pyodk._endpoints import bases
+from pyodk._utils import validators as pv
+from pyodk._utils.session import Session
+from pyodk.errors import PyODKError
+
+log = logging.getLogger(__name__)
+
+
+class SubmissionAttachment(bases.Model):
+    name: str
+    exists: bool
+
+
+class URLs(bases.FrozenModel):
+    _submission: str = "projects/{project_id}/forms/{form_id}/submissions/{{instance_id}}"
+    list: str = f"{_submission}/attachments"
+    get: str = f"{_submission}/attachments/{{fname}}"
+    post: str = f"{_submission}/attachments/{{fname}}"
+    delete: str = f"{_submission}/attachments/{{fname}}"
+
+
+class SubmissionAttachmentService(bases.Service):
+    __slots__ = (
+        "urls",
+        "session",
+        "default_project_id",
+        "default_form_id",
+        "default_submission_id",
+    )
+
+    def __init__(
+        self,
+        session: Session,
+        default_project_id: int | None = None,
+        default_form_id: str | None = None,
+        default_submission_id: str | None = None,
+        urls: URLs = None,
+    ):
+        self.urls: URLs = urls if urls is not None else URLs()
+        self.session: Session = session
+        self.default_project_id: int | None = default_project_id
+        self.default_form_id: str | None = default_form_id
+        self.default_submission_id: str | None = default_submission_id
+
+    def list(
+        self, form_id: str | None = None, project_id: int | None = None
+    ) -> list[SubmissionAttachment]:
+        """
+        Show all required submission attachments and their upload status.
+
+        :param form_id: The xmlFormId of the Form being referenced.
+        :param project_id: The id of the project the Submissions belong to.
+
+        :return: A list of the object representation of all Submission
+            attachment metadata.
+        """
+        try:
+            pid = pv.validate_project_id(project_id, self.default_project_id)
+            fid = pv.validate_form_id(form_id, self.default_form_id)
+        except PyODKError as err:
+            log.error(err, exc_info=True)
+            raise
+
+        response = self.session.response_or_error(
+            method="GET",
+            url=self.session.urlformat(self.urls.list, project_id=pid, form_id=fid),
+            logger=log,
+        )
+        data = response.json()
+        return [SubmissionAttachment(**r) for r in data]
+
+    def get(
+        self,
+        file_name: str,
+        instance_id: str,
+        form_id: str | None = None,
+        project_id: int | None = None,
+    ) -> bytes:
+        """
+        Read Submission metadata.
+
+        :param file_name: The file name of the Submission attachment being referenced.
+        :param instance_id: The instanceId of the Submission being referenced.
+        :param form_id: The xmlFormId of the Form being referenced.
+        :param project_id: The id of the project this form belongs to.
+
+        :return: The attachment bytes for download.
+        """
+        try:
+            pid = pv.validate_project_id(project_id, self.default_project_id)
+            fid = pv.validate_form_id(form_id, self.default_form_id)
+            iid = pv.validate_instance_id(instance_id)
+        except PyODKError as err:
+            log.error(err, exc_info=True)
+            raise
+
+        response = self.session.response_or_error(
+            method="GET",
+            url=self.session.urlformat(
+                self.urls.get,
+                project_id=pid,
+                form_id=fid,
+                instance_id=iid,
+                fname=file_name,
+            ),
+            logger=log,
+        )
+        return response.content
+
+    def upload(
+        self,
+        file_path_or_bytes: PathLike | str | bytes,
+        instance_id: str,
+        file_name: str | None = None,
+        form_id: str | None = None,
+        project_id: int | None = None,
+    ) -> bool:
+        """
+        Upload a Form Draft Attachment.
+
+        :param file_path_or_bytes: The path to the file or file bytes to upload.
+        :param instance_id: The instanceId of the Submission being referenced.
+        :param file_name: A name for the file, otherwise the name in file_path is used.
+        :param form_id: The xmlFormId of the Form being referenced.
+        :param project_id: The id of the project this form belongs to.
+        """
+        try:
+            pid = pv.validate_project_id(project_id, self.default_project_id)
+            fid = pv.validate_form_id(form_id, self.default_form_id)
+            iid = pv.validate_instance_id(instance_id)
+            if isinstance(file_path_or_bytes, bytes):
+                file_bytes = file_path_or_bytes
+                if not file_name:
+                    raise PyODKError(
+                        "A file_name param must be used if uploading file bytes directly"
+                    )
+            else:
+                file_path = pv.validate_file_path(file_path_or_bytes)
+                with open(file_path_or_bytes, "rb") as fd:
+                    file_bytes = fd.read()
+                if file_name is None:
+                    file_name = pv.validate_str(file_path.name, key="file_name")
+        except PyODKError as err:
+            log.error(err, exc_info=True)
+            raise
+
+        response = self.session.response_or_error(
+            method="POST",
+            url=self.session.urlformat(
+                self.urls.post,
+                project_id=pid,
+                form_id=fid,
+                instance_id=iid,
+                fname=file_name,
+            ),
+            logger=log,
+            data=file_bytes,
+        )
+        data = response.json()
+        return data["success"]

pyodk/client.py

@@ -6,6 +6,7 @@
 from pyodk._endpoints.forms import FormService
 from pyodk._endpoints.projects import ProjectService
 from pyodk._endpoints.submissions import SubmissionService
+from pyodk._endpoints.submission_attachments import SubmissionAttachmentService
 from pyodk._utils import config
 from pyodk._utils.session import Session
 
@@ -66,6 +67,11 @@ def __init__(
         self.submissions: SubmissionService = SubmissionService(
             session=self.session, default_project_id=self.project_id
         )
+        self.submission_attachments: SubmissionAttachmentService = (
+            SubmissionAttachmentService(
+                session=self.session, default_project_id=self.project_id
+            )
+        )
         self._comments: CommentService = CommentService(
             session=self.session, default_project_id=self.project_id
         )

tests/endpoints/test_submissions.py

@@ -1,11 +1,20 @@
 from unittest import TestCase
 from unittest.mock import MagicMock, patch
 
+from pyodk._endpoints.submission_attachments import (
+    SubmissionAttachmentService,
+    SubmissionAttachment,
+)
 from pyodk._endpoints.submissions import Submission
 from pyodk._utils.session import Session
 from pyodk.client import Client
 
-from tests.resources import CONFIG_DATA, submissions_data
+from tests.resources import (
+    RESOURCES,
+    CONFIG_DATA,
+    submissions_data,
+    submission_attachments_data,
+)
 
 
 @patch("pyodk._utils.session.Auth.login", MagicMock())
@@ -111,3 +120,74 @@ def test_review__ok(self):
                     review_state="edited",
                 )
                 self.assertIsInstance(observed, Submission)
+
+
+@patch("pyodk._utils.session.Auth.login", MagicMock())
+@patch("pyodk._utils.config.read_config", MagicMock(return_value=CONFIG_DATA))
+class TestSubmissionAttachments(TestCase):
+    def test_list__ok(self):
+        """Should return a list of SubmissionAttachment objects."""
+        fixture = submission_attachments_data.test_submission_attachments
+        with patch.object(Session, "request") as mock_session:
+            mock_session.return_value.status_code = 200
+            mock_session.return_value.json.return_value = fixture
+            with Client() as client:
+                observed = client.submission_attachments.list(form_id="sample_form")
+        self.assertEqual(len(fixture), len(observed))
+        for i, o in enumerate(observed):
+            with self.subTest(i):
+                self.assertIsInstance(o, SubmissionAttachment)
+                self.assertEqual(fixture[i]["name"], o.name)
+                self.assertEqual(fixture[i]["exists"], o.exists)
+
+    def test_get__ok(self):
+        """Should return the binary content of a submission attachment."""
+        fixture = submission_attachments_data.test_submission_attachment_get
+        with patch.object(Session, "request") as mock_session:
+            mock_session.return_value.status_code = 200
+            mock_session.return_value.content = fixture["content"]
+            with Client() as client:
+                observed = client.submission_attachments.get(
+                    file_name=fixture["file_name"],
+                    instance_id=fixture["instance_id"],
+                    form_id=fixture["form_id"],
+                    project_id=fixture["project_id"],
+                )
+        self.assertEqual(fixture["content"], observed)
+        self.assertIsInstance(observed, bytes)
+
+    def test_upload_bytes__ok(self):
+        """Should return True when the bytes attachment is successfully uploaded."""
+        fixture = submission_attachments_data.test_submission_attachment_upload
+        with patch.object(Session, "request") as mock_session:
+            mock_session.return_value.status_code = 200
+            mock_session.return_value.json.return_value = {"success": True}
+            with Client() as client:
+                observed = client.submission_attachments.upload(
+                    file_path_or_bytes=fixture["file_path_or_bytes"],
+                    instance_id=fixture["instance_id"],
+                    file_name=fixture["file_name"],
+                    form_id=fixture["form_id"],
+                    project_id=fixture["project_id"],
+                )
+        self.assertTrue(observed)
+
+    def test_upload_file__ok(self):
+        """Should return True when the file attachment is successfully uploaded."""
+        fixture = submission_attachments_data.test_submission_attachment_upload
+        submission_file_path = (
+            RESOURCES / "attachments" / "submission_image.png"
+        ).as_posix()
+        with patch.object(Session, "request") as mock_session:
+            mock_session.return_value.status_code = 200
+            mock_session.return_value.json.return_value = {"success": True}
+            with Client() as client:
+                # Perform the upload action, passing the file path directly
+                observed = client.submission_attachments.upload(
+                    file_path_or_bytes=submission_file_path,
+                    instance_id=fixture["instance_id"],
+                    file_name=fixture["file_name"],
+                    form_id=fixture["form_id"],
+                    project_id=fixture["project_id"],
+                )
+            self.assertTrue(observed)

tests/resources/submission_attachments_data.py

@@ -0,0 +1,30 @@
+test_submission_attachments = [
+    {
+        "name": "file1.jpg",
+        "exists": True,
+    },
+    {
+        "name": "file2.jpg",
+        "exists": False,
+    },
+    {
+        "name": "file3.jpg",
+        "exists": True,
+    },
+]
+
+test_submission_attachment_get = {
+    "content": b"Mock binary data for attachment download",
+    "file_name": "file1.jpg",
+    "instance_id": "uuid:96f2a014-eaa1-466a-abe2-3ccacc756d5a",
+    "form_id": "sub_attachments",
+    "project_id": 8,
+}
+
+test_submission_attachment_upload = {
+    "project_id": 8,
+    "form_id": "sub_attachments",
+    "instance_id": "uuid:96f2a014-eaa1-466a-abe2-3ccacc756d5a",
+    "file_path_or_bytes": b"Mock binary data for attachment download",
+    "file_name": "file1.jpg",
+}