Coverage for core/api/serializers_/tasks.py: 98%

1from rest_framework import serializers 

2from rest_framework.reverse import reverse 

3 

4__all__ = ( 

5 'BackgroundQueueSerializer', 

6 'BackgroundTaskSerializer', 

7 'BackgroundWorkerSerializer', 

8) 

9 

10 

11class BackgroundTaskSerializer(serializers.Serializer): 

12 id = serializers.CharField() 

13 url = serializers.HyperlinkedIdentityField( 

14 view_name='core-api:rqtask-detail', 

15 lookup_field='id', 

16 lookup_url_kwarg='id' 

17 ) 

18 description = serializers.CharField() 

19 origin = serializers.CharField() 

20 func_name = serializers.CharField() 

21 args = serializers.SerializerMethodField() 

22 kwargs = serializers.SerializerMethodField() 

23 result = serializers.CharField() 

24 timeout = serializers.IntegerField() 

25 result_ttl = serializers.IntegerField() 

26 created_at = serializers.DateTimeField() 

27 enqueued_at = serializers.DateTimeField() 

28 started_at = serializers.DateTimeField() 

29 ended_at = serializers.DateTimeField() 

30 worker_name = serializers.CharField() 

31 position = serializers.SerializerMethodField() 

32 status = serializers.SerializerMethodField() 

33 meta = serializers.DictField() 

34 last_heartbeat = serializers.CharField() 

35 

36 is_finished = serializers.BooleanField() 

37 is_queued = serializers.BooleanField() 

38 is_failed = serializers.BooleanField() 

39 is_started = serializers.BooleanField() 

40 is_deferred = serializers.BooleanField() 

41 is_canceled = serializers.BooleanField() 

42 is_scheduled = serializers.BooleanField() 

43 is_stopped = serializers.BooleanField() 

44 

45 def get_args(self, obj) -> list: 

46 return [ 

47 str(arg) for arg in obj.args 

48 ] 

49 

50 def get_kwargs(self, obj) -> dict: 

51 return { 

52 key: str(value) for key, value in obj.kwargs.items() 

53 } 

54 

55 def get_position(self, obj) -> int: 

56 return obj.get_position() 

57 

58 def get_status(self, obj) -> str: 

59 return obj.get_status() 

60 

61 

62class BackgroundQueueSerializer(serializers.Serializer): 

63 name = serializers.CharField() 

64 url = serializers.SerializerMethodField() 

65 jobs = serializers.IntegerField() 

66 oldest_job_timestamp = serializers.CharField() 

67 index = serializers.IntegerField() 

68 scheduler_pid = serializers.CharField() 

69 workers = serializers.IntegerField() 

70 finished_jobs = serializers.IntegerField() 

71 started_jobs = serializers.IntegerField() 

72 deferred_jobs = serializers.IntegerField() 

73 failed_jobs = serializers.IntegerField() 

74 scheduled_jobs = serializers.IntegerField() 

75 

76 def get_url(self, obj): 

77 return reverse('core-api:rqqueue-detail', args=[obj['name']], request=self.context.get("request")) 

78 

79 

80class BackgroundWorkerSerializer(serializers.Serializer): 

81 name = serializers.CharField() 

82 url = serializers.HyperlinkedIdentityField( 

83 view_name='core-api:rqworker-detail', 

84 lookup_field='name' 

85 ) 

86 state = serializers.SerializerMethodField() 

87 birth_date = serializers.DateTimeField() 

88 queue_names = serializers.ListField( 

89 child=serializers.CharField() 

90 ) 

91 pid = serializers.CharField() 

92 successful_job_count = serializers.IntegerField() 

93 failed_job_count = serializers.IntegerField() 

94 total_working_time = serializers.IntegerField() 

95 

96 def get_state(self, obj): 

97 return obj.get_state()