Unique validator (#229)

* Revert 7bc2015

* add unique field validator

* Don't allow empty Group.name

* Add Group tests for unique validator in edit mode

websauna/system/crud/views.py

@@ -1,23 +1,19 @@
 """Default CRUD views."""
 # Standard Library
 import csv
-import re
 import typing as t
 from abc import abstractmethod
 from io import StringIO
 
 # Pyramid
-import colander
 import deform
-import transaction
 from pyramid.httpexceptions import HTTPFound
 from pyramid.renderers import render
 from pyramid.request import Request
 from pyramid.response import Response
 from pyramid.view import view_config
 
 # SQLAlchemy
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Query
 
 from slugify import slugify
@@ -27,7 +23,6 @@
 from websauna.system.form import interstitial
 from websauna.system.form.fieldmapper import EditMode
 from websauna.system.form.resourceregistry import ResourceRegistry
-from websauna.system.user.models import User
 
 from . import CRUD
 from . import Resource
@@ -387,51 +382,6 @@ def pull_in_widget_resources(self, form: deform.Form):
 
         resource_registry.pull_in_resources(self.request, form)
 
-    def _cleanup_integrity_error(self, form: deform.Form, model: type,
-                                 obj_id: t.Any, user_id: t.Any,
-                                 error: IntegrityError) -> (t.Any, User):
-        """After form data conflicts with data already on the DB,
-        inspect the sqlalchemy error, pinpoint the responsible field,
-        and prepare the deform.field to raise a meaninful validation error
-        on that point.
-
-        Resuming clean request execution requires refetching all
-        objects that will be further needed from the DB.
-        """
-        # Things went as bad as they could.
-        # Let's rolback some steps and simulate a validation
-        # error in the problematic field, before reshowing the form.
-
-        # Find out field that errored from SQLAlchemy error msg.
-        integrity_error_msg = error._message().split("DETAIL: ")[-1].strip()
-        key = re.findall(r"\((.*?)\)", integrity_error_msg)[0]
-        for schema in form.children:
-            if schema.name == key:
-                schema_node = schema.schema
-                break
-        else:
-            # field not found
-            schema_node = form.schema
-
-        def validator(*args, **kw):
-            # TODO: How to actually get an actual translated message?
-            # Maybe force user to feed value from his app?
-            raise colander.Invalid(schema_node, msg='Value already exists.')
-        schema_node.validator = validator
-
-        transaction.abort()
-
-        # Objects are dettached - need to re-fetch a fresh copy
-        # from the DB or things will get ugly.
-
-        if obj_id is not None:
-            obj = self.request.dbsession.query(model).filter_by(id=obj_id).first()
-        else:
-            obj = None
-        user = self.request.dbsession.query(User).filter_by(id=user_id).first()
-
-        return obj, user
-
 
 class Show(FormView):
     """Show one instance of a model."""
@@ -546,27 +496,9 @@ def edit(self):
             controls = self.request.POST.items()
 
             try:
-                controls = list(controls)
                 appstruct = form.validate(controls)
-
-                # import colander
-                obj_id = obj.id
-                user_id = self.request.user.id
-                try:
-                    self.save_changes(form, appstruct, obj)
-                except IntegrityError as error:
-
-                    obj, user = self._cleanup_integrity_error(form, type(obj), obj_id, user_id, error)
-
-                    self.context.obj = obj
-                    self.request.user = user
-
-                    # Force raising a deform.ValidatinFailure, triggering
-                    # data fill in at the apropriate structures
-                    form.validate(controls)
-
-                else:
-                    return self.do_success()
+                self.save_changes(form, appstruct, obj)
+                return self.do_success()
 
             except deform.ValidationFailure as e:
                 # Whoops, bad things happened, render form with validation errors
@@ -672,25 +604,14 @@ def add(self):
             controls = self.request.POST.items()
 
             try:
-                controls = list(controls)
                 appstruct = form.validate(controls)
 
                 # Cannot update id, as it is read-only
                 if 'id' in appstruct:
                     del appstruct["id"]
 
-                user_id = self.request.user.id
-                try:
-                    obj = self.build_object(form, appstruct)
-                    resource = crud.wrap_to_resource(obj)
-                except IntegrityError as error:
-                    obj, user = self._cleanup_integrity_error(form, self.get_model(), None, user_id, error)
-
-                    self.request.user = user
-
-                    # Force raising a deform.ValidatinFailure, triggering
-                    # data fill in at the apropriate structures
-                    form.validate(controls)
+                obj = self.build_object(form, appstruct)
+                resource = crud.wrap_to_resource(obj)
 
                 return self.do_success(resource)
 

websauna/system/form/colander.py

@@ -344,8 +344,6 @@ def get_schema_from_column(self, prop, overrides):
             elif type_overrides_type == TypeOverridesHandling.unknown:
                 # type overrides callback doesn't know about this column
                 type_overrides_type = None
-                type_overrides_kwargs = {}
-
         else:
             type_overrides_type = None
             type_overrides_kwargs = {}

websauna/system/form/fieldmapper.py

@@ -39,6 +39,7 @@
 
 from . import fields
 from .editmode import EditMode
+from .schema import ValidateUnique
 
 
 # TODO: Clean this up when getting rid of colanderalchemy
@@ -152,6 +153,7 @@ def map_column(self, mode: EditMode, request: Request, node: colander.SchemaNode
         """
         logger.debug("Mapping field %s, mode %s, node %s, column %s, column type %s", name, mode, node, column, column_type)
 
+        validator = None
         # Check for autogenerated columns (updated_at)
         if column.onupdate:
             if mode in (EditMode.edit, EditMode.add):
@@ -162,43 +164,40 @@ def map_column(self, mode: EditMode, request: Request, node: colander.SchemaNode
             if mode == EditMode.add:
                 return TypeOverridesHandling.drop, {}
 
+        # Set unique validator
+        if column.unique and (mode in (EditMode.add, EditMode.edit)):
+            validator = ValidateUnique(model, mode)
         # Never add primary keys
         # NOTE: TODO: We need to preserve ids because of nesting mechanism and groupedit widget wants it id
         if column.primary_key:
             # TODO: Looks like column.autoincrement is set True by default, so we cannot use it here
-            # if mode in (EditMode.edit, EditMode.add):
-            #     return TypeOverridesHandling.drop, {}
-            return TypeOverridesHandling.drop, {}
-
+            if mode in (EditMode.edit, EditMode.add):
+                return TypeOverridesHandling.drop, {}
         if column.foreign_keys:
 
             # Handled by relationship mapper
             return TypeOverridesHandling.drop, {}
 
         elif isinstance(column_type, (PostgreSQLUUID, columns.UUID)):
 
-            # UUID's cannot be22 edited
+            # UUID's cannot be edited
             if mode in (EditMode.add, EditMode.edit):
                 return TypeOverridesHandling.drop, {}
 
             # But let's show them
             return fields.UUID(), dict(missing=colander.drop, widget=FriendlyUUIDWidget(readonly=True))
-
         elif isinstance(column_type, Text):
-            return colander.String(), dict(widget=deform.widget.TextAreaWidget())
+            return colander.String(), dict(widget=deform.widget.TextAreaWidget(), validator=validator)
         elif isinstance(column_type, JSONB):
             return JSONValue(), dict(widget=JSONWidget())
-        elif isinstance(column_type, LargeBinary):
-            # Can't edit binary
-            return TypeOverridesHandling.drop, {}
-        elif isinstance(column_type, Geometry):
-            # Can't edit geometry
+        elif isinstance(column_type, (LargeBinary, Geometry)):
+            # Can't edit binary and geometry
             return TypeOverridesHandling.drop, {}
         elif isinstance(column_type, (INET, columns.INET)):
-            return colander.String(), {}
+            return colander.String(), dict(validator=validator)
         else:
             # Default mapping / unknown, let the parent handle
-            return TypeOverridesHandling.unknown, {}
+            return TypeOverridesHandling.unknown, dict(validator=validator)
 
     def map(self, mode: EditMode, request: Request, context: t.Optional[Resource], model: type, includes: t.List, nested=None) -> colander.SchemaNode:
         """

websauna/system/form/schema.py

@@ -11,6 +11,7 @@
 #: Backwards compatibility
 from .csrf import CSRFSchema  # noQA
 from .csrf import add_csrf  # noQA
+from .editmode import EditMode
 
 
 def validate_json(node, value, **kwargs):
@@ -22,6 +23,39 @@ def validate_json(node, value, **kwargs):
         raise colander.Invalid(node, "Not valid JSON")
 
 
+class ValidateUnique:
+    """Check unique constraint."""
+
+    def __init__(self, model, mode):
+        assert mode in (EditMode.add, EditMode.edit)
+
+        self.mode = mode
+        self.model = model
+
+    def __call__(self, node, value):
+        request = node.bindings["request"]
+        dbsession = request.dbsession
+
+        # Add
+        if self.mode == EditMode.add:
+            query = dbsession.query(self.model).filter_by(
+                **{node.name: value}
+            )
+        else:
+            # Edit
+            obj = node.bindings["context"].get_object()
+            query = dbsession.query(self.model).filter_by(**{node.name: value}).filter(
+                getattr(self.model, "id") != obj.id)
+
+        if dbsession.query(query.exists()).scalar():
+            raise colander.Invalid(
+                node, "{model_name} with this `{field}` already exists.".format(
+                    model_name=self.model.__name__,
+                    field=node.name,
+                ),
+            )
+
+
 def enum_values(source: enum.Enum, default: t.Optional[t.Tuple] = ("", "Please choose"), name_transform=str.title) -> t.Iterable[t.Tuple]:
     """Turn Python Enum to key-value pairs lists to be used with selection widgets."""
 

websauna/system/user/usermixin.py

@@ -172,7 +172,7 @@ class GroupMixin:
     uuid = Column(UUID(as_uuid=True), default=uuid4)
 
     #: Human readable / machine referrable name of the group
-    name = Column(String(64), unique=True)
+    name = Column(String(64), nullable=False, unique=True)
 
     #: Human readable description of the group
     description = Column(String(256))

websauna/tests/user/test_groups.py

@@ -4,7 +4,7 @@
 from flaky import flaky
 
 # Websauna
-from websauna.system.user.models import User
+from websauna.system.user.models import User, Group
 from websauna.tests.test_utils import create_logged_in_user
 from websauna.utils.slug import uuid_to_slug
 
@@ -31,13 +31,61 @@ def test_add_group(web_server, browser, dbsession, init):
 
     assert b.is_text_present("Item added")
 
+    # Check name uniqueness
+    b.visit("{}/admin/models/group/add".format(web_server))
+    b.fill("name", GROUP_NAME)
+    b.fill("description", "Foobar")
+    b.find_by_name("add").click()
+    assert b.is_text_present("There was a problem")
+    assert b.is_text_present("Group with this `name` already exists")
+
     # Check we appear in the list
     b.visit("{}/admin/models/group/listing".format(web_server))
 
     # The description appears in the listing
     assert b.is_text_present("Foobar")
 
 
+def test_edit_group(web_server, browser, dbsession, init):
+    """Edit existen group through admin interface."""
+
+    b = browser
+    create_logged_in_user(dbsession, init.config.registry, web_server, browser, admin=True)
+
+    GROUP_NAME2 = GROUP_NAME + "2"
+    GROUP_NAME3 = GROUP_NAME + "3"
+
+    # Create two groups with difference names
+    with transaction.manager:
+        for gname in (GROUP_NAME, GROUP_NAME2):
+            g = Group(name=gname)
+            dbsession.add(g)
+
+    # Check name uniqueness: trying change GROUP_NAME2 to GROUP_NAME
+    b.find_by_css("#nav-admin").click()
+    b.find_by_css("#btn-panel-list-group").click()
+    b.find_by_css(".crud-row-3 .btn-crud-listing-edit").click()
+    b.fill("name", GROUP_NAME)
+    b.find_by_name("save").click()
+    assert b.is_text_present("There was a problem")
+    assert b.is_text_present("Group with this `name` already exists")
+
+    # Check empty Group name
+    b.fill("name", "")
+    b.find_by_name("save").click()
+    assert b.is_text_present("There was a problem")
+
+    # Set new name
+    b.fill("name", GROUP_NAME3)
+    b.find_by_name("save").click()
+    assert b.is_text_present("Changes saved")
+    # Check we appear in the list
+    b.visit("{}/admin/models/group/listing".format(web_server))
+
+    # The new name appears in the listing
+    assert b.is_text_present(GROUP_NAME3)
+
+
 def test_put_user_to_group(web_server, browser, dbsession, init):
     """Check that we can assign users to groups in admin interface."""