-
Notifications
You must be signed in to change notification settings - Fork 34
/
views.py
198 lines (156 loc) · 6.34 KB
/
views.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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
""".. Ignore pydocstyle D400.
=====
Views
=====
"""
from elasticsearch_dsl.query import Q
from rest_framework import viewsets, mixins, permissions
from rest_framework_filters.backends import DjangoFilterBackend
from resolwe.elastic.viewsets import ElasticSearchBaseViewSet
from .models import Feature, Mapping
from .serializers import FeatureSerializer, MappingSerializer
from .filters import MappingFilter
from .elastic_indexes import FeatureSearchDocument, MappingSearchDocument
class FeatureSearchViewSet(ElasticSearchBaseViewSet):
"""
Endpoint used for feature search.
Request:
- query
- source
Response:
- a list of matching features
"""
document_class = FeatureSearchDocument
serializer_class = FeatureSerializer
filtering_fields = ('name', 'source', 'species', 'feature_id')
ordering_fields = ('name',)
ordering = 'name'
def custom_filter_feature_id(self, value, search):
"""Support exact feature_id queries."""
if not isinstance(value, list):
value = [value]
return search.filter('terms', feature_id=value)
def custom_filter(self, search):
"""Support general query using the 'query' attribute."""
query = self.get_query_param('query', None)
if query:
if not isinstance(query, list):
query = [query]
search = search.filter(
'bool',
should=[
Q('terms', feature_id=query),
Q('terms', name=query),
Q('terms', aliases=query),
]
)
return search
def filter_permissions(self, search):
"""Feature objects have no permissions."""
return search
class FeatureAutocompleteViewSet(ElasticSearchBaseViewSet):
"""Endpoint used for feature autocompletion."""
document_class = FeatureSearchDocument
serializer_class = FeatureSerializer
filtering_fields = ('source', 'species')
def custom_filter(self, search):
"""Support autocomplete query using the 'query' attribute."""
return search.query(
'bool',
should=[
Q('match', autocomplete=self.get_query_param('query', '')),
# Boost exact name and feature_id matches. Use the 'lower' subfield in order
# to compare against lowercased terms.
Q('match', **{'feature_id.lower': {'query': self.get_query_param('query', ''), 'boost': 2}}),
Q('match', **{'name.lower': {'query': self.get_query_param('query', ''), 'boost': 2}}),
]
)
def filter_permissions(self, search):
"""Feature objects have no permissions."""
return search
class FeatureViewSet(mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.CreateModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet):
"""API view for :class:`Feature` objects."""
serializer_class = FeatureSerializer
permission_classes = [permissions.IsAdminUser]
filter_backends = [DjangoFilterBackend]
queryset = Feature.objects.all()
def create(self, request):
"""Instead of failing, update existing features with a custom create."""
try:
feature = Feature.objects.get(
source=request.data['source'],
feature_id=request.data['feature_id']
)
self.kwargs[self.lookup_field] = feature.pk
return super(FeatureViewSet, self).update(request) # pylint: disable=no-member
except (Feature.DoesNotExist, KeyError): # pylint: disable=no-member
return super(FeatureViewSet, self).create(request) # pylint: disable=no-member
class MappingSearchViewSet(ElasticSearchBaseViewSet):
"""
Endpoint used for mapping search.
Request:
- source_id
- source_db
- source_species
- target_id
- target_db
- target_species
- relation_type
Response:
- a list of matching mappings
"""
document_class = MappingSearchDocument
serializer_class = MappingSerializer
filtering_fields = ('source_db', 'source_species', 'target_db', 'target_species', 'relation_type')
ordering_fields = ('source_id',)
ordering = 'source_id'
def custom_filter(self, search):
"""Support correct searching by ``source_id`` and ``target_id``."""
for field in ('source_id', 'target_id'):
query = self.get_query_param(field, None)
if not query:
continue
if not isinstance(query, list):
query = [query]
search = search.filter(
'bool',
should=[
Q('terms', **{field: query}),
]
)
return search
def filter_permissions(self, search):
"""Filter permissions since Mapping objects have no permissions."""
return search
class MappingViewSet(mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.CreateModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet):
"""API view for :class:`Mapping` objects."""
serializer_class = MappingSerializer
permission_classes = [permissions.IsAdminUser]
filter_backends = [DjangoFilterBackend]
filter_class = MappingFilter
queryset = Mapping.objects.all()
def create(self, request):
"""Instead of failing, update existing mappings using a custom create."""
try:
mapping = Mapping.objects.get(
source_db=request.data['source_db'],
source_id=request.data['source_id'],
source_species=request.data['source_species'],
target_db=request.data['target_db'],
target_id=request.data['target_id'],
target_species=request.data['target_species'],
)
self.kwargs[self.lookup_field] = mapping.pk
return super(MappingViewSet, self).update(request) # pylint: disable=no-member
except (Mapping.DoesNotExist, KeyError): # pylint: disable=no-member
return super(MappingViewSet, self).create(request) # pylint: disable=no-member