-
-
Notifications
You must be signed in to change notification settings - Fork 4.2k
/
organization_searches.py
120 lines (107 loc) · 4.57 KB
/
organization_searches.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from django.db.models import Q
from rest_framework.request import Request
from rest_framework.response import Response
from sentry import analytics
from sentry.api.api_owners import ApiOwner
from sentry.api.api_publish_status import ApiPublishStatus
from sentry.api.base import region_silo_endpoint
from sentry.api.bases.organization import OrganizationEndpoint, OrganizationSearchPermission
from sentry.api.serializers import serialize
from sentry.api.serializers.rest_framework.savedsearch import (
BaseOrganizationSearchSerializer,
OrganizationSearchAdminSerializer,
OrganizationSearchMemberSerializer,
)
from sentry.models.organization import Organization
from sentry.models.savedsearch import SavedSearch, Visibility
from sentry.models.search_common import SearchType
@region_silo_endpoint
class OrganizationSearchesEndpoint(OrganizationEndpoint):
publish_status = {
"GET": ApiPublishStatus.UNKNOWN,
"POST": ApiPublishStatus.UNKNOWN,
}
owner = ApiOwner.ISSUES
permission_classes = (OrganizationSearchPermission,)
def get(self, request: Request, organization: Organization) -> Response:
"""
List an Organization's saved searches
`````````````````````````````````````
Retrieve a list of saved searches for a given Organization. For custom
saved searches, return them for all projects even if we have duplicates.
For default searches, just return one of each search
:auth: required
"""
try:
search_type = SearchType(int(request.GET.get("type", 0)))
except ValueError as e:
return Response({"detail": "Invalid input for `type`. Error: %s" % str(e)}, status=400)
query = (
SavedSearch.objects
# Do not include pinned or personal searches from other users in
# the same organization. DOES include the requesting users pinned
# search
.exclude(
~Q(owner_id=request.user.id),
visibility__in=(Visibility.OWNER, Visibility.OWNER_PINNED),
)
.filter(
Q(organization=organization) | Q(is_global=True),
type=search_type,
)
.extra(order_by=["name"])
)
return Response(serialize(list(query), request.user))
def post(self, request: Request, organization: Organization) -> Response:
serializer: BaseOrganizationSearchSerializer
if request.access.has_scope("org:write"):
serializer = OrganizationSearchAdminSerializer(data=request.data)
else:
serializer = OrganizationSearchMemberSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=400)
result = serializer.validated_data
if result["visibility"] == Visibility.ORGANIZATION:
# Check for conflicts against existing org searches
if SavedSearch.objects.filter(
is_global=False,
organization=organization,
type=SearchType.ISSUE.value,
visibility=Visibility.ORGANIZATION,
query=result["query"],
).exists():
return Response(
{"detail": f"An organization search for '{result['query']}' already exists"},
status=400,
)
elif result["visibility"] == Visibility.OWNER:
# Check for conflicts against the user's searches
if SavedSearch.objects.filter(
is_global=False,
organization=organization,
type=SearchType.ISSUE.value,
visibility=Visibility.OWNER,
owner_id=request.user.id,
query=result["query"],
).exists():
return Response(
{"detail": f"A search for '{result['query']}' already exists"},
status=400,
)
saved_search = SavedSearch.objects.create(
organization=organization,
owner_id=request.user.id,
type=result["type"],
name=result["name"],
query=result["query"],
sort=result["sort"],
visibility=result["visibility"],
)
assert saved_search.type is not None
analytics.record(
"organization_saved_search.created",
search_type=SearchType(saved_search.type).name,
org_id=organization.id,
query=saved_search.query,
)
return Response(serialize(saved_search, request.user))