[#16] Upload Image

backend/afs/settings.py

@@ -44,6 +44,7 @@
     "akvo.core_forms",
     "akvo.core_node",
     "akvo.core_data",
+    "akvo.core_storage",
     "akvo.utils",
 ]
 

backend/afs/urls.py

@@ -20,10 +20,9 @@
     path("api/", include("akvo.core_forms.urls"), name="core_forms"),
     path("api/", include("akvo.core_data.urls"), name="core_data"),
     path("api/", include("akvo.core_node.urls"), name="core_node"),
+    path("api/", include("akvo.core_storage.urls"), name="core_storage"),
     path("api/schema/", SpectacularAPIView.as_view(), name="schema"),
     path(
-        "api/doc/",
-        SpectacularSwaggerView.as_view(url_name="schema"),
-        name="swagger-ui"
+        "api/doc/", SpectacularSwaggerView.as_view(url_name="schema"), name="swagger-ui"
     ),
 ]

backend/akvo/core_storage/admin.py

@@ -0,0 +1,3 @@
+# from django.contrib import admin
+
+# Register your models here.

backend/akvo/core_storage/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class CoreStorageConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "akvo.core_storage"

backend/akvo/core_storage/functions.py

@@ -0,0 +1,23 @@
+from os import environ
+from uuid import uuid4
+from akvo.utils import storage
+
+storage = storage.Storage(storage_path=environ.get("STORAGE_PATH"))
+
+
+def generate_image_file(file, filename, folder="images"):
+    temp_file = open(f"./tmp/{filename}", "wb+")
+    for chunk in file.chunks():
+        temp_file.write(chunk)
+    storage.upload(file=f"./tmp/{filename}", filename=filename, folder=folder)
+    temp_file.close()
+
+
+def process_image(request):
+    file = request.FILES["file"]
+    extension = file.name.split(".")[-1]
+    original_filename = "-".join(file.name.split(".")[:-1])
+    original_filename = "_".join(original_filename.strip().split(" "))
+    new_filename = f"{original_filename}-{uuid4()}.{extension}"
+    generate_image_file(file, new_filename)
+    return new_filename

backend/akvo/core_storage/models.py

@@ -0,0 +1,3 @@
+# from django.db import models
+
+# Create your models here.

backend/akvo/core_storage/serializers.py

@@ -0,0 +1,7 @@
+from django.core.validators import FileExtensionValidator
+from rest_framework import serializers
+from akvo.utils.custom_serializer_fields import CustomFileField
+
+
+class UploadImagesSerializer(serializers.Serializer):
+    file = CustomFileField(validators=[FileExtensionValidator(["jpg", "png", "jpeg"])])

backend/akvo/core_storage/tests/tests.py

@@ -0,0 +1,45 @@
+import os
+from django.test import TestCase
+from akvo.utils import storage
+
+STORAGE_PATH = os.environ.get("STORAGE_PATH")
+
+
+def generate_image(filename: str, extension: str = "jpg"):
+    filename = f"./{filename}.{extension}"
+    f = open(filename, "a")
+    f.write("This is a test file!")
+    f.close()
+    return f"./{filename}"
+
+
+class ImageUploadTest(TestCase):
+    def test_image_upload(self):
+        filename = generate_image(filename="test", extension="png")
+        file = open(filename, "rb")
+        response = self.client.post(
+            "/api/upload/images/",
+            {"file": file},
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(list(response.json()), ["message", "file"])
+        uploaded_filename = response.json().get("file")
+        self.assertTrue(
+            storage.check(uploaded_filename),
+            "File exists",
+        )
+        os.remove(f"{STORAGE_PATH}/{uploaded_filename}")
+        os.remove(filename)
+
+    def test_wrong_extension_upload(self):
+        filename = generate_image(filename="test", extension="txt")
+        response = self.client.post(
+            "/api/upload/images/",
+            {"file": open(filename, "rb")},
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.json(),
+            "File extension “txt” is not allowed. Allowed extensions are: jpg, png, jpeg.",  # noqa
+        )
+        os.remove(filename)

backend/akvo/core_storage/urls.py

@@ -0,0 +1,6 @@
+from django.urls import re_path
+from .views import upload_images
+
+urlpatterns = [
+    re_path(r"upload/images", upload_images),
+]

backend/akvo/core_storage/views.py

@@ -0,0 +1,35 @@
+from .serializers import UploadImagesSerializer
+from rest_framework import serializers
+from drf_spectacular.utils import extend_schema, inline_serializer
+from rest_framework.decorators import api_view, parser_classes
+from rest_framework.parsers import MultiPartParser
+from rest_framework.response import Response
+from rest_framework import status
+from .functions import process_image
+from akvo.utils.custom_serializer_fields import validate_serializers_message
+
+
+@extend_schema(
+    tags=["Files"],
+    summary="Upload Images",
+    request=UploadImagesSerializer,
+    responses={
+        (200, "application/json"): inline_serializer(
+            "UploadImages", fields={"task_id": serializers.CharField()}
+        )
+    },
+)
+@api_view(["POST"])
+@parser_classes([MultiPartParser])
+def upload_images(request):
+    serializer = UploadImagesSerializer(data=request.data)
+    if not serializer.is_valid():
+        return Response(
+            validate_serializers_message(serializer.errors),
+            status=status.HTTP_400_BAD_REQUEST,
+        )
+    filename = process_image(request)
+    return Response(
+        {"message": "File uploaded successfully", "file": f"/images/{filename}"},
+        status=status.HTTP_200_OK,
+    )