Coverage for backend/django/core/auxiliary/models/UploadSession.py: 100%

1from django.conf import settings 

2from django.db import models 

3 

4from core.managers import AccessControlManager 

5from flowsheetInternals.unitops.models import SimulationObject 

6 

7 

8class UploadSessionPurpose(models.TextChoices): 

9 ML_TRAINING_CSV = "ml_training_csv" 

10 SCENARIO_CSV = "scenario_csv" 

11 PINCH_UTILITY_CSV = "pinch_utility_csv" 

12 

13 

14class UploadSessionStatus(models.TextChoices): 

15 INITIATED = "initiated" 

16 UPLOADING = "uploading" 

17 COMPLETED = "completed" 

18 ABORTED = "aborted" 

19 FAILED = "failed" 

20 EXPIRED = "expired" 

21 

22 

23class UploadSession(models.Model): 

24 created_by = models.ForeignKey( 

25 settings.AUTH_USER_MODEL, 

26 on_delete=models.CASCADE, 

27 related_name="upload_sessions", 

28 ) 

29 flowsheet = models.ForeignKey( 

30 "Flowsheet", 

31 on_delete=models.CASCADE, 

32 related_name="upload_sessions", 

33 ) 

34 scenario = models.ForeignKey( 

35 "Scenario", 

36 on_delete=models.CASCADE, 

37 related_name="upload_sessions", 

38 null=True, 

39 blank=True, 

40 ) 

41 simulationObject = models.ForeignKey( 

42 SimulationObject, 

43 on_delete=models.CASCADE, 

44 related_name="upload_sessions", 

45 null=True, 

46 blank=True, 

47 ) 

48 purpose = models.CharField(max_length=32, choices=UploadSessionPurpose.choices) 

49 bucket = models.CharField(max_length=255) 

50 object_key = models.CharField(max_length=1024, unique=True) 

51 s3_upload_id = models.CharField(max_length=512) 

52 original_filename = models.CharField(max_length=255) 

53 content_type = models.CharField(max_length=255, blank=True, default="") 

54 size_bytes = models.BigIntegerField() 

55 status = models.CharField( 

56 max_length=16, 

57 choices=UploadSessionStatus.choices, 

58 default=UploadSessionStatus.INITIATED, 

59 ) 

60 csv_headers = models.JSONField(default=list, blank=True) 

61 csv_delimiter = models.CharField(max_length=1, blank=True, default="") 

62 csv_warnings = models.JSONField(default=list, blank=True) 

63 csv_inspected_at = models.DateTimeField(null=True, blank=True) 

64 created_at = models.DateTimeField(auto_now_add=True) 

65 completed_at = models.DateTimeField(null=True, blank=True) 

66 expires_at = models.DateTimeField(null=True, blank=True) 

67 

68 objects = AccessControlManager() 

69 

70 class Meta: 

71 indexes = [ 

72 models.Index(fields=["purpose", "status"]), 

73 models.Index(fields=["expires_at"]), 

74 ]