(self)
| 110 | return context |
| 111 | |
| 112 | def get_queryset(self): # type: ignore[no-untyped-def] |
| 113 | if getattr(self, "swagger_fake_view", False): |
| 114 | return Environment.objects.none() |
| 115 | |
| 116 | if self.action == "list": |
| 117 | project_id = self.request.query_params.get( |
| 118 | "project" |
| 119 | ) or self.request.data.get("project") |
| 120 | |
| 121 | try: |
| 122 | project = Project.objects.get(id=project_id) |
| 123 | except Project.DoesNotExist: |
| 124 | raise ValidationError("Invalid or missing value for project parameter.") |
| 125 | |
| 126 | return self.request.user.get_permitted_environments( # type: ignore[union-attr] |
| 127 | "VIEW_ENVIRONMENT", project=project, prefetch_metadata=True |
| 128 | ) |
| 129 | |
| 130 | # Permission class handles validation of permissions for other actions |
| 131 | queryset = Environment.objects.all() |
| 132 | |
| 133 | if self.action == "retrieve": |
| 134 | # Since we don't have the environment at this stage, we would need to query the database |
| 135 | # regardless, so it seems worthwhile to just query the database for the latest versions |
| 136 | # and use their existence as a proxy to whether v2 feature versioning is enabled. |
| 137 | latest_versions = EnvironmentFeatureVersion.objects.get_latest_versions_by_environment_api_key( |
| 138 | environment_api_key=self.kwargs["api_key"] |
| 139 | ) |
| 140 | if latest_versions: |
| 141 | # if there are latest versions (and hence v2 feature versioning is enabled), then |
| 142 | # we need to ensure that we're only counting the feature segments for those |
| 143 | # latest versions against the limits. |
| 144 | queryset = queryset.annotate( |
| 145 | total_segment_overrides=Count( |
| 146 | "feature_segments", |
| 147 | filter=Q( |
| 148 | feature_segments__environment_feature_version__in=latest_versions |
| 149 | ), |
| 150 | ) |
| 151 | ) |
| 152 | else: |
| 153 | queryset = queryset.annotate( |
| 154 | total_segment_overrides=Count("feature_segments") |
| 155 | ) |
| 156 | |
| 157 | return queryset |
| 158 | |
| 159 | def perform_create(self, serializer): # type: ignore[no-untyped-def] |
| 160 | environment = serializer.save() |
no test coverage detected