feat: user submission management v1

forms_pro/api/form.py

@@ -56,6 +56,7 @@ def get_form(form_id: str) -> dict:
         "route": form.route,
         "is_published": form.is_published,
         "allow_incomplete": form.allow_incomplete,
+        "linked_doctype": form.linked_doctype,
     }
 
 

forms_pro/api/submission.py

@@ -1,14 +1,133 @@
+from datetime import datetime
+from typing import Any
+
 import frappe
+from frappe import _
+from frappe.share import add_docshare
+from pydantic import BaseModel, Field, field_validator
+
+from forms_pro.forms_pro.doctype.form.form import Form
+from forms_pro.utils.form_generator import SubmissionStatus
+
+
+class UserSubmissionResponse(BaseModel):
+    name: str = Field(description="Name of the submission")
+    creation: datetime = Field(description="Creation date of the submission")
+    modified: datetime = Field(description="Last modified date of the submission")
+    submission_status: str = Field(
+        description="Status of the submission",
+        alias="fp_submission_status",
+        default=SubmissionStatus.SUBMITTED.value,
+    )
+
+    @field_validator("creation", "modified", mode="before")
+    @classmethod
+    def parse_datetime(cls, v: Any) -> datetime:
+        """Convert datetime string to datetime object."""
+        if isinstance(v, str):
+            return frappe.utils.get_datetime(v)
+        if isinstance(v, datetime):
+            return v
+        raise ValueError(f"Invalid datetime value: {v}")
 
 
 @frappe.whitelist(allow_guest=True)
-def submit_form_response(form_id: str, form_data: list[dict]):
-    form = frappe.get_doc("Form", form_id)
+def submit_form_response(
+    form_id: str,
+    form_data: list[dict],
+    submission_status: SubmissionStatus = SubmissionStatus.SUBMITTED,
+) -> str:
+    """
+    Submit a form response
+
+    Args:
+        form_id: The ID of the form
+        form_data: The data of the form
+        submission_status: The status of the submission: Default is `Submitted`
+
+    Returns:
+        The name of the submission
+    """
+    try:
+        form: Form = frappe.get_doc("Form", form_id)
+        linked_doctype = form.linked_doctype
+
+        if not form.is_published:
+            frappe.throw(
+                _("This form is un-published, so responses are no longer being collected."),
+                frappe.PermissionError,
+            )
+
+        submission = frappe.new_doc(linked_doctype)
+        for data in form_data:
+            submission.set(data["fieldname"], data["value"])
+
+        submission.fp_linked_form = form_id
+        submission.fp_submission_status = submission_status.value
+        submission.insert(ignore_permissions=True)
+
+        # Share the submission with the owner
+        add_docshare(
+            doctype=linked_doctype,
+            name=submission.name,
+            user=frappe.session.user,
+            read=1,
+            write=1,
+            flags={
+                "ignore_share_permission": True,
+            },
+        )
+
+        return submission.name
+    except Exception:
+        raise
+
+
+@frappe.whitelist()
+def get_user_submissions(form_id: str) -> list[UserSubmissionResponse]:
+    """
+    Get the submissions for a user
+
+    Args:
+        form_id: The ID of the form
+
+    Returns:
+        A list of submissions for the user
+    """
+
+    if frappe.session.user == "Guest":
+        return []
+
+    form: Form = frappe.get_doc("Form", form_id)
     linked_doctype = form.linked_doctype
 
-    submission = frappe.new_doc(linked_doctype)
-    for data in form_data:
-        submission.set(data["fieldname"], data["value"])
-    submission.insert(ignore_permissions=True)
+    submissions = frappe.get_all(
+        doctype=linked_doctype,
+        filters={"owner": frappe.session.user},
+        fields=["name", "creation", "modified", "fp_submission_status"],
+        order_by="creation",
+    )
+
+    return [UserSubmissionResponse.model_validate(submission).model_dump() for submission in submissions]
+
+
+@frappe.whitelist()
+def get_submission(submission_doctype: str, submission_name: str) -> dict[str, Any]:
+    """
+    Get a submission by name
+
+    Args:
+        submission_name: The name of the submission
+
+    Returns:
+        The submission
+    """
+    submission = frappe.get_doc(submission_doctype, submission_name)
+
+    if not frappe.has_permission(doctype=submission.doctype, ptype="read", doc=submission.name):
+        frappe.throw(
+            _("You do not have permission to read this submission."),
+            frappe.PermissionError,
+        )
 
-    return submission.name
+    return submission.as_dict()

forms_pro/api/user.py

@@ -1,5 +1,6 @@
 import frappe
 from frappe.core.doctype.has_role.has_role import HasRole
+from frappe.core.doctype.user.user import User
 from pydantic import BaseModel, Field, field_validator
 
 from forms_pro.utils.teams import get_user_teams as get_user_teams_utils
@@ -51,10 +52,10 @@ def get_current_user() -> GetUserResponseSchema:
     """
 
     user_id = frappe.session.user
-    user_doc = frappe.get_doc("User", user_id)
+    user_doc: User = frappe.get_doc("User", user_id)
     data = user_doc.as_dict()
     data["roles"] = user_doc.get("roles")
-    data["has_desk_access"] = user_doc.has_desk_access()
+    data["has_desk_access"] = bool(user_doc.has_desk_access())
 
     return GetUserResponseSchema.model_validate(data).model_dump()
 

forms_pro/utils/form_generator.py

@@ -1,5 +1,8 @@
+from enum import Enum
+
 import frappe
 from frappe import _
+from frappe.custom.doctype.custom_field.custom_field import CustomField
 from frappe.share import add_docshare
 
 FORMS_PRO_ROLE = "Forms Pro User"
@@ -53,6 +56,33 @@ def create_form(team_id: str):
     }
 
 
+USER_FORM_MODULE = "User Forms"
+
+
+class SubmissionStatus(str, Enum):
+    DRAFT = "Draft"
+    SUBMITTED = "Submitted"
+
+
+LINKED_FORM_FIELDOPTIONS = {
+    "label": "Linked Form",
+    "fieldname": "fp_linked_form",
+    "fieldtype": "Link",
+    "options": "Form",
+    "read_only": 1,
+}
+
+SUBMISSION_STATUS_FIELDOPTIONS = {
+    "label": "Submission Status (Form)",
+    "fieldname": "fp_submission_status",
+    "fieldtype": "Select",
+    "options": "\n".join([status.value for status in SubmissionStatus]),
+    "default": SubmissionStatus.DRAFT.value,
+    "read_only": 1,
+    "in_list_view": 1,
+}
+
+
 class FormGenerator:
     def __init__(
         self,
@@ -66,6 +96,7 @@ def __init__(
 
     def generate(self) -> None:
         self._initialize_doctype()
+        self._add_status_field()
         self._initialize_form_document()
         frappe.clear_cache()
 
@@ -124,6 +155,39 @@ def _initialize_doctype(self) -> None:
 
         self.doctype = placeholder_doctype
 
+    def _add_status_field(self) -> None:
+        # If the doctype is not custom, add a `Custom Field` to the doctype
+        if not self.doctype.custom:
+            self._add_custom_field_for_status()
+            self._add_custom_field_for_linked_form()
+
+        # If the doctype is custom, we can go ahead and append the field directly to the doctype
+        self.doctype.append(
+            "fields",
+            SUBMISSION_STATUS_FIELDOPTIONS,
+        )
+
+        # Add the linked form field
+        self.doctype.append(
+            "fields",
+            LINKED_FORM_FIELDOPTIONS,
+        )
+        self.doctype.save(ignore_permissions=True)
+
+    def _add_custom_field_for_status(self) -> None:
+        custom_field: CustomField = frappe.new_doc("Custom Field")
+        custom_field.update(SUBMISSION_STATUS_FIELDOPTIONS)
+        custom_field.dt = self.doctype.name
+        custom_field.is_system_generated = True
+        custom_field.insert(ignore_permissions=True)
+
+    def _add_custom_field_for_linked_form(self) -> None:
+        custom_field: CustomField = frappe.new_doc("Custom Field")
+        custom_field.update(LINKED_FORM_FIELDOPTIONS)
+        custom_field.dt = self.doctype.name
+        custom_field.is_system_generated = True
+        custom_field.insert(ignore_permissions=True)
+
     def _initialize_form_document(self) -> None:
         form_document = frappe.new_doc("Form")
         form_document.linked_doctype = self.doctype.name

forms_pro/utils/test_form_generator.py

@@ -211,3 +211,128 @@ def test_generate_creates_form_docshare(self):
         self.assertEqual(docshare.write, 1, "DocShare should have write permission")
         self.assertEqual(docshare.share, 1, "DocShare should have share permission")
         self.assertEqual(docshare.submit, 0, "DocShare should not have submit permission")
+
+    def test_status_field_is_added_custom_doctype(self):
+        """Test that status field is added to custom doctype"""
+        from forms_pro.utils.form_generator import SUBMISSION_STATUS_FIELDOPTIONS
+
+        frappe.set_user(self.test_user)
+        form_generator = FormGenerator(team_id=self.test_team)
+        form_generator.generate()
+
+        frappe.set_user("Administrator")
+        # Assertions
+        self.assertIsNotNone(form_generator.doctype.fields)
+        status_field = next(
+            field
+            for field in form_generator.doctype.fields
+            if field.fieldname == SUBMISSION_STATUS_FIELDOPTIONS["fieldname"]
+        )
+        assert status_field is not None, "Status field should be added to doctype"
+        self.assertEqual(status_field.label, SUBMISSION_STATUS_FIELDOPTIONS["label"])
+        self.assertEqual(status_field.fieldname, SUBMISSION_STATUS_FIELDOPTIONS["fieldname"])
+        self.assertEqual(status_field.fieldtype, SUBMISSION_STATUS_FIELDOPTIONS["fieldtype"])
+        self.assertEqual(status_field.options, SUBMISSION_STATUS_FIELDOPTIONS["options"])
+        self.assertEqual(status_field.default, SUBMISSION_STATUS_FIELDOPTIONS["default"])
+        self.assertEqual(status_field.read_only, SUBMISSION_STATUS_FIELDOPTIONS["read_only"])
+        self.assertEqual(status_field.in_list_view, SUBMISSION_STATUS_FIELDOPTIONS["in_list_view"])
+
+    def test_linked_form_field_is_added_custom_doctype(self):
+        """Test that linked form field is added to custom doctype"""
+        from forms_pro.utils.form_generator import LINKED_FORM_FIELDOPTIONS
+
+        frappe.set_user(self.test_user)
+        form_generator = FormGenerator(team_id=self.test_team)
+        form_generator.generate()
+
+        frappe.set_user("Administrator")
+        self.assertIsNotNone(form_generator.doctype.fields)
+        linked_form_field = next(
+            field
+            for field in form_generator.doctype.fields
+            if field.fieldname == LINKED_FORM_FIELDOPTIONS["fieldname"]
+        )
+        assert linked_form_field is not None, "Linked form field should be added to doctype"
+        self.assertEqual(linked_form_field.label, LINKED_FORM_FIELDOPTIONS["label"])
+        self.assertEqual(linked_form_field.fieldname, LINKED_FORM_FIELDOPTIONS["fieldname"])
+        self.assertEqual(linked_form_field.fieldtype, LINKED_FORM_FIELDOPTIONS["fieldtype"])
+        self.assertEqual(linked_form_field.options, LINKED_FORM_FIELDOPTIONS["options"])
+        self.assertEqual(linked_form_field.read_only, LINKED_FORM_FIELDOPTIONS["read_only"])
+
+    def test_status_field_is_added_core_doctype(self):
+        """Test that status field is added to core doctype as a custom field"""
+        from forms_pro.utils.form_generator import SUBMISSION_STATUS_FIELDOPTIONS
+
+        test_doctype = frappe.new_doc("DocType")
+        test_doctype.name = "Test Status Field Doctype" + frappe.utils.random_string(8)
+        test_doctype.module = "User Forms"
+        test_doctype.custom = False
+        test_doctype.insert(ignore_permissions=True)
+
+        frappe.set_user(self.test_user)
+        form_generator = FormGenerator(linked_doctype=test_doctype.name, team_id=self.test_team)
+        form_generator.generate()
+
+        # Assertions
+        frappe.set_user("Administrator")
+        self.assertIsNotNone(form_generator.doctype.fields)
+        status_field = next(
+            field
+            for field in form_generator.doctype.fields
+            if field.fieldname == SUBMISSION_STATUS_FIELDOPTIONS["fieldname"]
+        )
+        assert status_field is not None, "Status field should be added to doctype"
+
+        custom_field_id = frappe.db.exists(
+            "Custom Field",
+            {"dt": test_doctype.name, "fieldname": SUBMISSION_STATUS_FIELDOPTIONS["fieldname"]},
+        )
+        self.assertIsNotNone(custom_field_id, "Custom field should be created for doctype")
+
+        custom_field = frappe.get_doc("Custom Field", custom_field_id)
+
+        self.assertEqual(custom_field.dt, test_doctype.name)
+        self.assertEqual(custom_field.fieldname, SUBMISSION_STATUS_FIELDOPTIONS["fieldname"])
+        self.assertEqual(custom_field.fieldtype, SUBMISSION_STATUS_FIELDOPTIONS["fieldtype"])
+        self.assertEqual(custom_field.options, SUBMISSION_STATUS_FIELDOPTIONS["options"])
+        self.assertEqual(custom_field.default, SUBMISSION_STATUS_FIELDOPTIONS["default"])
+        self.assertEqual(custom_field.read_only, SUBMISSION_STATUS_FIELDOPTIONS["read_only"])
+        self.assertEqual(custom_field.in_list_view, SUBMISSION_STATUS_FIELDOPTIONS["in_list_view"])
+
+    def test_linked_form_field_is_added_core_doctype(self):
+        """Test that linked form field is added to core doctype as a custom field"""
+        from forms_pro.utils.form_generator import LINKED_FORM_FIELDOPTIONS
+
+        test_doctype = frappe.new_doc("DocType")
+        test_doctype.name = "Test Linked Form Doctype" + frappe.utils.random_string(8)
+        test_doctype.module = "User Forms"
+        test_doctype.custom = False
+        test_doctype.insert(ignore_permissions=True)
+
+        frappe.set_user(self.test_user)
+        form_generator = FormGenerator(linked_doctype=test_doctype.name, team_id=self.test_team)
+        form_generator.generate()
+
+        # Assertions
+        frappe.set_user("Administrator")
+        self.assertIsNotNone(form_generator.doctype.fields)
+        linked_form_field = next(
+            field
+            for field in form_generator.doctype.fields
+            if field.fieldname == LINKED_FORM_FIELDOPTIONS["fieldname"]
+        )
+        assert linked_form_field is not None, "Linked form field should be added to doctype"
+
+        custom_field_id = frappe.db.exists(
+            "Custom Field",
+            {"dt": test_doctype.name, "fieldname": LINKED_FORM_FIELDOPTIONS["fieldname"]},
+        )
+        self.assertIsNotNone(custom_field_id, "Custom field should be created for doctype")
+
+        custom_field = frappe.get_doc("Custom Field", custom_field_id)
+
+        self.assertEqual(custom_field.dt, test_doctype.name)
+        self.assertEqual(custom_field.fieldname, LINKED_FORM_FIELDOPTIONS["fieldname"])
+        self.assertEqual(custom_field.fieldtype, LINKED_FORM_FIELDOPTIONS["fieldtype"])
+        self.assertEqual(custom_field.options, LINKED_FORM_FIELDOPTIONS["options"])
+        self.assertEqual(custom_field.read_only, LINKED_FORM_FIELDOPTIONS["read_only"])

frontend/auto-imports.d.ts

@@ -5,6 +5,4 @@
 // Generated by unplugin-auto-import
 // biome-ignore lint: disable
 export {}
-declare global {
-
-}
+declare global {}