Merge pull request #68 from mapswipe/feature/assest-validations

Author: tnagorra

apps/common/models.py

@@ -32,12 +32,12 @@ class UserResource(Model):
     )
     created_at = models.DateTimeField(auto_now_add=True)
     modified_at = models.DateTimeField(auto_now=True)
-    created_by = models.ForeignKey(
+    created_by: User = models.ForeignKey(  # type: ignore[reportIncompatibleVariableOverride]
         User,
         related_name="%(class)s_created",
         on_delete=models.PROTECT,
     )
-    modified_by = models.ForeignKey(
+    modified_by: User = models.ForeignKey(  # type: ignore[reportIncompatibleVariableOverride]
         User,
         related_name="%(class)s_modified",
         on_delete=models.PROTECT,
@@ -60,7 +60,7 @@ def __str__(self):
 class ArchivableResource(Model):
     is_archived = models.BooleanField(default=False)
     archived_at = models.DateTimeField(null=True, blank=True)
-    archived_by = models.ForeignKey(
+    archived_by: User = models.ForeignKey(  # type: ignore[reportIncompatibleVariableOverride]
         User,
         related_name="+",
         on_delete=models.SET_NULL,
@@ -161,3 +161,75 @@ def update_firebase_push_status(
 
     class Meta(TypedModelMeta):  # type: ignore[reportIncompatibleVariableOverride]
         abstract = True
+
+
+class AssetMimetypeEnum(models.IntegerChoices):
+    GEOJSON = 100, "application/geo+json"
+
+    IMAGE_JPEG = 201, "image/jpeg"
+    IMAGE_PNG = 202, "image/png"
+    IMAGE_GIF = 203, "image/gif"
+
+    @classmethod
+    def get_display(cls, value: typing.Self | int) -> str:
+        if value in cls:
+            return str(cls(value).label)
+        return "Unknown"
+
+    @classmethod
+    def is_valid_mimetype(cls, mimetype: str) -> bool:
+        """
+        Check if the given mimetype is valid for project assets.
+        """
+        return mimetype in [choice.label for choice in cls]
+
+    @classmethod
+    def get_mimetype_by_label(cls, label: str) -> typing.Self | None:
+        for choice in cls:
+            if choice.label == label:
+                return choice
+        return None
+
+
+# FIXME(tnagorra): Finalize the enum labels
+class AssetTypeEnum(models.IntegerChoices):
+    INPUT = 100, "Input"
+    OUTPUT = 200, "Output"
+    STATS = 300, "Stats"
+
+    @classmethod
+    def get_display(cls, value: typing.Self | int) -> str:
+        if value in cls:
+            return str(cls(value).label)
+        return "Unknown"
+
+
+class CommonAsset(Model):
+    Mimetype = AssetMimetypeEnum
+    MAX_FILE_SIZE: int = 10 * 1024 * 1024  # MB
+    Type = AssetTypeEnum
+
+    type = IntegerChoicesField(
+        choices_enum=AssetTypeEnum,
+    )
+
+    mimetype = IntegerChoicesField(
+        choices_enum=AssetMimetypeEnum,
+    )
+
+    file_size = models.PositiveIntegerField(
+        help_text=gettext_lazy("The size of the file in bytes"),
+    )
+
+    marked_as_deleted = models.BooleanField(
+        default=False,
+        help_text=gettext_lazy("If this flag is enabled, this project asset will be deleted in the future"),
+    )
+
+    class Meta(TypedModelMeta):  # type: ignore[reportIncompatibleVariableOverride]
+        abstract = True
+
+    @classmethod
+    def usable_objects(cls):
+        """Returns objects that are mot marked for deletion"""
+        return cls.objects.filter(marked_as_deleted=False)

apps/common/serializers.py

@@ -1,14 +1,17 @@
 import logging
+import mimetypes
 import typing
 
 from django.core.exceptions import ValidationError
 from django.http.request import HttpRequest
+from django.template.defaultfilters import filesizeformat
 from django.utils import timezone
 from django.utils.translation import gettext
 from firebase_admin import auth
 from rest_framework import serializers
 
 from apps.common.models import ArchivableResource, UserResource
+from apps.project.models import CommonAsset
 from apps.user.models import User
 from utils.common import validate_ulid
 
@@ -17,6 +20,10 @@
 logger = logging.getLogger(__name__)
 
 
+if typing.TYPE_CHECKING:
+    from django.core.files.base import ContentFile
+
+
 class DrfContextType(typing.TypedDict):
     request: HttpRequest
 
@@ -160,3 +167,46 @@ def validate(self, attrs: dict[str, typing.Any]):
 class FirebaseAuthResponseSerializer(serializers.Serializer[typing.Any]):
     ok = serializers.BooleanField(required=True)
     error = serializers.CharField()
+
+
+class CommonAssetSerializer(serializers.ModelSerializer[CommonAsset]):
+    def _validate_file(self, attrs: dict[str, typing.Any]) -> None:
+        file_content: ContentFile[bytes] = attrs["file"]
+        file_size = file_content.size
+        max_file_size = CommonAsset.MAX_FILE_SIZE
+
+        if file_size > max_file_size:
+            raise serializers.ValidationError(
+                gettext("Filesize should be less than: %s. Current is: %s")
+                % (
+                    filesizeformat(max_file_size),
+                    filesizeformat(file_size),
+                ),
+            )
+
+        # https://docs.python.org/3/library/mimetypes.html#mimetypes.guess_type
+        mimetype, *_ = mimetypes.guess_type(file_content.name)  # type: ignore[reportUnknownArgumentType]
+
+        # TODO(susilnem): Use library like filemagic to determine mimetype instead?
+        if mimetype is None:
+            raise serializers.ValidationError(
+                gettext("Could not determine mimetype of the file: %s") % file_content.name,
+            )
+
+        if not CommonAsset.Mimetype.is_valid_mimetype(mimetype):
+            raise serializers.ValidationError(
+                gettext("File mimetype is not supported: %s") % mimetype,
+            )
+
+        if "mimetype" in attrs and CommonAsset.Mimetype.get_mimetype_by_label(mimetype) != attrs["mimetype"]:
+            raise serializers.ValidationError(
+                gettext("File mimetype does not match the provided mimetype: %s") % mimetype,
+            )
+
+        attrs["mimetype"] = CommonAsset.Mimetype.get_mimetype_by_label(mimetype)
+        attrs["file_size"] = file_size
+
+    @typing.override
+    def validate(self, attrs: dict[str, typing.Any]) -> dict[str, typing.Any]:
+        self._validate_file(attrs)
+        return super().validate(attrs)

apps/community_dashboard/models.py

@@ -32,8 +32,8 @@ def __str__(self):
 
 class AggregatedUserStatData(Model):
     # Ref Fields
-    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="+")
-    user = models.ForeignKey(ContributorUser, on_delete=models.CASCADE, related_name="+")
+    project: Project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="+")  # type: ignore[reportIncompatibleVariableOverride]
+    user: ContributorUser = models.ForeignKey(ContributorUser, on_delete=models.CASCADE, related_name="+")  # type: ignore[reportIncompatibleVariableOverride]
     timestamp_date = models.DateField()
     # Aggregated Fields
     total_time = models.IntegerField()  # seconds
@@ -59,9 +59,9 @@ def __str__(self):
 
 class AggregatedUserGroupStatData(Model):
     # Ref Fields
-    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="+")
-    user = models.ForeignKey(ContributorUser, on_delete=models.CASCADE, related_name="+")
-    user_group = models.ForeignKey(ContributorUserGroup, on_delete=models.CASCADE, related_name="+")
+    project: Project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="+")  # type: ignore[reportIncompatibleVariableOverride]
+    user: ContributorUser = models.ForeignKey(ContributorUser, on_delete=models.CASCADE, related_name="+")  # type: ignore[reportIncompatibleVariableOverride]
+    user_group: ContributorUserGroup = models.ForeignKey(ContributorUserGroup, on_delete=models.CASCADE, related_name="+")  # type: ignore[reportIncompatibleVariableOverride]
     timestamp_date = models.DateField()
     # Aggregated Fields
     total_time = models.IntegerField()  # seconds

apps/contributor/models.py

@@ -19,7 +19,7 @@ class ContributorUser(models.Model):
         unique=True,
         help_text="Firebase User ID",
     )
-    team = models.ForeignKey(
+    team: "ContributorTeam | None" = models.ForeignKey(  # type: ignore[reportIncompatibleVariableOverride]
         "ContributorTeam",
         on_delete=models.SET_NULL,
         null=True,
@@ -46,8 +46,8 @@ def __str__(self):
 
 
 class ContributorUserGroupMembership(models.Model):
-    user_group = models.ForeignKey(ContributorUserGroup, on_delete=models.CASCADE)
-    user = models.ForeignKey(ContributorUser, on_delete=models.CASCADE)
+    user_group: ContributorUserGroup = models.ForeignKey(ContributorUserGroup, on_delete=models.CASCADE)  # type: ignore[reportIncompatibleVariableOverride]
+    user: ContributorUser = models.ForeignKey(ContributorUser, on_delete=models.CASCADE)  # type: ignore[reportIncompatibleVariableOverride]
     is_active = models.BooleanField()
 
     # Type hints
@@ -67,7 +67,7 @@ class ContributorUserGroupMembershipLogActionEnum(models.IntegerChoices):
 class ContributorUserGroupMembershipLog(models.Model):
     ACTION = ContributorUserGroupMembershipLogActionEnum
 
-    membership = models.ForeignKey(ContributorUserGroupMembership, on_delete=models.CASCADE)
+    membership: ContributorUserGroupMembership = models.ForeignKey(ContributorUserGroupMembership, on_delete=models.CASCADE)  # type: ignore[reportIncompatibleVariableOverride]
     # Sync with firebase
     action = IntegerChoicesField(choices_enum=ContributorUserGroupMembershipLogActionEnum)
     date = models.DateTimeField()

apps/mapping/models.py

@@ -28,8 +28,8 @@ class MappingSession(models.Model):
     # FIXME(tnagorra): We might need to skip the indexing
     old_id = models.CharField(max_length=30, db_index=True, null=True)
 
-    project_task_group = models.ForeignKey(ProjectTaskGroup, on_delete=models.PROTECT)
-    contributor_user = models.ForeignKey(ContributorUser, on_delete=models.PROTECT)
+    project_task_group: ProjectTaskGroup = models.ForeignKey(ProjectTaskGroup, on_delete=models.PROTECT)  # type: ignore[reportIncompatibleVariableOverride]
+    contributor_user: ContributorUser = models.ForeignKey(ContributorUser, on_delete=models.PROTECT)  # type: ignore[reportIncompatibleVariableOverride]
 
     app_version = models.CharField(max_length=10)
     client_type = IntegerChoicesField(choices_enum=MappingSessionClientTypeEnum)
@@ -54,8 +54,8 @@ class MappingSessionUserGroup(models.Model):
     # FIXME(tnagorra): We might need to skip the indexing
     old_id = models.CharField(max_length=30, db_index=True, null=True)
 
-    mapping_session = models.ForeignKey(MappingSession, on_delete=models.PROTECT)
-    user_group = models.ForeignKey(
+    mapping_session: MappingSession = models.ForeignKey(MappingSession, on_delete=models.PROTECT)  # type: ignore[reportIncompatibleVariableOverride]
+    user_group: ContributorUserGroup = models.ForeignKey(  # type: ignore[reportIncompatibleVariableOverride]
         ContributorUserGroup,
         on_delete=models.PROTECT,
         related_name="+",
@@ -73,8 +73,8 @@ class MappingSessionResult(models.Model):
     # FIXME(tnagorra): We might need to skip the indexing
     old_id = models.CharField(max_length=30, db_index=True, null=True)
 
-    session = models.ForeignKey(MappingSession, on_delete=models.PROTECT)
-    project_task = models.ForeignKey(ProjectTask, on_delete=models.PROTECT)
+    session: MappingSession = models.ForeignKey(MappingSession, on_delete=models.PROTECT)  # type: ignore[reportIncompatibleVariableOverride]
+    project_task: ProjectTaskGroup = models.ForeignKey(ProjectTask, on_delete=models.PROTECT)  # type: ignore[reportIncompatibleVariableOverride]
     result = models.PositiveSmallIntegerField()
 
     # TODO(thenav56): Add constraint to make sure we have non-duplicate row with task_id, .session.user_id

apps/project/migrations/0001_initial.py

@@ -76,8 +76,8 @@ class Migration(migrations.Migration):
                 ('client_id', models.CharField(editable=False, max_length=26, unique=True, validators=[apps.common.models.validate_ulid])),
                 ('created_at', models.DateTimeField(auto_now_add=True)),
                 ('modified_at', models.DateTimeField(auto_now=True)),
-                ('type', django_choices_field.fields.IntegerChoicesField(choices=[(100, 'Input'), (200, 'Output'), (300, 'Stats')], choices_enum=apps.project.models.ProjectAssetTypeEnum)),
-                ('mimetype', django_choices_field.fields.IntegerChoicesField(choices=[(100, 'application/geo+json'), (201, 'image/jpeg'), (202, 'image/png'), (203, 'image/gif')], choices_enum=apps.project.models.ProjectAssetMimetypeEnum)),
+                ('type', django_choices_field.fields.IntegerChoicesField(choices=[(100, 'Input'), (200, 'Output'), (300, 'Stats')], choices_enum=apps.common.models.AssetTypeEnum)),
+                ('mimetype', django_choices_field.fields.IntegerChoicesField(choices=[(100, 'application/geo+json'), (201, 'image/jpeg'), (202, 'image/png'), (203, 'image/gif')], choices_enum=apps.common.models.AssetMimetypeEnum)),
                 ('file', models.FileField(help_text='The file associated with the asset', upload_to=apps.project.models.UploadHelper.project_asset)),
                 ('marked_as_deleted', models.BooleanField(default=False, help_text='If this flag is enabled, this project asset will be deleted in the future')),
             ],

apps/project/migrations/0011_projectasset_file_size.py

@@ -0,0 +1,19 @@
+# Generated by Django 5.2.4 on 2025-07-29 04:37
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('project', '0010_merge_20250728_1145'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='projectasset',
+            name='file_size',
+            field=models.PositiveIntegerField(default=1, help_text='The size of the file in bytes'),
+            preserve_default=False,
+        ),
+    ]

apps/project/migrations/0012_merge_20250730_1323.py

@@ -0,0 +1,14 @@
+# Generated by Django 5.1.6 on 2025-07-30 13:23
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('project', '0011_project_canonical_id'),
+        ('project', '0011_projectasset_file_size'),
+    ]
+
+    operations = [
+    ]

apps/project/models.py

@@ -10,41 +10,14 @@
 from django.utils.translation import gettext_lazy
 from django_choices_field import IntegerChoicesField
 
-from apps.common.models import ArchivableResource, FirebasePushStatusEnum, FirebaseResource, UserResource
+from apps.common.models import ArchivableResource, CommonAsset, FirebasePushStatusEnum, FirebaseResource, UserResource
 from apps.contributor.models import ContributorTeam
 from utils.fields import validate_percentage
 
 if typing.TYPE_CHECKING:
     from apps.tutorial.models import Tutorial
 
 
-class ProjectAssetMimetypeEnum(models.IntegerChoices):
-    GEOJSON = 100, "application/geo+json"
-
-    IMAGE_JPEG = 201, "image/jpeg"
-    IMAGE_PNG = 202, "image/png"
-    IMAGE_GIF = 203, "image/gif"
-
-    @classmethod
-    def get_display(cls, value: typing.Self | int) -> str:
-        if value in cls:
-            return str(cls(value).label)
-        return "Unknown"
-
-
-# FIXME(tnagorra): Finalize the enum labels
-class ProjectAssetTypeEnum(models.IntegerChoices):
-    INPUT = 100, "Input"
-    OUTPUT = 200, "Output"
-    STATS = 300, "Stats"
-
-    @classmethod
-    def get_display(cls, value: typing.Self | int) -> str:
-        if value in cls:
-            return str(cls(value).label)
-        return "Unknown"
-
-
 class ProjectTypeEnum(models.IntegerChoices):
     FIND = 1, "Find"
     """ Find project type. Previously known as Classification / Build Area. """
@@ -407,39 +380,18 @@ def clean(self):
         #     self.requiredResults += group.requiredCount * group.numberOfTasks
 
 
-class ProjectAsset(UserResource):
-    Type = ProjectAssetTypeEnum
-    Mimetype = ProjectAssetMimetypeEnum
-
-    type = IntegerChoicesField(
-        choices_enum=ProjectAssetTypeEnum,
-    )
-
-    mimetype = IntegerChoicesField(
-        choices_enum=ProjectAssetMimetypeEnum,
-    )
-
-    file = models.FileField(
-        upload_to=UploadHelper.project_asset,
-        help_text=gettext_lazy("The file associated with the asset"),
-    )
-
+class ProjectAsset(UserResource, CommonAsset):  # type: ignore[reportIncompatibleVariableOverride]
     project: Project = models.ForeignKey(  # type: ignore[reportAssignmentType]
         Project,
         on_delete=models.CASCADE,
         related_name="+",
     )
 
-    marked_as_deleted = models.BooleanField(
-        default=False,
-        help_text=gettext_lazy("If this flag is enabled, this project asset will be deleted in the future"),
+    file = models.FileField(
+        upload_to=UploadHelper.project_asset,
+        help_text=gettext_lazy("The file associated with the asset"),
     )
 
-    @classmethod
-    def usable_objects(cls):
-        """Returns objects that are mot marked for deletion"""
-        return cls.objects.filter(marked_as_deleted=False)
-
     # Type hints
     project_id: int