service_layer

service/service.py

@@ -0,0 +1,49 @@
+from io import BytesIO
+
+import pandas as pd
+
+from service import models
+
+
+class PointService:
+
+    def update_fact(self, postamat_id: str, fact: int):
+        qs = self.get_point_by_postamat_id(postamat_id)
+        qs.update(**{'fact': fact})
+
+    def update_postamat_id(self, point_id: int, postamat_id: str):
+        qs = self.get_point_by_id(point_id)
+        qs.update(**{'postamat_id': postamat_id})
+
+    @staticmethod
+    def update_status(qs: models.PlacementPoint, new_status: str) -> models.PlacementPoint:
+        qs.update(**{'status': new_status})
+
+    @staticmethod
+    def get_point_by_id(point_id: int):
+        return models.PlacementPoint.objects.filter(pk=point_id)
+
+    @staticmethod
+    def get_point_by_postamat_id(postamat_id: str):
+        return models.PlacementPoint.objects.filter(postamat_id=postamat_id)
+
+    @staticmethod
+    def to_excel(qs: models.PlacementPoint):
+        data = pd.DataFrame(list(qs.values()))
+        data['start_date'] = data['start_date'].dt.tz_localize(None)
+        data['sample_trn'] = data['sample_trn'].astype(int)
+        with BytesIO() as b:
+            with pd.ExcelWriter(b) as writer:
+                data.to_excel(
+                    writer, sheet_name='Placement Points',
+                    index=False,
+                )
+            return b.getvalue()
+
+    @staticmethod
+    def get_first_10_k():
+        if models.PlacementPoint.objects.count() > 10000:
+            qs = models.PlacementPoint.objects.order_by('-prediction_current').all()[10000]
+            return qs.prediction_current
+        else:
+            return models.PlacementPoint.objects.order_by('prediction_current').first().prediction_current

service/views.py

@@ -1,7 +1,5 @@
 import warnings
-from io import BytesIO
 
-import pandas as pd
 from django.db.models import Q
 from django.http import HttpResponse
 from django.http import JsonResponse
@@ -20,6 +18,7 @@ from service import pagination
 from service import serializers
 from service import utils
 from service.permissions import UserPermission
+from service.service import PointService
 from service.utils import load_data
 
 
@@ -141,7 +140,7 @@ class PlacementPointViewSet(ReadOnlyModelViewSet):
         new_status = self.request.GET.get('status')
         if not new_status:
             return Response({'message': 'No status'}, 400)
-        qs.update(**{'status': new_status})
+        PointService.update_status(qs, new_status)
         return Response(
             {'message': 'status updated'},
             status=http_status.HTTP_200_OK,
@@ -165,51 +164,27 @@ class PlacementPointViewSet(ReadOnlyModelViewSet):
         point_id = request.GET.get('id')
         if not point_id or not postamat_id:
             return Response(status=http_status.HTTP_400_BAD_REQUEST)
-        qs = models.PlacementPoint.objects.filter(pk=point_id)
+        qs = PointService.get_point_by_id(point_id)
         if not qs:
             return Response(status=http_status.HTTP_404_NOT_FOUND)
-        qs.update(**{'postamat_id': postamat_id})
+        PointService().update_postamat_id(point_id, postamat_id)
         return Response({'message': 'Postamat id updated'}, status=http_status.HTTP_200_OK)
 
     @action(detail=False, methods=['get'])
     def to_excel(self, request):
         qs = self.get_queryset()
-        data = pd.DataFrame(list(qs.values()))
+        filename = 'placement_points.xlsx'
-        data['start_date'] = data['start_date'].dt.tz_localize(None)
+        res = HttpResponse(
-        data['sample_trn'] = data['sample_trn'].astype(int)
+            PointService.to_excel(qs),
-        with BytesIO() as b:
+            content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
-            with pd.ExcelWriter(b) as writer:
+        )
-                data.to_excel(
+        res['Content-Disposition'] = f'attachment; filename={filename}'
-                    writer, sheet_name='Placement Points',
+        return res
-                    index=False,
-                )
-
-            filename = 'placement_points.xlsx'
-            res = HttpResponse(
-                b.getvalue(),
-                content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
-            )
-            res['Content-Disposition'] = f'attachment; filename={filename}'
-            return res
 
     @action(detail=False, methods=['get'])
     def get_10k(self, request):
-        if models.PlacementPoint.objects.count() > 10000:
+        pred = PointService.get_first_10_k()
-            qs = \
+        return Response({'prediction_current': pred}, status=http_status.HTTP_200_OK)
-                models.PlacementPoint.objects.order_by(
-                    '-prediction_current',
-                ).all()[
-                    10000
-                ]
-            return Response({'prediction_current': qs.prediction_current}, 200)
-        return Response(
-            {
-                'prediction_current': models.PlacementPoint.objects.order_by(
-                    'prediction_current',
-                ).first().prediction_current,
-            },
-            200,
-        )
 
 
 class refresh_placement_points(APIView):