109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
from apps.warehouse.web.api.v1 import serializers as warehouse_serializers
|
|
from apps.core.mixins.search_mixin import DynamicSearchMixin
|
|
from apps.warehouse import models as warehouse_models
|
|
from common.helpers import get_organization_by_user
|
|
from common.generics import base64_to_image_file
|
|
from common.liara_tools import upload_to_liara
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from rest_framework import viewsets, filters
|
|
from django.db import transaction
|
|
from rest_framework import status
|
|
import typing
|
|
|
|
|
|
class InventoryEntryViewSet(viewsets.ModelViewSet, DynamicSearchMixin):
|
|
queryset = warehouse_models.InventoryEntry.objects.all()
|
|
serializer_class = warehouse_serializers.InventoryEntrySerializer
|
|
filter_backends = [filters.SearchFilter]
|
|
search_fields = [
|
|
"distribution__distribution_id",
|
|
"organization__name",
|
|
"weight",
|
|
"balance",
|
|
"lading_number",
|
|
"is_confirmed",
|
|
]
|
|
date_field = "create_date"
|
|
|
|
def upload_confirmation_document(self, request, inventory: int) -> typing.Any:
|
|
""" upload document for inventory entry confirmation """
|
|
|
|
inventory = self.queryset.get(id=inventory)
|
|
|
|
# convert base64 to document file
|
|
document = base64_to_image_file(
|
|
request.data['document'],
|
|
filename=f'inventory_entry_document_{inventory.id}'
|
|
)
|
|
file, file_format = document[0], document[1]
|
|
|
|
# upload document to liara
|
|
document_url = upload_to_liara(
|
|
file,
|
|
f'inventory_entry_document_{inventory.id}.{file_format}'
|
|
)
|
|
inventory.document = document_url
|
|
inventory.is_confirmed = True
|
|
inventory.save()
|
|
return Response(status=status.HTTP_200_OK)
|
|
|
|
@transaction.atomic
|
|
def create(self, request, *args, **kwargs):
|
|
""" custom create of inventory entry """
|
|
|
|
# create inventory entry
|
|
request.data.update({
|
|
'organization': (get_organization_by_user(request.user)).id
|
|
})
|
|
serializer = self.serializer_class(data=request.data)
|
|
if serializer.is_valid():
|
|
inventory_entry = serializer.save()
|
|
|
|
# upload document for confirmation entry
|
|
if 'document' in request.data.keys():
|
|
self.upload_confirmation_document(request, inventory=inventory_entry.id)
|
|
|
|
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
|
return Response(serializer.errors, status=status.HTTP_403_FORBIDDEN)
|
|
|
|
@action(
|
|
methods=['post'],
|
|
detail=True,
|
|
url_path='confirm_entry',
|
|
url_name='confirm_entry',
|
|
name='confirm_entry'
|
|
)
|
|
@transaction.atomic
|
|
def confirm_inventory_entry(self, request, pk=None):
|
|
""" confirm inventory entry """
|
|
|
|
self.upload_confirmation_document(request, inventory=pk)
|
|
return Response(status=status.HTTP_200_OK)
|
|
|
|
@action(
|
|
methods=['get'],
|
|
detail=False,
|
|
url_path='my_entries',
|
|
url_name='my_entries',
|
|
name='my_entries'
|
|
)
|
|
def my_inventory_entries(self, request):
|
|
""" list of my inventory entries """
|
|
|
|
queryset = self.filter_queryset(self.queryset) # return by search param or all objects
|
|
entries = queryset.filter(organization=get_organization_by_user(request.user))
|
|
|
|
# paginate & response
|
|
page = self.paginate_queryset(entries)
|
|
if page is not None:
|
|
serializer = self.get_serializer(page, many=True)
|
|
return self.get_paginated_response(serializer.data)
|
|
|
|
|
|
class InventoryQuotaSaleTransactionViewSet(viewsets.ModelViewSet):
|
|
queryset = warehouse_models.InventoryQuotaSaleTransaction.objects.all()
|
|
serializer_class = warehouse_serializers.InventoryQuotaSaleTransactionSerializer
|
|
filter_backends = [filters.SearchFilter]
|
|
search_fields = ['']
|