89 lines
4.9 KiB
Python
89 lines
4.9 KiB
Python
from rest_framework import viewsets, permissions, status
|
|
from rest_framework.response import Response
|
|
from rest_framework.decorators import action
|
|
from drf_spectacular.utils import extend_schema, extend_schema_view
|
|
|
|
from ..models import AiModel
|
|
from ..serializers.ai_model_serializer import AiModelSerializer
|
|
from ..repositories.ai_model_repository import AiModelRepository
|
|
from ..services.ai_model_service import AiModelService, ConnectionError
|
|
|
|
from permissions.permission_classes import IsAdminOrManager
|
|
from rest_framework.permissions import IsAuthenticated
|
|
|
|
# Dependency Injection: สร้าง Instance ของ Repository และ Service
|
|
# สำหรับโปรเจกต์ขนาดเล็กสามารถทำแบบนี้ได้
|
|
repo = AiModelRepository()
|
|
service = AiModelService(repository=repo)
|
|
|
|
@extend_schema_view(
|
|
# 1. การดำเนินการ CRUD ปกติ: จัดอยู่ใน Model Registry
|
|
list=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
retrieve=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
create=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
update=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
partial_update=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
destroy=extend_schema(tags=['2. Model Registry & Metadata Management']),
|
|
)
|
|
class AiModelRegistryViewSet(viewsets.ModelViewSet):
|
|
queryset = AiModel.objects.all()
|
|
serializer_class = AiModelSerializer
|
|
permission_classes = [permissions.IsAdminUser]
|
|
|
|
# -----------------------------------------------
|
|
# Override Create/List (เรียกใช้ Service Layer)
|
|
# -----------------------------------------------
|
|
def get_queryset(self):
|
|
# List/Retrieve จะเรียก Service Layer แทนการเรียก ORM โดยตรง
|
|
return service.get_all_models()
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
serializer = self.get_serializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
try:
|
|
new_model = service.create_model(serializer.validated_data)
|
|
return Response(self.get_serializer(new_model).data, status=status.HTTP_201_CREATED)
|
|
except Exception as e:
|
|
return Response({"detail": str(e)}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
def get_permissions(self):
|
|
# การทำงานที่เกี่ยวข้องกับการเขียน/แก้ไข/ลบ ถือเป็นงาน Admin/Manager
|
|
if self.action in ['create', 'update', 'partial_update', 'destroy', 'set_status']:
|
|
# จำกัดสิทธิ์ ต้องเป็น Admin หรือ Manager เท่านั้น
|
|
return [IsAdminOrManager()]
|
|
|
|
# การทำงานที่เกี่ยวข้องกับการอ่านข้อมูล (List, Retrieve, Test Connection)
|
|
# ถือเป็นงานสำหรับผู้ใช้งานที่ล็อกอินแล้ว (Viewer ขึ้นไป)
|
|
elif self.action in ['list', 'retrieve', 'test_connection']:
|
|
return [IsAuthenticated()]
|
|
|
|
# Default สำหรับเมธอดอื่น ๆ หรือ Custom Action ที่ไม่ได้ระบุ
|
|
return [IsAuthenticated()]
|
|
|
|
# -----------------------------------------------
|
|
# Custom Action: ทดสอบการเชื่อมต่อ
|
|
# -----------------------------------------------
|
|
@extend_schema(tags=['3. MLOps Control & Service Orchestration'])
|
|
@action(detail=True, methods=['post'], url_path='test-connection')
|
|
def test_connection(self, request, pk=None):
|
|
try:
|
|
result = service.test_connection(pk=int(pk))
|
|
return Response(result)
|
|
except ValueError as e:
|
|
return Response({"detail": str(e)}, status=status.HTTP_404_NOT_FOUND)
|
|
except ConnectionError as e:
|
|
# Response ด้วย error ที่ชัดเจนจากการเชื่อมต่อ
|
|
return Response({"status": "error", "detail": str(e)}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
@extend_schema(tags=['3. MLOps Control & Service Orchestration'])
|
|
@action(detail=True, methods=['patch'], url_path='set-status')
|
|
def set_status(self, request, pk=None):
|
|
new_status = request.data.get('status')
|
|
if not new_status or new_status not in [choice[0] for choice in AiModel.status_choices]:
|
|
return Response({"detail": "Invalid status provided."}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
updated_model = service.set_status(pk=int(pk), new_status=new_status)
|
|
if updated_model:
|
|
return Response(self.get_serializer(updated_model).data)
|
|
return Response({"detail": "Model not found."}, status=status.HTTP_404_NOT_FOUND) |