IVS-609 - Pydantic for common validation scenarios

backend/apps/ifc_validation/api/v1/pydantic_bridge.py

@@ -0,0 +1,39 @@
+from rest_framework import serializers
+from pydantic import ValidationError as PydanticValidationError
+
+class PydanticValidatorMixin:
+    """
+    Bridge for Pydantic & DRF. Set `Schema` on subclasses.
+
+    """
+    Schema = None 
+
+    # allow subclasses to filter which keys from Pydantic we keep
+    _pydantic_keep_keys: set[str] | None = None
+
+    def validate(self, attrs):
+        if self.Schema is None:
+            return attrs
+        try:
+            model = self.Schema.model_validate(attrs)
+        except PydanticValidationError as exc:
+            err_map = {}
+            for err in exc.errors():
+                loc = err.get("loc", ())
+                field = "non_field_errors"
+                if loc and loc[0] != "__root__":
+                    field = ".".join(map(str, loc))
+                msg = err.get("msg", "")
+                if msg.lower().startswith("value error, "):
+                    msg = msg[len("Value error, "):]
+                err_map.setdefault(field, []).append(msg)
+            raise serializers.ValidationError(err_map)
+
+        normalized = model.model_dump()
+
+        normalized.pop("files", None)
+
+        if self._pydantic_keep_keys is not None:
+            normalized = {k: v for k, v in normalized.items() if k in self._pydantic_keep_keys}
+
+        return {**attrs, **normalized}

backend/apps/ifc_validation/api/v1/schemas.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+from typing import List, Optional
+from pydantic import BaseModel, field_validator
+from django.core.files.uploadedfile import UploadedFile
+from core.settings import MAX_FILE_SIZE_IN_MB
+
+class _Base(BaseModel):
+    model_config = {"arbitrary_types_allowed": True}
+
+# note: by convention, this class is suffixed with "In" to indicate it is used for input validation
+class ValidationRequestIn(_Base):
+    file: Optional[UploadedFile] = None
+    files: Optional[List[UploadedFile]] = None
+    file_name: Optional[str] = None
+    size: Optional[int] = None
+
+    @field_validator("file")
+    @classmethod
+    def file_required_and_under_limit(cls, v: Optional[UploadedFile]):
+        if v is None:
+            raise ValueError("File is required.")
+        max_bytes = MAX_FILE_SIZE_IN_MB * 1024 * 1024
+        if v.size > max_bytes:
+            raise ValueError(f"File size exceeds allowed file size limit ({MAX_FILE_SIZE_IN_MB} MB).")
+        return v
+
+    @field_validator("file_name")
+    @classmethod
+    def file_name_ifc(cls, v: Optional[str]):
+        if not v:
+            raise ValueError("File name is required.")
+        if not v.lower().endswith(".ifc"):
+            raise ValueError("File name must end with '.ifc'.")
+        return v
+
+    @field_validator("size")
+    @classmethod
+    def size_positive_and_under_limit(cls, v: Optional[int]):
+        if v is None:
+            return v
+        if v <= 0:
+            raise ValueError("Size must be positive.")
+        max_bytes = MAX_FILE_SIZE_IN_MB * 1024 * 1024
+        if v > max_bytes:
+            raise ValueError(f"File size exceeds allowed file size limit ({MAX_FILE_SIZE_IN_MB} MB).")
+        return v

backend/apps/ifc_validation/api/v1/serializers.py

@@ -5,7 +5,8 @@
 from apps.ifc_validation_models.models import ValidationOutcome
 from apps.ifc_validation_models.models import Model
 
-from core.settings import MAX_FILE_SIZE_IN_MB
+from .pydantic_bridge import PydanticValidatorMixin
+from .schemas import ValidationRequestIn
 
 class BaseSerializer(serializers.ModelSerializer):
 
@@ -27,7 +28,9 @@ class Meta:
         abstract = True
 
 
-class ValidationRequestSerializer(BaseSerializer):
+class ValidationRequestSerializer(PydanticValidatorMixin, BaseSerializer):
+
+    Schema = ValidationRequestIn
 
     class Meta:
         model = ValidationRequest
@@ -36,70 +39,26 @@ class Meta:
         hide = ["id", "model", "deleted", "created_by", "updated_by", "status_reason"]
         read_only_fields = ['size', 'created_by', 'updated_by']
 
-    def validate_file(self, value):
 
-        # ensure file is not empty
-        if not value:
-            raise serializers.ValidationError("File is required.")
-
-        # ensure size is under MAX_FILE_SIZE_IN_MB
-        if value.size > MAX_FILE_SIZE_IN_MB * 1024 * 1024:
-            raise serializers.ValidationError(f"File size exceeds allowed file size limit ({MAX_FILE_SIZE_IN_MB} MB).")
-
-        return value
+class ValidationTaskSerializer(PydanticValidatorMixin, BaseSerializer):
 
-    def validate_files(self, value):
-
-        # ensure exactly one file is uploaded
-        if len(value) > 1:
-            raise serializers.ValidationError({"file": "Only one file can be uploaded at a time."})
-
-        return value
-
-    def validate_file_name(self, value):
-
-        # ensure file name is not empty
-        if not value:
-            raise serializers.ValidationError("File name is required.")
-
-        # ensure file name ends with .ifc
-        if not value.lower().endswith('.ifc'):
-            raise serializers.ValidationError(f"File name must end with '.ifc'.")
-
-        return value
-
-    def validate_size(self, value):
-
-        # ensure size is positive
-        if value <= 0:
-            raise serializers.ValidationError("Size must be positive.")
-
-        # ensure size is under MAX_FILE_SIZE_IN_MB
-        if value > MAX_FILE_SIZE_IN_MB * 1024 * 1024:
-            raise serializers.ValidationError(f"File size exceeds allowed file size limit ({MAX_FILE_SIZE_IN_MB} MB).")
-
-        return value
-
-
-class ValidationTaskSerializer(BaseSerializer):
-
     class Meta:
         model = ValidationTask
         fields = '__all__'
         show = ["public_id", "request_public_id"]
         hide = ["id", "process_id", "process_cmd", "request"]
 
 
-class ValidationOutcomeSerializer(BaseSerializer):
-
+class ValidationOutcomeSerializer(PydanticValidatorMixin, BaseSerializer):
+    
     class Meta:
         model = ValidationOutcome
         fields = '__all__'
         show = ["public_id", "instance_public_id", "validation_task_public_id"]
         hide = ["id", "instance", "validation_task"]
 
 
-class ModelSerializer(BaseSerializer):
+class ModelSerializer(PydanticValidatorMixin, BaseSerializer):
 
     class Meta:
         model = Model

backend/apps/ifc_validation/api/v1/tests/test_pydantic.py

@@ -0,0 +1,48 @@
+"""
+Initial tests moving away from Django towards FastAPI/Pydantic    
+"""
+import pytest
+from pydantic import ValidationError, TypeAdapter
+from django.core.files.uploadedfile import SimpleUploadedFile # to be deleted after making schema DjangoDRF/FastAPI agnostic
+
+from apps.ifc_validation.api.v1.schemas import ValidationRequestIn
+from core.settings import MAX_FILE_SIZE_IN_MB
+
+TA = TypeAdapter(ValidationRequestIn)
+
+def make_uploaded(name: str, size: int) -> SimpleUploadedFile:
+    return SimpleUploadedFile(
+        name=name,
+        content=b"\x00" * size,
+        content_type="application/octet-stream",
+    )
+
+def test_valid_file_passes():
+    f = make_uploaded("ok.ifc", 100)
+    m = TA.validate_python({"file": f, "file_name": "ok.ifc", "size": 100})
+    assert m.file_name == "ok.ifc"
+    assert m.file.size == 100
+
+def test_missing_file_raises():
+    with pytest.raises(ValidationError) as e:
+        TA.validate_python({"file": None, "file_name": "ok.ifc", "size": 1})
+    assert "File is required." in str(e.value)
+
+def test_bad_extension_raises():
+    f = make_uploaded("bad.txt", 100)
+    with pytest.raises(ValidationError) as e:
+        TA.validate_python({"file": f, "file_name": "bad.txt", "size": 100})
+    assert "File name must end with '.ifc'." in str(e.value)
+
+def test_negative_size_raises():
+    f = make_uploaded("ok.ifc", 100)
+    with pytest.raises(ValidationError) as e:
+        TA.validate_python({"file": f, "file_name": "ok.ifc", "size": -5})
+    assert "Size must be positive." in str(e.value)
+
+def test_too_large_file_raises():
+    too_big = (MAX_FILE_SIZE_IN_MB + 1) * 1024 * 1024
+    f = make_uploaded("ok.ifc", too_big)
+    with pytest.raises(ValidationError) as e:
+        TA.validate_python({"file": f, "file_name": "ok.ifc", "size": too_big})
+    assert f"File size exceeds allowed file size limit ({MAX_FILE_SIZE_IN_MB} MB)." in str(e.value)

backend/apps/ifc_validation/api/v1/views.py

@@ -153,7 +153,8 @@ def post(self, request, *args, **kwargs):
 
                     # only accept one file (for now) - note: can't be done easily in serializer,
                     # as we need access to request.FILES and our model only accepts one file
-                    serializer.validate_files(files)
+                    if len(files) > 1:
+                        raise serializers.ValidationError({'file': 'Only one file can be uploaded at a time.'})
 
                     # retrieve file size and save
                     uploaded_file = serializer.validated_data