Coverage for core/api/serializers_/jobs.py: 78%

1from drf_spectacular.utils import extend_schema_field 

2from rest_framework import serializers 

3 

4from core.choices import * 

5from core.models import Job 

6from netbox.api.exceptions import SerializerNotFound 

7from netbox.api.fields import ChoiceField, ContentTypeField 

8from netbox.api.serializers import BaseModelSerializer 

9from users.api.serializers_.users import UserSerializer 

10from utilities.api import get_serializer_for_model 

11 

12__all__ = ( 

13 'JobSerializer', 

14) 

15 

16 

17class JobSerializer(BaseModelSerializer): 

18 user = UserSerializer( 

19 nested=True, 

20 read_only=True 

21 ) 

22 status = ChoiceField(choices=JobStatusChoices, read_only=True) 

23 object_type = ContentTypeField( 

24 read_only=True 

25 ) 

26 object = serializers.SerializerMethodField( 

27 read_only=True 

28 ) 

29 notifications = ChoiceField(choices=JobNotificationChoices, read_only=True) 

30 

31 class Meta: 

32 model = Job 

33 fields = [ 

34 'id', 'url', 'display_url', 'display', 'object_type', 'object_id', 'object', 'name', 'status', 'created', 

35 'scheduled', 'interval', 'started', 'completed', 'execution_time', 'user', 'data', 'error', 'job_id', 

36 'queue_name', 'notifications', 'log_entries', 

37 ] 

38 brief_fields = ('url', 'created', 'completed', 'user', 'status') 

39 

40 @extend_schema_field(serializers.JSONField(allow_null=True)) 

41 def get_object(self, obj): 

42 """ 

43 Serialize a nested representation of the object. 

44 """ 

45 if obj.object is None: 45 ↛ 47line 45 didn't jump to line 47 because the condition on line 45 was always true

46 return None 

47 try: 

48 serializer = get_serializer_for_model(obj.object) 

49 except SerializerNotFound: 

50 return obj.object_repr 

51 context = {'request': self.context['request']} 

52 return serializer(obj.object, nested=True, context=context).data