from django.shortcuts import render # Create your views here. from django.http import JsonResponse from django.views import View from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt from typing import Union, List from rest_framework.parsers import JSONParser from api.serializers import ItemSerializer from main.main import mainDet from api.models import ImageURL import json import asyncio import torch async def process_item(item): try: result = await mainDet(item) result = json.loads(result) return result finally: torch.cuda.empty_cache() pass async def process_items(items): if type(items)==list: coroutines = [process_item(item) for item in items] results = await asyncio.gather(*coroutines) print("multi : ",results) else: results = await process_item(items) print("single : ", results) return results @method_decorator(csrf_exempt, name="dispatch") class status(View): async def get(self, request): return JsonResponse({"Status":"AI Server is Running"}, status=200) @method_decorator(csrf_exempt, name="dispatch") class body(View): async def post(self, request): try: data = json.loads(request.body.decode("utf-8")) # print(data) except json.JSONDecodeError: return JsonResponse({"ERROR":"Invalid Body"}, status=400) serializer = ItemSerializer(data=data) if serializer.is_valid(): # print(serializer) result = await process_items(serializer.validated_data["url"]) return JsonResponse(result, safe=False, status=200) else: return JsonResponse(serializer.errors,status=400)