Coverage for django_napse/api/bots/views/bot_view.py: 22%

1from django.db.models import QuerySet 

2from rest_framework import status 

3from rest_framework.response import Response 

4from rest_framework_api_key.permissions import HasAPIKey 

5 

6from django_napse.api.bots.serializers.bot_serializers import BotDetailSerializer, BotSerializer 

7from django_napse.api.custom_permissions import HasSpace 

8from django_napse.api.custom_viewset import CustomViewSet 

9from django_napse.core.models import Bot, NapseSpace 

10from django_napse.utils.errors import APIError 

11 

12 

13class BotView(CustomViewSet): 

14 permission_classes = [HasAPIKey, HasSpace] 

15 serializer_class = BotSerializer 

16 

17 def get_queryset(self) -> list[QuerySet[Bot]] | dict[str, QuerySet[Bot]]: 

18 """Return bot queryset. 

19 

20 Can return 

21 Free bots across all available spaces 

22 Bots with connections without space containerization 

23 Bots with connections with a specific space containerization 

24 Bots with connections with space containerization 

25 

26 Raises: 

27 ValueError: Not space_containers mode is only available for master key. 

28 ValueError: Space not found. 

29 """ 

30 api_key = self.get_api_key(self.request) 

31 spaces = NapseSpace.objects.all() if api_key.is_master_key else [permission.space for permission in api_key.permissions.all()] 

32 

33 # Free bots across all available spaces 

34 if self.request.query_params.get("free", False): 

35 # Exchange account containerization 

36 request_space = self.get_space(self.request) 

37 if request_space is None: 

38 raise APIError.MissingSpace() 

39 spaces = [space for space in spaces if space.exchange_account == request_space.exchange_account] 

40 # Cross space free bots 

41 return [bot for bot in Bot.objects.filter(strategy__config__space__in=spaces) if bot.is_free] 

42 

43 # Bots with connections without space containerization 

44 if not self.request.query_params.get("space_containers", True): 

45 if not api_key.is_master_key: 

46 error_msg: str = "Not space_containers mode is only available for master key." 

47 raise ValueError(error_msg) 

48 return Bot.objects.exclude(connections__isnull=True) 

49 

50 # Filter by specific space 

51 space_uuid = self.request.query_params.get("space", None) 

52 if space_uuid is not None: 

53 for space in spaces: 

54 if space.uuid == space_uuid: 

55 spaces = [space] 

56 break 

57 else: 

58 error_msg: str = "Space not found." 

59 raise ValueError(error_msg) 

60 

61 # Space container mode 

62 bots_per_space: dict[str, QuerySet[Bot]] = {} 

63 for space in spaces: 

64 bots_per_space[space] = space.bots.exclude(connections__isnull=True) 

65 return bots_per_space 

66 

67 def get_serializer_class(self, *args, **kwargs): 

68 actions: dict = { 

69 "list": BotSerializer, 

70 "retrieve": BotDetailSerializer, 

71 } 

72 result = actions.get(self.action) 

73 return result if result else super().get_serializer_class() 

74 

75 def get_permissions(self): 

76 match self.action: 

77 case "list" | "create": 

78 return [HasAPIKey()] 

79 case _: 

80 return super().get_permissions() 

81 

82 def get_object(self): 

83 uuid = self.kwargs.get("pk", None) 

84 if uuid is None: 

85 return super().get_object() 

86 return Bot.objects.get(uuid=uuid) 

87 

88 def _get_boolean_query_param(self, param: str) -> bool | None: 

89 """Return None if a boolean cannot be found.""" 

90 if isinstance(param, bool): 

91 return param 

92 

93 if not isinstance(param, str): 

94 return None 

95 

96 match param.lower(): 

97 case "true" | "1": 

98 return True 

99 case "false" | "0": 

100 return False 

101 case _: 

102 return None 

103 

104 def list(self, request): 

105 """Return a list of bots. 

106 

107 Warning: space_containers can lead to undesirable behaviour. 

108 """ 

109 try: 

110 queryset = self.get_queryset() 

111 except ValueError as error: 

112 return Response({"detail": str(error)}, status=status.HTTP_400_BAD_REQUEST) 

113 

114 if isinstance(queryset, list): 

115 serializer = self.get_serializer(queryset, many=True) 

116 return Response(serializer.data, status=status.HTTP_200_OK) 

117 

118 if isinstance(queryset, dict): 

119 serialized_bots: QuerySet[Bot] = [] 

120 for space, query in queryset.items(): 

121 serializer = self.get_serializer(query, many=True, space=space) 

122 if serializer.data != []: 

123 serialized_bots += serializer.data 

124 return Response(serialized_bots, status=status.HTTP_200_OK) 

125 

126 return Response(status=status.HTTP_400_BAD_REQUEST) 

127 

128 def retrieve(self, request, pk=None): 

129 instance = self.get_object() 

130 space = self.get_space(request) 

131 serializer = self.get_serializer(instance, space=space) 

132 return Response(serializer.data, status=status.HTTP_200_OK)