Add unique field prevalidation

orm/models.py

@@ -48,6 +48,52 @@ def __new__(
         return new_model
 
 
+class ModelObject(typesystem.Object):
+    # NOTE: to be updated if/when typesystem handles validation of unique fields.
+
+    def __init__(self, *args, queryset: "QuerySet", **kwargs):
+        super().__init__(*args, **kwargs)
+        self.queryset = queryset
+
+    async def validate_unique(self, value: dict):
+        unique_properties = {
+            key: value[key] for key, val in self.properties.items() if val.unique
+        }
+
+        error_messages = []
+
+        for key, val in unique_properties.items():
+            expr = self.queryset.table.select()
+            column = getattr(self.queryset.table.c, key)
+            expr = expr.where(column == val)
+            row = await self.queryset.database.fetch_one(query=expr)
+            if row is not None:
+                text = f"{self.queryset.table.name} with {key}='{val}' already exists"
+                message = typesystem.Message(text=text, code="unique_exists")
+                error_messages.append(message)
+
+        if error_messages:
+            raise typesystem.ValidationError(messages=error_messages)
+
+    async def validate(self, value: dict, strict: bool = False) -> typing.Any:
+        messages = []
+
+        try:
+            validated = super().validate(value, strict=strict)
+        except typesystem.ValidationError as exc:
+            messages.extend(exc.messages())
+
+        try:
+            await self.validate_unique(value)
+        except typesystem.ValidationError as exc:
+            messages.extend(exc.messages())
+
+        if messages:
+            raise typesystem.ValidationError(messages=messages)
+
+        return validated
+
+
 class QuerySet:
     def __init__(self, model_cls=None, filter_clauses=None, select_related=None):
         self.model_cls = model_cls
@@ -189,10 +235,13 @@ async def create(self, **kwargs):
         # Validate the keyword arguments.
         fields = self.model_cls.fields
         required = [key for key, value in fields.items() if not value.has_default()]
-        validator = typesystem.Object(
-            properties=fields, required=required, additional_properties=False
+        validator = ModelObject(
+            properties=fields,
+            required=required,
+            additional_properties=False,
+            queryset=self,
         )
-        kwargs = validator.validate(kwargs)
+        kwargs = await validator.validate(kwargs)
 
         # Build the insert expression.
         expr = self.table.insert()
@@ -223,10 +272,16 @@ def pk(self, value):
         setattr(self, self.__pkname__, value)
 
     async def update(self, **kwargs):
+        # Prevent primary key from being updated.
+        if "pk" in kwargs or self.__pkname__ in kwargs:
+            raise ValueError(
+                f"the primary key of a model instance cannot be updated"
+            )
+
         # Validate the keyword arguments.
         fields = {key: field for key, field in self.fields.items() if key in kwargs}
-        validator = typesystem.Object(properties=fields)
-        kwargs = validator.validate(kwargs)
+        validator = ModelObject(properties=fields, queryset=self.objects)
+        kwargs = await validator.validate(kwargs)
 
         # Build the update expression.
         pk_column = getattr(self.__table__.c, self.__pkname__)

tests/test_models.py

@@ -6,6 +6,7 @@
 
 import databases
 import orm
+import typesystem
 
 DATABASE_URL = "sqlite:///test.db"
 database = databases.Database(DATABASE_URL, force_rollback=True)
@@ -32,6 +33,15 @@ class Product(orm.Model):
     in_stock = orm.Boolean(default=False)
 
 
+class Post(orm.Model):
+    __tablename__ = "post"
+    __metadata__ = metadata
+    __database__ = database
+
+    id = orm.Integer(primary_key=True)
+    slug = orm.String(max_length=100, unique=True)
+
+
 @pytest.fixture(autouse=True, scope="module")
 def create_test_database():
     engine = sqlalchemy.create_engine(DATABASE_URL)
@@ -81,6 +91,9 @@ async def test_model_crud():
         assert user.pk is not None
         assert users == [user]
 
+        with pytest.raises(typesystem.ValidationError):
+            await User.objects.create(foo="is not a User field")
+
         lookup = await User.objects.get()
         assert lookup == user
 
@@ -90,6 +103,9 @@ async def test_model_crud():
         assert user.pk is not None
         assert users == [user]
 
+        with pytest.raises(ValueError):
+            await user.update(pk=42)
+
         await user.delete()
         users = await User.objects.all()
         assert users == []
@@ -148,3 +164,17 @@ async def test_model_count():
 
         assert await User.objects.count() == 3
         assert await User.objects.filter(name__icontains="T").count() == 1
+
+
+@async_adapter
+async def test_validate_unique():
+    slug = "hello-world"
+    async with database:
+        await Post.objects.create(slug=slug)
+
+        with pytest.raises(typesystem.ValidationError):
+            await Post.objects.create(slug=slug)
+
+        other = await Post.objects.create(slug="world-hello")
+        with pytest.raises(typesystem.ValidationError):
+            await other.update(slug=slug)