Refactor FileView

Co-authored-by: VeryBigSad <VeryBigSad@users.noreply.github.com>

Author: yakser

files/helpers.py

@@ -0,0 +1,86 @@
+import requests
+import time
+from files.exceptions import SelectelUploadError
+
+from procollab.settings import (
+    DEBUG,
+    SELECTEL_ACCOUNT_ID,
+    SELECTEL_CONTAINER_NAME,
+    SELECTEL_CONTAINER_PASSWORD,
+    SELECTEL_CONTAINER_USERNAME,
+)
+
+
+class FileAPI:
+    def __init__(self, file, user) -> None:
+        self.file = file
+        self.user = user
+
+    @staticmethod
+    def delete(url: str) -> int:
+        """Deletes file from selcdn"""
+        token = FileAPI._get_selectel_swift_token()
+        response = requests.delete(url, headers={"X-Auth-Token": token})
+        return response.status_code
+
+    def upload(self):
+        return self._upload_via_selectel_swift()
+
+    def _upload_via_selectel_swift(self) -> tuple[int, str]:
+        token = self._get_selectel_swift_token()
+        url = self._generate_selectel_swift_file_url()
+
+        with self.file.open(mode="rb") as file_object:
+            response = requests.put(
+                url,
+                headers={"X-Auth-Token": token, "Content-Type": file_object.content_type},
+                data=file_object.read(),
+            )
+
+        return response.status_code, url
+
+    def _generate_selectel_swift_link(sefl):
+        link = f"https://api.selcdn.ru/v1/SEL_{SELECTEL_ACCOUNT_ID}/{SELECTEL_CONTAINER_NAME}/"
+        if DEBUG:
+            link += "debug/"
+        return link
+
+    @staticmethod
+    def _get_selectel_swift_token():
+        """Returns auth token for selcdn"""
+        data = {
+            "auth": {
+                "identity": {
+                    "methods": ["password"],
+                    "password": {
+                        "user": {
+                            "id": SELECTEL_CONTAINER_USERNAME,
+                            "password": SELECTEL_CONTAINER_PASSWORD,
+                        }
+                    },
+                }
+            }
+        }
+        r = requests.post("https://api.selcdn.ru/v3/auth/tokens", json=data)
+        if r.status_code not in [200, 201]:
+            raise SelectelUploadError("Couldn't generate a token for selcdn")
+        return r.headers["x-subject-token"]
+
+    def _get_file_extension(self) -> str:
+        if len(self.file.name.split(".")) > 1:
+            return "." + self.file.name.split(".")[1]
+        return ""
+
+    def _generate_selectel_swift_file_url(self) -> str:
+        """
+        Generates url for selcdn
+
+        Returns:
+            url: str looks like /hashedEmail/hashedFilename_hashedTime.extension
+        """
+        link = self._generate_selectel_swift_link()
+        extension = self._get_file_extension()
+        return (
+            link
+            + f"{abs(hash(self.user.email))}/{abs(hash(self.file.name))}_{abs(hash(time.time()))}{extension}"
+        )

files/models.py

@@ -1,5 +1,5 @@
-from django.db import models
 from django.contrib.auth import get_user_model
+from django.db import models
 
 User = get_user_model()
 
@@ -18,9 +18,5 @@ class UserFile(models.Model):
     link = models.URLField(primary_key=True, null=False)
     datetime_uploaded = models.DateTimeField(auto_now_add=True)
 
-    def delete(self, using=None, keep_parents=False):
-        # TODO: add request to CDN to delete the object
-        super(UserFile, self).delete(using=using, keep_parents=keep_parents)
-
     def __str__(self):
         return f"UserFile by {self.user}, {self.link}"

files/urls.py

@@ -1,11 +1,9 @@
 from django.urls import path
 
-
 from files.views import FileView
 
 app_name = "industries"
 
 urlpatterns = [
     path("", FileView.as_view()),
-    path("<int:pk>", FileView.as_view()),
 ]

files/views.py

@@ -1,97 +1,22 @@
-import time
-
-import requests
 from django.db import transaction
 from rest_framework import permissions, status
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from files.exceptions import SelectelUploadError
+from files.helpers import FileAPI
 from files.models import UserFile
-from procollab.settings import (
-    DEBUG,
-    SELECTEL_ACCOUNT_ID,
-    SELECTEL_CONTAINER_NAME,
-    SELECTEL_CONTAINER_PASSWORD,
-    SELECTEL_CONTAINER_USERNAME,
-)
 
 
 class FileView(APIView):
     permission_classes = [permissions.AllowAny]
 
     @transaction.atomic
     def post(self, request):
-        if DEBUG is True:
-            return Response(
-                {"message": "Files doesn't save in development mode, sorry <3"},
-                status=status.HTTP_406_NOT_ACCEPTABLE,
-            )
-
-        file = request.FILES["file"]
-        link = f"https://api.selcdn.ru/v1/SEL_{SELECTEL_ACCOUNT_ID}/{SELECTEL_CONTAINER_NAME}/"
-        user = request.user
-        token = self._get_token()
-
-        if len(file.name.split(".")) > 1:
-            extension = file.name.split(".")[1]
-        else:
-            extension = ""
-
-        # looks like /hashedEmail/hashedFilename_hashedTime.extension
-        url = (
-            link + f"{SELECTEL_CONTAINER_NAME}/{abs(hash(user.email))}/"
-            f"{abs(hash(file.name))}_{abs(hash(time.time()))}{'.' + extension if extension else ''}"
-        )
-        with file.open(mode="rb") as file_object:
-            r = requests.put(
-                url,
-                headers={"X-Auth-Token": token, "Content-Type": file_object.content_type},
-                data=file_object.read(),
-            )
-        if r.status_code != 201:
-            return Response("Failed to upload file", status=status.HTTP_409_CONFLICT)
-        self._save_to_db(user, url)
-        return Response({"url": url}, status=status.HTTP_201_CREATED)
-
-    def delete(self, request, pk):
-        try:
-            UserFile.objects.get(pk=pk).delete()
-        except UserFile.DoesNotExist:
-            return Response(status=status.HTTP_404_NOT_FOUND)
-        return Response(status=status.HTTP_200_OK)
+        file_api = FileAPI(request.FILES["file"], request.user)
+        status_code, url = file_api.upload()
 
-    @classmethod
-    def _save_to_db(cls, user, url):
-        """creates userfile object for file uploads"""
-        return UserFile.objects.create(user=user, link=url)
+        if status_code == 201:
+            UserFile.objects.create(user=request.user, link=url)
+            return Response({"url": url}, status=status.HTTP_201_CREATED)
 
-    @classmethod
-    def _get_token(cls):
-        """returns auth token for sentry"""
-        data = {
-            "auth": {
-                "identity": {
-                    "methods": ["password"],
-                    "password": {
-                        "user": {
-                            "id": SELECTEL_CONTAINER_USERNAME,
-                            "password": SELECTEL_CONTAINER_PASSWORD,
-                        }
-                    },
-                }
-            }
-        }
-        r = requests.post("https://api.selcdn.ru/v3/auth/tokens", json=data)
-        if r.status_code not in [200, 201]:
-            raise SelectelUploadError("couldn't generate a token for selcdn")
-        return r.headers["x-subject-token"]
-        # async with server.post(
-        #     "https://api.selcdn.ru/v3/auth/tokens",
-        #     data=json.dumps(data),
-        # ) as response:
-        #     if response.status != 201:
-        #         return Response(
-        #             "Failed to get token", status_code=status.HTTP_409_CONFLICT
-        #         )
-        #     return response.json()
+        return Response("Failed to upload file", status=status.HTTP_409_CONFLICT)