ProviderView.get_concept() - Code Metrics - Inspection of "Refactor flattening" - OnroerendErfgoed/pyramid_skosprovider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — refactor_query ( 910c8b...1aad74 )

by Koen

created 2016-05-10 21:32 UTC

ProviderView.get_concept() A

↳ Parent: ProviderView

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	0	Features	0

Metric	Value
cc	2
c	5
b	0
f	0
dl	0
loc	9
rs	9.6666

# -*- coding: utf8 -*-
'''
This module contains the pyramid views that expose services.
'''

from __future__ import unicode_literals

import itertools

from pyramid.view import view_config, view_defaults

from pyramid.compat import ascii_native_

from pyramid.httpexceptions import (
    HTTPNotFound,
    HTTPBadRequest
)

from pyramid_skosprovider.utils import (
    parse_range_header
)

import logging
log = logging.getLogger(__name__)


class RestView(object):

    def __init__(self, request):
        self.request = request
        self.skos_registry = self.request.skos_registry


@view_defaults(renderer='skosjson', accept='application/json')
class ProviderView(RestView):
    '''
    A set of views that expose information from a certain provider.
    '''

    @view_config(route_name='skosprovider.uri', request_method='GET')
    @view_config(route_name='skosprovider.uri.deprecated', request_method='GET')
    def get_uri(self):
        uri = self.request.params.get('uri')
        if not uri:
            if 'uri' in self.request.matchdict:
                uri = self.request.matchdict['uri']
            else:
                return HTTPBadRequest()
        provider = self.skos_registry.get_provider(uri)
        if provider:
            return {
                'type': 'concept_scheme',
                'uri': provider.concept_scheme.uri,
                'id': provider.get_vocabulary_id()
            }
        c = self.skos_registry.get_by_uri(uri)
        if c:
            return {
                'type': c.type,
                'uri': c.uri,
                'id': c.id,
                'concept_scheme': {
                    'uri': c.concept_scheme.uri,
                    'id': self.skos_registry.get_provider(c.concept_scheme.uri).get_vocabulary_id()
                }
            }
        if not c:
            return HTTPNotFound()

    @view_config(route_name='skosprovider.conceptschemes', request_method='GET')
    def get_conceptschemes(self):
        language = self.request.params.get('language', self.request.locale_name)
        return [
            {
                'id': p.get_vocabulary_id(),
                'uri': p.concept_scheme.uri,
                'label': p.concept_scheme.label(language).label if p.concept_scheme.label(language) else None,
                'subject': p.metadata['subject'] if p.metadata['subject'] else []
            } for p in self.skos_registry.get_providers()
        ]

    @view_config(route_name='skosprovider.conceptscheme', request_method='GET')
    def get_conceptscheme(self):
        scheme_id = self.request.matchdict['scheme_id']
        provider = self.skos_registry.get_provider(scheme_id)
        if not provider:
            return HTTPNotFound()
        language = self.request.params.get('language', self.request.locale_name)
        return {
            'id': provider.get_vocabulary_id(),
            'uri': provider.concept_scheme.uri,
            'label': provider.concept_scheme.label(language).label if provider.concept_scheme.label(language) else None,
            'subject': provider.metadata['subject'] if provider.metadata['subject'] else [],
            'labels': provider.concept_scheme.labels,
            'notes': provider.concept_scheme.notes,
            'sources': provider.concept_scheme.sources,
            'languages': provider.concept_scheme.languages
        }

    @view_config(route_name='skosprovider.conceptscheme.tc', request_method='GET')
    def get_conceptscheme_top_concepts(self):
        scheme_id = self.request.matchdict['scheme_id']
        provider = self.skos_registry.get_provider(scheme_id)
        if not provider:
            return HTTPNotFound()
        language = self.request.params.get('language', self.request.locale_name)
        return provider.get_top_concepts(language=language)

    @view_config(route_name='skosprovider.conceptscheme.display_top', request_method='GET')
    def get_conceptscheme_display_top(self):
        scheme_id = self.request.matchdict['scheme_id']
        provider = self.skos_registry.get_provider(scheme_id)
        if not provider:
            return HTTPNotFound()
        language = self.request.params.get('language', self.request.locale_name)
        return provider.get_top_display(language=language)

    def _build_providers(self, request):
        '''
        :param pyramid.request.Request request:
        :rtype: :class:`dict`
        '''
        # determine targets
        providers = {}
        ids = request.params.get('providers.ids', None)
        if ids:
            ids = ids.split(',')
            providers['ids'] = ids
        subject = self.request.params.get('providers.subject', None)
        if subject:
            providers['subject'] = subject
        return providers

    @view_config(route_name='skosprovider.cs', request_method='GET')
    def get_concepts(self):
        query = {}
        mode = self.request.params.get('mode', 'default')
        label = self.request.params.get('label', None)
        postprocess = False
        language = self.request.params.get('language', self.request.locale_name)
        if mode == 'dijitFilteringSelect' and label == '':
            concepts = []
        else:
            if label not in [None, '*', '']:
                if mode == 'dijitFilteringSelect' and '*' in label:
                    postprocess = True
                    query['label'] = label.replace('*', '')
                else:
                    query['label'] = label
            type = self.request.params.get('type', None)
            if type in ['concept', 'collection']:
                query['type'] = type

            concepts = self.skos_registry.find(
                query,
                providers=self._build_providers(self.request),
                language=language
            )
            # Flatten it all
            concepts = list(itertools.chain.from_iterable([c['concepts'] for c in concepts]))

        if postprocess:
            concepts = self._postprocess_wildcards(concepts, label)

        concepts = self._sort_concepts(concepts)

        return self._page_results(concepts)

    @view_config(route_name='skosprovider.conceptscheme.cs', request_method='GET')
    def get_conceptscheme_concepts(self):
        scheme_id = self.request.matchdict['scheme_id']
        provider = self.skos_registry.get_provider(scheme_id)
        if not provider:
            return HTTPNotFound()
        query = {}
        mode = self.request.params.get('mode', 'default')
        label = self.request.params.get('label', None)
        postprocess = False
        if mode == 'dijitFilteringSelect' and label == '':
            concepts = []
        else:
            if label not in [None, '*', '']:
                if mode == 'dijitFilteringSelect' and '*' in label:
                    postprocess = True
                    query['label'] = label.replace('*', '')
                else:
                    query['label'] = label
            type = self.request.params.get('type', None)
            if type in ['concept', 'collection']:
                query['type'] = type
            coll = self.request.params.get('collection', None)
            if coll is not None:
                query['collection'] = {'id': coll, 'depth': 'all'}
            language = self.request.params.get('language', self.request.locale_name)
            concepts = provider.find(query, language=language)

        if postprocess:
            concepts = self._postprocess_wildcards(concepts, label)

        concepts = self._sort_concepts(concepts)

        return self._page_results(concepts)

    def _postprocess_wildcards(self, concepts, label):
        # We need to refine results further
        if label.startswith('*') and label.endswith('*'):
            concepts = [c for c in concepts if label[1:-1] in c['label']]
        elif label.endswith('*'):
            concepts = [c for c in concepts if c['label'].startswith(label[0:-1])]
        elif label.startswith('*'):
            concepts = [c for c in concepts if c['label'].endswith(label[1:])]
        return concepts

    def _sort_concepts(self, concepts):
        sort = self.request.params.get('sort', None)
        # Result sorting
        if sort:
            sort_desc = (sort[0:1] == '-')
            sort = sort[1:] if sort[0:1] in ['-', '+'] else sort
            sort = sort.strip()  # dojo store does not encode '+'
            if (len(concepts) > 0) and (sort in concepts[0]):
                if sort == 'label':
                    concepts.sort(
                        key=lambda concept: concept[sort].lower(),
                        reverse=sort_desc
                    )
                else:
                    concepts.sort(
                        key=lambda concept: concept[sort],
                        reverse=sort_desc
                    )
        return concepts

    def _page_results(self, concepts):
        # Result paging
        paging_data = False
        if 'Range' in self.request.headers:
            paging_data = parse_range_header(self.request.headers['Range'])
        count = len(concepts)
        if not paging_data:
            paging_data = {
                'start': 0,
                'finish': count - 1 if count > 0 else 0,
                'number': count
            }
        cslice = concepts[paging_data['start']:paging_data['finish']+1]
        self.request.response.headers[ascii_native_('Content-Range')] = \
            ascii_native_('items %d-%d/%d' % (
                paging_data['start'], paging_data['finish'], count
            ))
        return cslice

    @view_config(route_name='skosprovider.c', request_method='GET')
    def get_concept(self):
        scheme_id = self.request.matchdict['scheme_id']
        concept_id = self.request.matchdict['c_id']
        provider = self.skos_registry.get_provider(scheme_id)
        concept = provider.get_by_id(concept_id)
        if not concept:
            return HTTPNotFound()
        return concept

    @view_config(route_name='skosprovider.c.display_children', request_method='GET')
    def get_concept_display_children(self):
        scheme_id = self.request.matchdict['scheme_id']
        concept_id = self.request.matchdict['c_id']
        provider = self.skos_registry.get_provider(scheme_id)
        language = self.request.params.get('language', self.request.locale_name)
        children = provider.get_children_display(concept_id, language=language)
        if children is False:
            return HTTPNotFound()
        return children

    @view_config(route_name='skosprovider.c.expand', request_method='GET')
    def get_expand(self):
        scheme_id = self.request.matchdict['scheme_id']
        concept_id = self.request.matchdict['c_id']
        provider = self.skos_registry.get_provider(scheme_id)
        expanded = provider.expand(concept_id)
        if not expanded:
            return HTTPNotFound()
        return expanded


1			# -- coding: utf8 --
2			'''
3			This module contains the pyramid views that expose services.
4			'''
5
6			from __future__ import unicode_literals
7
8			import itertools
9
10			from pyramid.view import view_config, view_defaults
11
12			from pyramid.compat import ascii_native_
13
14			from pyramid.httpexceptions import (
15			HTTPNotFound,
16			HTTPBadRequest
17			)
18
19			from pyramid_skosprovider.utils import (
20			parse_range_header
21			)
22
23			import logging
24			log = logging.getLogger(__name__)
25
26
27			class RestView(object):
28
29			def __init__(self, request):
30			self.request = request
31			self.skos_registry = self.request.skos_registry
32
33
34			@view_defaults(renderer='skosjson', accept='application/json')
35			class ProviderView(RestView):
36			'''
37			A set of views that expose information from a certain provider.
38			'''
39
40			@view_config(route_name='skosprovider.uri', request_method='GET')
41			@view_config(route_name='skosprovider.uri.deprecated', request_method='GET')
42			def get_uri(self):
43			uri = self.request.params.get('uri')
44			if not uri:
45			if 'uri' in self.request.matchdict:
46			uri = self.request.matchdict['uri']
47			else:
48			return HTTPBadRequest()
49			provider = self.skos_registry.get_provider(uri)
50			if provider:
51			return {
52			'type': 'concept_scheme',
53			'uri': provider.concept_scheme.uri,
54			'id': provider.get_vocabulary_id()
55			}
56			c = self.skos_registry.get_by_uri(uri)
57			if c:
58			return {
59			'type': c.type,
60			'uri': c.uri,
61			'id': c.id,
62			'concept_scheme': {
63			'uri': c.concept_scheme.uri,
64			'id': self.skos_registry.get_provider(c.concept_scheme.uri).get_vocabulary_id()
65			}
66			}
67			if not c:
68			return HTTPNotFound()
69
70			@view_config(route_name='skosprovider.conceptschemes', request_method='GET')
71			def get_conceptschemes(self):
72			language = self.request.params.get('language', self.request.locale_name)
73			return [
74			{
75			'id': p.get_vocabulary_id(),
76			'uri': p.concept_scheme.uri,
77			'label': p.concept_scheme.label(language).label if p.concept_scheme.label(language) else None,
78			'subject': p.metadata['subject'] if p.metadata['subject'] else []
79			} for p in self.skos_registry.get_providers()
80			]
81
82			@view_config(route_name='skosprovider.conceptscheme', request_method='GET')
83			def get_conceptscheme(self):
84			scheme_id = self.request.matchdict['scheme_id']
85			provider = self.skos_registry.get_provider(scheme_id)
86			if not provider:
87			return HTTPNotFound()
88			language = self.request.params.get('language', self.request.locale_name)
89			return {
90			'id': provider.get_vocabulary_id(),
91			'uri': provider.concept_scheme.uri,
92			'label': provider.concept_scheme.label(language).label if provider.concept_scheme.label(language) else None,
93			'subject': provider.metadata['subject'] if provider.metadata['subject'] else [],
94			'labels': provider.concept_scheme.labels,
95			'notes': provider.concept_scheme.notes,
96			'sources': provider.concept_scheme.sources,
97			'languages': provider.concept_scheme.languages
98			}
99
100			@view_config(route_name='skosprovider.conceptscheme.tc', request_method='GET')
101			def get_conceptscheme_top_concepts(self):
102			scheme_id = self.request.matchdict['scheme_id']
103			provider = self.skos_registry.get_provider(scheme_id)
104			if not provider:
105			return HTTPNotFound()
106			language = self.request.params.get('language', self.request.locale_name)
107			return provider.get_top_concepts(language=language)
108
109			@view_config(route_name='skosprovider.conceptscheme.display_top', request_method='GET')
110			def get_conceptscheme_display_top(self):
111			scheme_id = self.request.matchdict['scheme_id']
112			provider = self.skos_registry.get_provider(scheme_id)
113			if not provider:
114			return HTTPNotFound()
115			language = self.request.params.get('language', self.request.locale_name)
116			return provider.get_top_display(language=language)
117
118			def _build_providers(self, request):
119			'''
120			:param pyramid.request.Request request:
121			:rtype: :class:`dict`
122			'''
123			# determine targets
124			providers = {}
125			ids = request.params.get('providers.ids', None)
126			if ids:
127			ids = ids.split(',')
128			providers['ids'] = ids
129			subject = self.request.params.get('providers.subject', None)
130			if subject:
131			providers['subject'] = subject
132			return providers
133
134			@view_config(route_name='skosprovider.cs', request_method='GET')
135			def get_concepts(self):
136			query = {}
137			mode = self.request.params.get('mode', 'default')
138			label = self.request.params.get('label', None)
139			postprocess = False
140			language = self.request.params.get('language', self.request.locale_name)
141			if mode == 'dijitFilteringSelect' and label == '':
142			concepts = []
143			else:
144			if label not in [None, '*', '']:
145			if mode == 'dijitFilteringSelect' and '*' in label:
146			postprocess = True
147			query['label'] = label.replace('*', '')
148			else:
149			query['label'] = label
150			type = self.request.params.get('type', None)
151			if type in ['concept', 'collection']:
152			query['type'] = type
153
154			concepts = self.skos_registry.find(
155			query,
156			providers=self._build_providers(self.request),
157			language=language
158			)
159			# Flatten it all
160			concepts = list(itertools.chain.from_iterable([c['concepts'] for c in concepts]))
161
162			if postprocess:
163			concepts = self._postprocess_wildcards(concepts, label)
164
165			concepts = self._sort_concepts(concepts)
166
167			return self._page_results(concepts)
168
169			@view_config(route_name='skosprovider.conceptscheme.cs', request_method='GET')
170			def get_conceptscheme_concepts(self):
171			scheme_id = self.request.matchdict['scheme_id']
172			provider = self.skos_registry.get_provider(scheme_id)
173			if not provider:
174			return HTTPNotFound()
175			query = {}
176			mode = self.request.params.get('mode', 'default')
177			label = self.request.params.get('label', None)
178			postprocess = False
179			if mode == 'dijitFilteringSelect' and label == '':
180			concepts = []
181			else:
182			if label not in [None, '*', '']:
183			if mode == 'dijitFilteringSelect' and '*' in label:
184			postprocess = True
185			query['label'] = label.replace('*', '')
186			else:
187			query['label'] = label
188			type = self.request.params.get('type', None)
189			if type in ['concept', 'collection']:
190			query['type'] = type
191			coll = self.request.params.get('collection', None)
192			if coll is not None:
193			query['collection'] = {'id': coll, 'depth': 'all'}
194			language = self.request.params.get('language', self.request.locale_name)
195			concepts = provider.find(query, language=language)
196
197			if postprocess:
198			concepts = self._postprocess_wildcards(concepts, label)
199
200			concepts = self._sort_concepts(concepts)
201
202			return self._page_results(concepts)
203
204			def _postprocess_wildcards(self, concepts, label):
205			# We need to refine results further
206			if label.startswith('') and label.endswith(''):
207			concepts = [c for c in concepts if label[1:-1] in c['label']]
208			elif label.endswith('*'):
209			concepts = [c for c in concepts if c['label'].startswith(label[0:-1])]
210			elif label.startswith('*'):
211			concepts = [c for c in concepts if c['label'].endswith(label[1:])]
212			return concepts
213
214			def _sort_concepts(self, concepts):
215			sort = self.request.params.get('sort', None)
216			# Result sorting
217			if sort:
218			sort_desc = (sort[0:1] == '-')
219			sort = sort[1:] if sort[0:1] in ['-', '+'] else sort
220			sort = sort.strip() # dojo store does not encode '+'
221			if (len(concepts) > 0) and (sort in concepts[0]):
222			if sort == 'label':
223			concepts.sort(
224			key=lambda concept: concept[sort].lower(),
225			reverse=sort_desc
226			)
227			else:
228			concepts.sort(
229			key=lambda concept: concept[sort],
230			reverse=sort_desc
231			)
232			return concepts
233
234			def _page_results(self, concepts):
235			# Result paging
236			paging_data = False
237			if 'Range' in self.request.headers:
238			paging_data = parse_range_header(self.request.headers['Range'])
239			count = len(concepts)
240			if not paging_data:
241			paging_data = {
242			'start': 0,
243			'finish': count - 1 if count > 0 else 0,
244			'number': count
245			}
246			cslice = concepts[paging_data['start']:paging_data['finish']+1]
247			self.request.response.headers[ascii_native_('Content-Range')] = \
248			ascii_native_('items %d-%d/%d' % (
249			paging_data['start'], paging_data['finish'], count
250			))
251			return cslice
252
253			@view_config(route_name='skosprovider.c', request_method='GET')
254			def get_concept(self):
255			scheme_id = self.request.matchdict['scheme_id']
256			concept_id = self.request.matchdict['c_id']
257			provider = self.skos_registry.get_provider(scheme_id)
258			concept = provider.get_by_id(concept_id)
259			if not concept:
260			return HTTPNotFound()
261			return concept
262
263			@view_config(route_name='skosprovider.c.display_children', request_method='GET')
264			def get_concept_display_children(self):
265			scheme_id = self.request.matchdict['scheme_id']
266			concept_id = self.request.matchdict['c_id']
267			provider = self.skos_registry.get_provider(scheme_id)
268			language = self.request.params.get('language', self.request.locale_name)
269			children = provider.get_children_display(concept_id, language=language)
270			if children is False:
271			return HTTPNotFound()
272			return children
273
274			@view_config(route_name='skosprovider.c.expand', request_method='GET')
275			def get_expand(self):
276			scheme_id = self.request.matchdict['scheme_id']
277			concept_id = self.request.matchdict['c_id']
278			provider = self.skos_registry.get_provider(scheme_id)
279			expanded = provider.expand(concept_id)
280			if not expanded:
281			return HTTPNotFound()
282			return expanded
283

OnroerendErfgoed / pyramid_skosprovider

Push — refactor_query ( 910c8b...1aad74 )

ProviderView.get_concept() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like