from django.db import models from django.conf import settings class LabInstance(models.Model): class LabInstanceStatus(models.TextChoices): CREATING = "creating" CREATING_FAILED = "creating_failed" RUNNING = "running" TERMINATING = "terminating" TERMINATING_FAILED = "terminating_failed" TERMINATED = "terminated" TIMEOUT = "timeout" # This is a virtual status for user class LabAccessType(models.TextChoices): CREDITS = "credits" WORKSHOP = "workshop" INTERNAL = "internal" user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE) lab_code = models.CharField(max_length=20) cloud_provider = models.CharField(max_length=20, default="aws") region_name = models.CharField(max_length=40) tidb_version = models.CharField(max_length=20, null=True, blank=True) tidb_environment = models.CharField(max_length=40, null=True, blank=True) resources = models.JSONField(default=dict) unique_tag = models.CharField( max_length=100, unique=True, blank=True, null=True, help_text="Can be used as the name of AWS resources.", ) status = models.CharField( max_length=20, choices=LabInstanceStatus.choices, default=LabInstanceStatus.CREATING, ) user_serverless_cluster = models.ForeignKey( "labs.UserServerlessCluster", on_delete=models.CASCADE, null=True, blank=True, ) create_job = models.ForeignKey( "labs.LabJob", on_delete=models.CASCADE, null=True, blank=True, related_name="create_lab_instance", ) terminate_job = models.ForeignKey( "labs.LabJob", on_delete=models.CASCADE, null=True, blank=True, related_name="terminate_lab_instance", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) ready_at = models.DateTimeField(null=True, blank=True) scheduled_terminate_at = models.DateTimeField(null=True, blank=True) start_terminating_at = models.DateTimeField(null=True, blank=True) terminated_at = models.DateTimeField(null=True, blank=True) exited_at = models.DateTimeField( null=True, blank=True, help_text="The time when the user exits the lab" ) access_type = models.CharField( max_length=20, choices=LabAccessType.choices, null=True, blank=True, help_text="The access type of the lab instance", ) # access_detail is a JSON field that contains additional details about the lab access # { # "workshop_name": "Workshop Name", # for WORKSHOP access type # "redemption_id": 1, # for CREDITS access type # } access_detail = models.JSONField( default=dict, help_text="Additional details about the lab access (e.g., workshop name)", ) xip_domain = models.CharField(max_length=32, default="sslip.io") class Meta: db_table = "lab_instances" def get_user_status(self) -> LabInstanceStatus: """ Get the status of the lab instance from the user's perspective. """ if self.status in [ self.LabInstanceStatus.CREATING, self.LabInstanceStatus.CREATING_FAILED, self.LabInstanceStatus.RUNNING, ]: return self.status if self.status in [ self.LabInstanceStatus.TERMINATING, self.LabInstanceStatus.TERMINATING_FAILED, self.LabInstanceStatus.TERMINATED, ]: if not self.exited_at: return self.LabInstanceStatus.TIMEOUT else: return self.LabInstanceStatus.TERMINATED raise ValueError(f"Invalid lab instance status: {self.status}")