""" NHN Cloud Async Task Models 비동기 작업 상태 추적 """ import uuid from django.db import models class AsyncTask(models.Model): """비동기 작업 상태 추적 모델""" class Status(models.TextChoices): PENDING = "pending", "대기중" RUNNING = "running", "실행중" SUCCESS = "success", "성공" FAILED = "failed", "실패" class TaskType(models.TextChoices): # Compute INSTANCE_CREATE = "instance_create", "인스턴스 생성" INSTANCE_DELETE = "instance_delete", "인스턴스 삭제" INSTANCE_ACTION = "instance_action", "인스턴스 액션" # VPC/Network VPC_CREATE = "vpc_create", "VPC 생성" VPC_DELETE = "vpc_delete", "VPC 삭제" SUBNET_CREATE = "subnet_create", "서브넷 생성" SUBNET_DELETE = "subnet_delete", "서브넷 삭제" # NKS NKS_CREATE = "nks_create", "NKS 클러스터 생성" NKS_DELETE = "nks_delete", "NKS 클러스터 삭제" # Storage STORAGE_CREATE = "storage_create", "스토리지 컨테이너 생성" STORAGE_DELETE = "storage_delete", "스토리지 컨테이너 삭제" # Load Balancer LB_CREATE = "lb_create", "로드밸런서 생성" LB_DELETE = "lb_delete", "로드밸런서 삭제" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) task_type = models.CharField(max_length=50, choices=TaskType.choices) status = models.CharField( max_length=20, choices=Status.choices, default=Status.PENDING ) # 요청 정보 request_data = models.JSONField(default=dict, help_text="요청 데이터") # 결과 정보 result_data = models.JSONField(default=dict, help_text="결과 데이터") error_message = models.TextField(blank=True, help_text="에러 메시지") # NHN Cloud 리소스 정보 resource_id = models.CharField(max_length=255, blank=True, help_text="생성된 리소스 ID") resource_name = models.CharField(max_length=255, blank=True, help_text="리소스 이름") # 메타 정보 created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) completed_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ["-created_at"] verbose_name = "비동기 작업" verbose_name_plural = "비동기 작업들" def __str__(self): return f"{self.task_type} - {self.status} ({self.id})" def mark_running(self): """작업 시작""" self.status = self.Status.RUNNING self.save(update_fields=["status", "updated_at"]) def mark_success(self, result_data=None, resource_id=None): """작업 성공""" from django.utils import timezone self.status = self.Status.SUCCESS if result_data: self.result_data = result_data if resource_id: self.resource_id = resource_id self.completed_at = timezone.now() self.save() def mark_failed(self, error_message): """작업 실패""" from django.utils import timezone self.status = self.Status.FAILED self.error_message = error_message self.completed_at = timezone.now() self.save()