import export

Author: ImToster

application/entrypoint.sh

@@ -4,7 +4,7 @@ python3 pay_parking/manage.py makemigrations
 python3 pay_parking/manage.py migrate
 if [ ! -e $CONTAINER_ALREADY_STARTED ]; then
     touch $CONTAINER_ALREADY_STARTED
-    python3 pay_parking/manage.py loaddata pay_parking/db.json 
+    python3 pay_parking/manage.py import pay_parking/db.json 
 fi
 python3 pay_parking/manage.py collectstatic
 cp -r /app/pay_parking/collected_static/. /backend_static/static/

application/pay_parking/db.json

@@ -0,0 +1,48 @@
+from django.contrib import admin
+from django.urls import path
+from .import_data import import_data
+from .export_data import export_data
+from django.http import HttpResponse
+from rest_framework import status
+from rest_framework.response import Response
+import json
+from .forms import ImportForm
+from django.shortcuts import redirect
+from django.contrib import messages
+
+
+class AdminSite(admin.AdminSite):
+    def get_urls(self):
+        new_urls = [
+            path('import/', self.admin_view(self.import_view), name='import'),
+            path('export/', self.admin_view(self.export_view), name='export')
+        ]
+        return new_urls + super().get_urls()
+
+    def import_view(self, request):
+        if request.method == 'POST':
+            form = ImportForm(request.POST, request.FILES)
+            if form.is_valid():
+                try:
+                    data = json.load(form.cleaned_data['import_file'])
+                    try:
+                        import_data(data)
+                        messages.success(request, 'Данные загружены и обновлены')
+                    except Exception as e:
+                        messages.error(request, e)
+                except Exception as e:
+                    messages.error(request, "Ошибка при чтении файла")
+                return redirect('admin:index')
+        return Response(status=status.HTTP_400_BAD_REQUEST)
+    
+    def export_view(self, request):
+        if request.method == 'GET':
+            data = export_data()
+            content = json.dumps(data, ensure_ascii=False)
+            response = HttpResponse(content, content_type='application/json')
+            response['Content-Disposition'] = 'attachment; filename="export_data.json"'
+            return response
+        return Response(status=status.HTTP_400_BAD_REQUEST)
+
+
+admin.site = AdminSite('')

application/pay_parking/import_export/__init__.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class ImportExportConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'import_export'

application/pay_parking/import_export/admin.py

@@ -0,0 +1,18 @@
+from import_export.serializers import ParkingSerializer, UserSerializer, PaymentSerializer
+
+
+def export_data():
+    data = {}
+    for serializer_class in [
+        ParkingSerializer,
+        UserSerializer,
+        PaymentSerializer
+    ]:
+        model = serializer_class.Meta.model
+        db_table = model._meta.db_table
+        queryset = model.objects.all()
+        data[db_table] = serializer_class(
+            queryset, many=True
+        ).data
+
+    return data

application/pay_parking/import_export/apps.py

@@ -0,0 +1,9 @@
+from django import forms
+# import magic
+
+
+class ImportForm(forms.Form):
+    import_file = forms.FileField()
+
+    # def clean_import_file(self):
+    #     import_file = self.cleaned_data.get('import_file')

application/pay_parking/import_export/export_data.py

@@ -0,0 +1,55 @@
+from .serializers import ParkingSerializer, UserSerializer, PaymentSerializer
+
+
+def import_data(data: dict):
+    for serializer_class in [
+        ParkingSerializer,
+        UserSerializer,
+        PaymentSerializer
+    ]:
+        model = serializer_class.Meta.model
+        db_table = model._meta.db_table
+
+        ids = [item['id'] for item in data.get(db_table)]
+        existing_objects = model.objects.filter(id__in=ids)
+        existing_ids = [str(item.id) for item in existing_objects]
+        non_existing_ids = set(ids) - set(existing_ids)
+        non_existing_items = [
+            item for item in data.get(db_table) if item['id'] in non_existing_ids
+        ]
+
+        create_serializer = serializer_class(
+            many=True, data=non_existing_items
+        )
+
+        if create_serializer.is_valid():
+            # model.objects.bulk_create(
+            #     [model(**item) for item in create_serializer.validated_data]
+            # )
+            create_serializer.save()
+        else:
+            raise ValueError(
+                f'Invalid data {db_table}'
+            )
+
+        items_to_update = {
+            item['id']: item for item in data.get(db_table) if item['id'] in existing_ids
+        }
+        for existing_object in existing_objects:
+            update_serializer = serializer_class(
+                existing_object,
+                data=items_to_update[str(existing_object.id)]
+            )
+            if update_serializer.is_valid():
+                update_serializer.save()
+                pass
+            else:
+                raise ValueError(
+                    f'Invalid data {db_table}, pk={existing_object.id}'
+                )
+        #     for field_name, value in update_serializer.validated_data.items():
+        #         setattr(existing_object, field_name, value)
+        
+        # fields = serializer_class().get_fields().keys()
+        # fields = [field for field in fields if field != 'id'] 
+        # model.objects.bulk_update(existing_objects, fields)