add django backend in /core

ported from laravel in c3lf/lfbackend repo

core/files/admin.py

@@ -0,0 +1,10 @@
+from django.contrib import admin
+
+from files.models import File
+
+
+class FileAdmin(admin.ModelAdmin):
+    pass
+
+
+admin.site.register(File, FileAdmin)

core/files/api_v1.py

@@ -0,0 +1,25 @@
+from rest_framework import serializers, viewsets, routers
+
+from files.models import File
+
+
+class FileSerializer(serializers.ModelSerializer):
+    data = serializers.CharField(max_length=1000000, write_only=True)
+
+    class Meta:
+        model = File
+        fields = ['hash', 'data']
+        read_only_fields = ['hash']
+
+
+class FileViewSet(viewsets.ModelViewSet):
+    serializer_class = FileSerializer
+    queryset = File.objects.all()
+    lookup_field = 'hash'
+
+
+router = routers.SimpleRouter()
+router.register(r'files', FileViewSet, basename='files')
+router.register(r'file', FileViewSet, basename='files')
+
+urlpatterns = router.urls

core/files/media_urls.py

@@ -0,0 +1,57 @@
+from coverage.annotate import os
+from django.http import HttpResponse
+from django.urls import path
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import status
+from rest_framework.decorators import api_view
+from rest_framework.response import Response
+
+from core.settings import MEDIA_ROOT
+from files.models import File
+
+
+@swagger_auto_schema(method='GET', auto_schema=None)
+@api_view(['GET'])
+def media_urls(request, hash_path):
+    try:
+        file = File.objects.get(file=hash_path)
+
+        return HttpResponse(status=status.HTTP_200_OK,
+                            content_type=file.mime_type,
+                            headers={
+                                'X-Accel-Redirect': f'/redirect_media/{hash_path}',
+                                'Access-Control-Allow-Origin': '*',
+                            })  # TODO Expires and Cache-Control
+
+    except File.DoesNotExist:
+        return Response(status=status.HTTP_404_NOT_FOUND)
+
+
+@swagger_auto_schema(method='GET', auto_schema=None)
+@api_view(['GET'])
+def thumbnail_urls(request, size, hash_path):
+    if size not in [32, 64, 256]:
+        return Response(status=status.HTTP_404_NOT_FOUND)
+    try:
+        file = File.objects.get(file=hash_path)
+        if not os.path.exists(MEDIA_ROOT + f'/thumbnails/{size}/{hash_path}'):
+            from PIL import Image
+            iamge = Image.open(file.file)
+            iamge.thumbnail((size, size))
+            iamge.save(MEDIA_ROOT + f'/media/thumbnails/{size}/{hash_path}', quality=90)
+
+        return HttpResponse(status=status.HTTP_200_OK,
+                            content_type=file.mime_type,
+                            headers={
+                                'X-Accel-Redirect': f'/redirect_thumbnail/{size}/{hash_path}',
+                                'Access-Control-Allow-Origin': '*',
+                            })  # TODO Expires and Cache-Control
+
+    except File.DoesNotExist:
+        return Response(status=status.HTTP_404_NOT_FOUND)
+
+
+urlpatterns = [
+    path('<int:size>/<path:hash_path>', thumbnail_urls),
+    path('<path:hash_path>', media_urls),
+]

core/files/migrations/0001_initial.py

@@ -0,0 +1,30 @@
+# Generated by Django 4.2.7 on 2023-11-18 11:28
+
+from django.db import migrations, models
+import django.db.models.deletion
+import files.models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('inventory', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='File',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(blank=True, null=True)),
+                ('updated_at', models.DateTimeField(blank=True, null=True)),
+                ('deleted_at', models.DateTimeField(blank=True, null=True)),
+                ('file', models.ImageField(upload_to=files.models.hash_upload)),
+                ('mime_type', models.CharField(max_length=255)),
+                ('hash', models.CharField(max_length=64, unique=True)),
+                ('item', models.ForeignKey(blank=True, db_column='iid', null=True, on_delete=django.db.models.deletion.CASCADE, related_name='files', to='inventory.item')),
+            ],
+        ),
+    ]

core/files/models.py

@@ -0,0 +1,57 @@
+from django.core.files.base import ContentFile
+from django.db import models, IntegrityError
+from django_softdelete.models import SoftDeleteModel
+
+from inventory.models import Item
+
+
+def hash_upload(instance, filename):
+    return f"{instance.hash[:2]}/{instance.hash[2:4]}/{instance.hash[4:6]}/{instance.hash[6:]}"
+
+
+class FileManager(models.Manager):
+    def get_or_create(self, **kwargs):
+        if 'data' in kwargs and type(kwargs['data']) == str:
+            import base64
+            from hashlib import sha256
+            content = base64.b64decode(kwargs['data'], validate=True)
+            kwargs.pop('data')
+            content_hash = sha256(content).hexdigest()
+            kwargs['file'] = ContentFile(content, content_hash)
+            kwargs['hash'] = content_hash
+        else:
+            raise ValueError('data must be a base64 encoded string or file and hash must be provided')
+        try:
+            return self.get(hash=kwargs['hash']), False
+        except self.model.DoesNotExist:
+            return self.create(**kwargs), True
+
+    def create(self, **kwargs):
+        if 'data' in kwargs and type(kwargs['data']) == str:
+            import base64
+            from hashlib import sha256
+            content = base64.b64decode(kwargs['data'], validate=True)
+            kwargs.pop('data')
+            content_hash = sha256(content).hexdigest()
+            kwargs['file'] = ContentFile(content, content_hash)
+            kwargs['hash'] = content_hash
+        elif 'file' in kwargs and 'hash' in kwargs and type(kwargs['file']) == ContentFile:
+            pass
+        else:
+            raise ValueError('data must be a base64 encoded string or file and hash must be provided')
+        if not self.filter(hash=kwargs['hash']).exists():
+            return super().create(**kwargs)
+        else:
+            raise IntegrityError('File with this hash already exists')
+
+
+class File(models.Model):
+    item = models.ForeignKey(Item, models.CASCADE, db_column='iid', null=True, blank=True, related_name='files')
+    created_at = models.DateTimeField(blank=True, null=True)
+    updated_at = models.DateTimeField(blank=True, null=True)
+    deleted_at = models.DateTimeField(blank=True, null=True)
+    file = models.ImageField(upload_to=hash_upload)
+    mime_type = models.CharField(max_length=255, null=False, blank=False)
+    hash = models.CharField(max_length=64, null=False, blank=False, unique=True)
+
+    objects = FileManager()

core/files/tests.py

@@ -0,0 +1,47 @@
+from django.test import TestCase, Client
+
+from files.models import File
+from inventory.models import Event, Container, Item
+
+client = Client()
+
+
+class FileTestCase(TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.event = Event.objects.create(slug='EVENT', name='Event')
+        self.box = Container.objects.create(name='BOX')
+
+    def test_list_files(self):
+        import base64
+        item = File.objects.create(data=base64.b64encode(b"foo").decode('utf-8'))
+        response = client.get('/1/files/')
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()[0]['hash'], item.hash)
+        self.assertEqual(len(response.json()[0]['hash']), 64)
+
+    def test_one_file(self):
+        import base64
+        item = File.objects.create(data=base64.b64encode(b"foo").decode('utf-8'))
+        response = client.get(f'/1/file/{item.hash}/')
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()['hash'], item.hash)
+        self.assertEqual(len(response.json()['hash']), 64)
+
+    def test_create_file(self):
+        import base64
+        Item.objects.create(container=self.box, event=self.event, description='1')
+        item = Item.objects.create(container=self.box, event=self.event, description='2')
+        response = client.post('/1/file/', {'data': base64.b64encode(b"foo").decode('utf-8')}, content_type='application/json')
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(len(response.json()['hash']), 64)
+
+    def test_delete_file(self):
+        import base64
+        item = Item.objects.create(container=self.box, event=self.event, description='1')
+        File.objects.create(item=item, data=base64.b64encode(b"foo").decode('utf-8'))
+        file = File.objects.create(item=item, data=base64.b64encode(b"bar").decode('utf-8'))
+        self.assertEqual(len(File.objects.all()), 2)
+        response = client.delete(f'/1/file/{file.hash}/')
+        self.assertEqual(response.status_code, 204)