Add supporto for limit number of submit in collective.volto.formsuppo…

…rt + added possibility to add a unique constraint for form block

CHANGES.rst

@@ -4,7 +4,10 @@ Changelog
 6.2.15 (unreleased)
 -------------------
 
-- Nothing changed yet.
+- Added support for field limit that limits form submission and the exceeding ones are added to a waiting list.
+  [eikichi18]
+- Added support to make one or more fields unique.
+  [eikichi18]
 
 
 6.2.14 (2024-07-11)

setup.py

@@ -59,6 +59,7 @@
         "plone.app.dexterity>2.6.9",
         "collective.venue[geolocation]",
         "collective.volto.blocksfield",
+        "collective.volto.formsupport",
         "collective.z3cform.datagridfield",
         "plone.formwidget.geolocation",
         "redturtle.volto>=5.0.0",

src/design/plone/contenttypes/patches/__init__.py

@@ -3,6 +3,15 @@
     patch_base_folder_serializer,
 )
 from design.plone.contenttypes.patches.baseserializer import patch_base_serializer
+from design.plone.contenttypes.patches.collective_volto_formsupport import (
+    patch_FormDataExportGet_get_data,
+)
+from design.plone.contenttypes.patches.collective_volto_formsupport import (
+    patch_FormDataStore_methods,
+)
+from design.plone.contenttypes.patches.collective_volto_formsupport import (
+    patch_SubmitPost_reply,
+)
 
 import logging
 
@@ -13,3 +22,7 @@
 logger.info("Patching plone.restapi.serializer.dxcontent.SerializeFolderToJson._call__")
 patch_base_serializer()
 patch_base_folder_serializer()
+
+patch_FormDataExportGet_get_data()
+patch_SubmitPost_reply()
+patch_FormDataStore_methods()

src/design/plone/contenttypes/patches/collective_volto_formsupport.py

@@ -0,0 +1,214 @@
+# -*- coding: utf-8 -*-
+"""
+We use this file to change the base behavior of collective.volto.formsupport
+to support some new feature:
+    - limit on form submit
+    - unique field in one form
+
+Why do we use monkeypatch instead of overriding the classes?
+Because it's temporary, until collective.volto.formsupport can support backend
+validation for data
+"""
+from collective.volto.formsupport import _
+from collective.volto.formsupport import logger
+from collective.volto.formsupport.datamanager.catalog import FormDataStore
+from collective.volto.formsupport.interfaces import IFormDataStore
+from collective.volto.formsupport.restapi.services.form_data.csv import (
+    FormDataExportGet,
+)
+from collective.volto.formsupport.restapi.services.submit_form.post import logger
+from collective.volto.formsupport.restapi.services.submit_form.post import (
+    PostEventService,
+)
+from collective.volto.formsupport.restapi.services.submit_form.post import SubmitPost
+from datetime import datetime
+from io import StringIO
+from plone.protect.interfaces import IDisableCSRFProtection
+from plone.restapi.serializer.converters import json_compatible
+from souper.soup import Record
+from zExceptions import BadRequest
+from zope.component import getMultiAdapter
+from zope.event import notify
+from zope.i18n import translate
+from zope.interface import alsoProvides
+
+import csv
+
+
+SKIP_ATTRS = ["block_id", "fields_labels", "fields_order"]
+
+
+def get_data(self):
+    store = getMultiAdapter((self.context, self.request), IFormDataStore)
+    sbuf = StringIO()
+    fixed_columns = ["date"]
+    columns = []
+    custom_colums = []
+    if self.form_block.get("limit", None) is not None:
+        limit = int(self.form_block["limit"])
+        if limit > -1:
+            custom_colums.append("waiting_list")
+
+    rows = []
+    for index, item in enumerate(store.search()):
+        data = {}
+        fields_labels = item.attrs.get("fields_labels", {})
+        for k in self.get_ordered_keys(item):
+            if k in SKIP_ATTRS:
+                continue
+            value = item.attrs.get(k, None)
+            label = fields_labels.get(k, k)
+            if label not in columns and label not in fixed_columns:
+                columns.append(label)
+            data[label] = json_compatible(value)
+        for k in fixed_columns:
+            # add fixed columns values
+            value = item.attrs.get(k, None)
+            data[k] = json_compatible(value)
+        if "waiting_list" in custom_colums:
+            data.update(
+                {
+                    "waiting_list": (
+                        translate(_("yes_label", default="Yes"))
+                        if not (index < limit)
+                        else translate(_("no_label", default="No"))
+                    )
+                }
+            )
+
+        rows.append(data)
+    columns.extend(fixed_columns)
+    columns.extend(custom_colums)
+    writer = csv.DictWriter(sbuf, fieldnames=columns, quoting=csv.QUOTE_ALL)
+    writer.writeheader()
+    for row in rows:
+        writer.writerow(row)
+    res = sbuf.getvalue()
+    sbuf.close()
+    return res
+
+
+def patch_FormDataExportGet_get_data():
+    logger.info(
+        "Patch get_data methos of class FormDataExporterGet from collective.volto.formsupport"  # noqa
+    )
+    FormDataExportGet.get_data = get_data
+
+
+def reply(self):
+    self.validate_form()
+
+    self.store_action = self.block.get("store", False)
+    self.send_action = self.block.get("send", [])
+    self.submit_limit = int(self.block.get("limit", "-1"))
+
+    # Disable CSRF protection
+    alsoProvides(self.request, IDisableCSRFProtection)
+
+    notify(PostEventService(self.context, self.form_data))
+    data = self.form_data.get("data", [])
+
+    if self.send_action:
+        try:
+            self.send_data()
+        except BadRequest as e:
+            raise e
+        except Exception as e:
+            logger.exception(e)
+            message = translate(
+                _(
+                    "mail_send_exception",
+                    default="Unable to send confirm email. Please retry later or contact site administrator.",  # noqa
+                ),
+                context=self.request,
+            )
+            self.request.response.setStatus(500)
+            return dict(type="InternalServerError", message=message)
+    if self.store_action:
+        try:
+            data = self.store_data()
+        except ValueError as e:
+            logger.exception(e)
+            message = translate(
+                _(
+                    "save_data_exception",
+                    default="Unable to save data. Value not unique: '${fields}'",
+                    mapping={"fields": e.args[0]},
+                ),
+                context=self.request,
+            )
+            self.request.response.setStatus(500)
+            return dict(type="InternalServerError", message=message)
+
+    return {"data": data}
+
+
+def patch_SubmitPost_reply():
+    logger.info(
+        "Patch reply method of class SubmitPost from collective.volto.formsupport"
+    )
+    SubmitPost.reply = reply
+
+
+def add(self, data):
+    form_fields = self.get_form_fields()
+    if not form_fields:
+        logger.error(
+            'Block with id {} and type "form" not found in context: {}.'.format(
+                self.block_id, self.context.absolute_url()
+            )
+        )
+        return None
+
+    fields = {
+        x["field_id"]: x.get("custom_field_id", x.get("label", x["field_id"]))
+        for x in form_fields
+    }
+    record = Record()
+    fields_labels = {}
+    fields_order = []
+    for field_data in data["form_data"]:
+        field_id = field_data.get("field_id", "")
+        value = field_data.get("value", "")
+        if field_id in fields:
+            record.attrs[field_id] = value
+            fields_labels[field_id] = fields[field_id]
+            fields_order.append(field_id)
+    record.attrs["fields_labels"] = fields_labels
+    record.attrs["fields_order"] = fields_order
+    record.attrs["date"] = datetime.now()
+    record.attrs["block_id"] = self.block_id
+
+    keys = [(x["field_id"], x["label"]) for x in form_fields if x.get("unique", False)]
+    if keys:
+        saved_data = self.soup.data.values()
+        for saved_record in saved_data:
+            unique = False
+            for key in keys:
+                if record.attrs.storage[key[0]] != saved_record.attrs.storage[key[0]]:
+                    unique = True
+                    break
+
+            if not unique:
+                raise ValueError(f" {', '.join([x[1] for x in keys])}")
+
+        return self.soup.add(record)
+
+
+def count(self, query=None):
+    records = []
+    if not query:
+        records = self.soup.data.values()
+
+    return len(records)
+
+
+def patch_FormDataStore_methods():
+    logger.info(
+        "Patch method add of FormDataStore class for collective.volto.formsupport"
+    )
+    FormDataStore.add = add
+    logger.info(
+        "Add method count of FormDataStore class for collective.volto.formsupport"
+    )
+    FormDataStore.count = count