Coverage for django_napse/api/fleets/views/fleet_view.py: 26%

1from rest_framework import status 

2from rest_framework.decorators import action 

3from rest_framework.response import Response 

4from rest_framework_api_key.permissions import HasAPIKey 

5 

6from django_napse.api.custom_permissions import HasSpace 

7from django_napse.api.custom_viewset import CustomViewSet 

8from django_napse.api.fleets.serializers import FleetDetailSerializer, FleetMoneyFlowSerializer, FleetSerializer 

9from django_napse.core.models import Fleet, NapseSpace 

10 

11 

12class FleetView(CustomViewSet): 

13 """View of a fleet. 

14 

15 Query parameters: 

16 space: uuid of the space to filter on. 

17 space_containers (bool): If True, list endpoint returns fleets for each space (default = True). 

18 """ 

19 

20 permission_classes = [HasAPIKey, HasSpace] 

21 serializer_class = FleetSerializer 

22 

23 def get_queryset(self): 

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

25 if space_uuid is None: 

26 return Fleet.objects.all() 

27 self.get_space = NapseSpace.objects.get(uuid=space_uuid) 

28 return self.get_space.fleets 

29 

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

31 actions: dict = { 

32 "list": FleetSerializer, 

33 "retrieve": FleetDetailSerializer, 

34 "create": FleetSerializer, 

35 "invest": FleetMoneyFlowSerializer, 

36 "withdraw": FleetMoneyFlowSerializer, 

37 } 

38 result = actions.get(self.action) 

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

40 

41 def get_permissions(self): 

42 match self.action: 

43 case "list" | "create": 

44 return [HasAPIKey()] 

45 case _: 

46 return super().get_permissions() 

47 

48 def get_object(self): 

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

50 if uuid is None: 

51 return super().get_object() 

52 return Fleet.objects.get(uuid=uuid) 

53 

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

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

56 if isinstance(param, bool): 

57 return param 

58 

59 if not isinstance(param, str): 

60 return None 

61 

62 match param.lower(): 

63 case "true" | "1": 

64 return True 

65 case "false" | "0": 

66 return False 

67 case _: 

68 return None 

69 

70 def list(self, request): 

71 space_containers = self._get_boolean_query_param(request.query_params.get("space_containers", True)) 

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

73 api_key = self.get_api_key(request) 

74 

75 if not space_containers and api_key.is_master_key: 

76 # Not space_containers mode is only available for master key 

77 serializer = self.get_serializer(self.get_queryset(), many=True) 

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

79 

80 # Get spaces from API key 

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

82 # Filter by specific space 

83 if space_uuid is not None: 

84 space = NapseSpace.objects.get(uuid=space_uuid) 

85 if space not in spaces: 

86 return Response(status=status.HTTP_403_FORBIDDEN) 

87 spaces = [space] 

88 

89 # Fleet list 

90 fleets = [] 

91 for space in spaces: 

92 serializer = self.get_serializer(space.fleets, many=True, space=space) 

93 if serializer.data != []: 

94 fleets += serializer.data 

95 return Response(fleets, status=status.HTTP_200_OK) 

96 

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

98 instance = self.get_object() 

99 space = self.get_space(request) 

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

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

102 

103 def create(self, request, *args, **kwargs): 

104 serializer = self.get_serializer(data=request.data) 

105 serializer.is_valid(raise_exception=True) 

106 fleet = serializer.save() 

107 space = serializer.space 

108 fleet.invest(space, 0, "USDT") 

109 return Response(serializer.data, status=status.HTTP_201_CREATED) 

110 

111 def delete(self): 

112 return Response(status=status.HTTP_501_NOT_IMPLEMENTED) 

113 

114 @action(detail=True, methods=["post"]) 

115 def invest(self, request, pk=None): 

116 fleet = self.get_object() 

117 space = self.get_space(request) 

118 

119 if not space.testing: 

120 error_msg: str = "Investing in real is not allowed yet." 

121 return Response(error_msg, status=status.HTTP_403_FORBIDDEN) 

122 

123 serializer = self.get_serializer( 

124 data=request.data, 

125 instance=fleet, 

126 space=space, 

127 side="INVEST", 

128 ) 

129 serializer.is_valid(raise_exception=True) 

130 serializer.save() 

131 

132 return Response(status=status.HTTP_200_OK) 

133 

134 @action(detail=True, methods=["post"]) 

135 def withdraw(self, request, pk=None): 

136 return Response(status=status.HTTP_501_NOT_IMPLEMENTED)