pyramid_skosprovider.views.ProviderView.get_conceptschemes() - Code Metrics - Inspection of "Better error handling of conceptschemes that fail..." - OnroerendErfgoed/pyramid_skosprovider - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#103)

by Koen

created 2023-10-19 21:10 UTC

ProviderView.get_conceptschemes() B

↳ Parent: pyramid_skosprovider.views

Complexity

Conditions

Size

Total Lines	37
Code Lines	33

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	33
nop	1
dl	0
loc	37
rs	8.1546
c	0
b	0
f	0

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

import itertools

from pyramid.view import view_config, view_defaults

from pyramid.httpexceptions import (
    HTTPNotFound,
    HTTPBadRequest
)

from skosprovider.exceptions import ProviderUnavailableException

from pyramid_skosprovider.utils import (
    parse_range_header,
    QueryBuilder
)

from skosprovider.jsonld import (
    MINI_CONTEXT,
    CONTEXT
)

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='json')
class StaticView(RestView):

    @view_config(
        route_name='skosprovider.context',
        request_method='GET',
        accept='application/json',
        http_cache=(3600, {'public': True})
    )
    @view_config(
        route_name='skosprovider.context',
        request_method='GET',
        accept='application/ld+json',
        http_cache=(3600, {'public': True})
    )
    def get_context(self):
        if 'application/ld+json' in self.request.accept:
            self.request.response.content_type = 'application/ld+json'
        return {
            '@context': CONTEXT
        }


class ProviderView(RestView):
    '''
    A set of views that expose information from a certain provider.
    '''

    @view_config(
        route_name='skosprovider.uri',
        request_method='GET',
        accept='application/json+ld',
        renderer='skosjson'
    )
    @view_config(
        route_name='skosprovider.uri',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    @view_config(
        route_name='skosprovider.uri.deprecated',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    def get_uri(self):
        uri = self.request.params.get('uri', self.request.matchdict.get('uri', None))
        if not uri:
            return HTTPBadRequest()
        if 'application/ld+json' in self.request.accept:
            self.request.response.content_type = 'application/ld+json'
        provider = self.skos_registry.get_provider(uri)
        uri_context = MINI_CONTEXT
        uri_context['concept_scheme'] = {
            '@id': 'skos:inScheme',
            '@type': '@id'
        }
        if provider:
            uri_context['concept_scheme'] = 'skos:ConceptScheme'
            return {
                '@context': uri_context,
                'type': 'concept_scheme',
                'uri': provider.concept_scheme.uri,
                'id': provider.get_vocabulary_id()
            }
        c = self.skos_registry.get_by_uri(uri)
        if not c:
            return HTTPNotFound()
        return {
            '@context': uri_context,
            'type': c.type,
            'uri': c.uri,
            'id': c.id,
            'concept_scheme': {
                'type': 'skos:ConceptScheme',
                'uri': c.concept_scheme.uri,
                'id': self.skos_registry.get_provider(c.concept_scheme.uri).get_vocabulary_id()
            }
        }

    @view_config(
        route_name='skosprovider.conceptschemes',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    @view_config(
        route_name='skosprovider.conceptschemes',
        request_method='GET',
        accept='application/ld+json'
    )
    def get_conceptschemes(self):
        language = self.request.params.get('language', self.request.locale_name)
        if 'application/ld+json' in self.request.accept:
            self.request.response.content_type = 'application/ld+json'
        context = MINI_CONTEXT
        context['subject'] = {
            '@id': 'dct:subject',
            '@type': '@id'
        }
        conceptschemes = []
        for p in self.skos_registry.get_providers():
            try:
                cslabel = p.concept_scheme.label(language).label if p.concept_scheme.label(language) else None,
            except ProviderUnavailableException as e:
                log.error('Could not fetch label for {p.get_vocabulary_uri()}', e) 
                cslabel = p.get_vocabulary_uri()
            cs = {
                '@context': context,
                'type': 'skos:ConceptScheme',
                'id': p.get_vocabulary_id(),
                'uri': p.get_vocabulary_uri(),
                'label': cslabel,
                'subject': p.metadata['subject'] if p.metadata['subject'] else []
            }
            conceptschemes.append(cs)
        return conceptschemes

    @view_config(
        route_name='skosprovider.conceptscheme',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    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',
        request_method='GET',
        renderer='skosjsonld',
        accept='application/ld+json'
    )
    @view_config(
        route_name='skosprovider.conceptscheme.jsonld',
        request_method='GET',
        renderer='skosjsonld'
    )
    def get_conceptscheme_jsonld(self):
        scheme_id = self.request.matchdict['scheme_id']
        provider = self.skos_registry.get_provider(scheme_id)
        if not provider:
            return HTTPNotFound()
        return provider.concept_scheme

    @view_config(
        route_name='skosprovider.conceptscheme.tc',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    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',
        accept='application/json',
        renderer='skosjson'
    )
    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',
        accept='application/json',
        renderer='skosjson'
    )
    def get_concepts(self):
        qb = QueryBuilder(self.request)
        query = qb()
        kwargs = {"language": qb.language, "providers": self._build_providers(self.request)}
        kwargs.update(self._get_sort_params())
        if qb.no_result:
            concepts = []
        else:
            concepts = self.skos_registry.find(query, **kwargs)
            # Flatten it all
            concepts = list(itertools.chain.from_iterable([c['concepts'] for c in concepts]))

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

        return self._page_results(concepts)

    @view_config(
        route_name='skosprovider.conceptscheme.cs',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    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()
        qb = QueryBuilder(self.request)
        query = qb()
        kwargs = {"language": qb.language}
        kwargs.update(self._get_sort_params())
        if qb.no_result:
            concepts = []
        else:
            concepts = provider.find(query, **kwargs)

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

        return self._page_results(concepts)

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

    def _get_sort_params(self):
        sort = self.request.params.get('sort', None)
        # Result sorting
        if sort:
            sort_order = 'desc' if sort[0:1] == '-' else 'asc'
            sort = sort[1:] if sort[0:1] in ['-', '+'] else sort
            sort = sort.strip().lower()  # dojo store does not encode '+'
            return {"sort": sort, "sort_order": sort_order}
        return {}

    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]
        if len(cslice):
            cslice[0]['@context'] = self.request.route_url('skosprovider.context')
        self.request.response.headers['Content-Range'] = \
            'items %d-%d/%d' % (
                paging_data['start'], paging_data['finish'], count
            )
        return cslice

    @view_config(
        route_name='skosprovider.c',
        request_method='GET',
        accept='application/json',
        renderer='skosjson'
    )
    @view_config(
        route_name='skosprovider.c',
        request_method='GET',
        renderer='skosjsonld',
        accept='application/ld+json'
    )
    @view_config(
        route_name='skosprovider.c.jsonld',
        request_method='GET',
        renderer='skosjsonld'
    )
    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)
        if not provider:
            return HTTPNotFound()
        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',
        accept='application/json',
        renderer='skosjson'
    )
    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)
        if not provider:
            return HTTPNotFound()
        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',
        accept='application/json',
        renderer='skosjson'
    )
    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)
        if not provider:
            return HTTPNotFound()
        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			import itertools
7
8			from pyramid.view import view_config, view_defaults
9
10			from pyramid.httpexceptions import (
11			HTTPNotFound,
12			HTTPBadRequest
13			)
14
15			from skosprovider.exceptions import ProviderUnavailableException
16
17			from pyramid_skosprovider.utils import (
18			parse_range_header,
19			QueryBuilder
20			)
21
22			from skosprovider.jsonld import (
23			MINI_CONTEXT,
24			CONTEXT
25			)
26
27			import logging
28			log = logging.getLogger(__name__)
29
30
31			class RestView(object):
32
33			def __init__(self, request):
34			self.request = request
35			self.skos_registry = self.request.skos_registry
36
37
38			@view_defaults(renderer='json')
39			class StaticView(RestView):
40
41			@view_config(
42			route_name='skosprovider.context',
43			request_method='GET',
44			accept='application/json',
45			http_cache=(3600, {'public': True})
46			)
47			@view_config(
48			route_name='skosprovider.context',
49			request_method='GET',
50			accept='application/ld+json',
51			http_cache=(3600, {'public': True})
52			)
53			def get_context(self):
54			if 'application/ld+json' in self.request.accept:
55			self.request.response.content_type = 'application/ld+json'
56			return {
57			'@context': CONTEXT
58			}
59
60
61			class ProviderView(RestView):
62			'''
63			A set of views that expose information from a certain provider.
64			'''
65
66			@view_config(
67			route_name='skosprovider.uri',
68			request_method='GET',
69			accept='application/json+ld',
70			renderer='skosjson'
71			)
72			@view_config(
73			route_name='skosprovider.uri',
74			request_method='GET',
75			accept='application/json',
76			renderer='skosjson'
77			)
78			@view_config(
79			route_name='skosprovider.uri.deprecated',
80			request_method='GET',
81			accept='application/json',
82			renderer='skosjson'
83			)
84			def get_uri(self):
85			uri = self.request.params.get('uri', self.request.matchdict.get('uri', None))
86			if not uri:
87			return HTTPBadRequest()
88			if 'application/ld+json' in self.request.accept:
89			self.request.response.content_type = 'application/ld+json'
90			provider = self.skos_registry.get_provider(uri)
91			uri_context = MINI_CONTEXT
92			uri_context['concept_scheme'] = {
93			'@id': 'skos:inScheme',
94			'@type': '@id'
95			}
96			if provider:
97			uri_context['concept_scheme'] = 'skos:ConceptScheme'
98			return {
99			'@context': uri_context,
100			'type': 'concept_scheme',
101			'uri': provider.concept_scheme.uri,
102			'id': provider.get_vocabulary_id()
103			}
104			c = self.skos_registry.get_by_uri(uri)
105			if not c:
106			return HTTPNotFound()
107			return {
108			'@context': uri_context,
109			'type': c.type,
110			'uri': c.uri,
111			'id': c.id,
112			'concept_scheme': {
113			'type': 'skos:ConceptScheme',
114			'uri': c.concept_scheme.uri,
115			'id': self.skos_registry.get_provider(c.concept_scheme.uri).get_vocabulary_id()
116			}
117			}
118
119			@view_config(
120			route_name='skosprovider.conceptschemes',
121			request_method='GET',
122			accept='application/json',
123			renderer='skosjson'
124			)
125			@view_config(
126			route_name='skosprovider.conceptschemes',
127			request_method='GET',
128			accept='application/ld+json'
129			)
130			def get_conceptschemes(self):
131			language = self.request.params.get('language', self.request.locale_name)
132			if 'application/ld+json' in self.request.accept:
133			self.request.response.content_type = 'application/ld+json'
134			context = MINI_CONTEXT
135			context['subject'] = {
136			'@id': 'dct:subject',
137			'@type': '@id'
138			}
139			conceptschemes = []
140			for p in self.skos_registry.get_providers():
141			try:
142			cslabel = p.concept_scheme.label(language).label if p.concept_scheme.label(language) else None,
143			except ProviderUnavailableException as e:
144			log.error('Could not fetch label for {p.get_vocabulary_uri()}', e)
145			cslabel = p.get_vocabulary_uri()
146			cs = {
147			'@context': context,
148			'type': 'skos:ConceptScheme',
149			'id': p.get_vocabulary_id(),
150			'uri': p.get_vocabulary_uri(),
151			'label': cslabel,
152			'subject': p.metadata['subject'] if p.metadata['subject'] else []
153			}
154			conceptschemes.append(cs)
155			return conceptschemes
156
157			@view_config(
158			route_name='skosprovider.conceptscheme',
159			request_method='GET',
160			accept='application/json',
161			renderer='skosjson'
162			)
163			def get_conceptscheme(self):
164			scheme_id = self.request.matchdict['scheme_id']
165			provider = self.skos_registry.get_provider(scheme_id)
166			if not provider:
167			return HTTPNotFound()
168			language = self.request.params.get('language', self.request.locale_name)
169			return {
170			'id': provider.get_vocabulary_id(),
171			'uri': provider.concept_scheme.uri,
172			'label': provider.concept_scheme.label(language).label if provider.concept_scheme.label(language) else None,
173			'subject': provider.metadata['subject'] if provider.metadata['subject'] else [],
174			'labels': provider.concept_scheme.labels,
175			'notes': provider.concept_scheme.notes,
176			'sources': provider.concept_scheme.sources,
177			'languages': provider.concept_scheme.languages
178			}
179
180			@view_config(
181			route_name='skosprovider.conceptscheme',
182			request_method='GET',
183			renderer='skosjsonld',
184			accept='application/ld+json'
185			)
186			@view_config(
187			route_name='skosprovider.conceptscheme.jsonld',
188			request_method='GET',
189			renderer='skosjsonld'
190			)
191			def get_conceptscheme_jsonld(self):
192			scheme_id = self.request.matchdict['scheme_id']
193			provider = self.skos_registry.get_provider(scheme_id)
194			if not provider:
195			return HTTPNotFound()
196			return provider.concept_scheme
197
198			@view_config(
199			route_name='skosprovider.conceptscheme.tc',
200			request_method='GET',
201			accept='application/json',
202			renderer='skosjson'
203			)
204			def get_conceptscheme_top_concepts(self):
205			scheme_id = self.request.matchdict['scheme_id']
206			provider = self.skos_registry.get_provider(scheme_id)
207			if not provider:
208			return HTTPNotFound()
209			language = self.request.params.get('language', self.request.locale_name)
210			return provider.get_top_concepts(language=language)
211
212			@view_config(
213			route_name='skosprovider.conceptscheme.display_top',
214			request_method='GET',
215			accept='application/json',
216			renderer='skosjson'
217			)
218			def get_conceptscheme_display_top(self):
219			scheme_id = self.request.matchdict['scheme_id']
220			provider = self.skos_registry.get_provider(scheme_id)
221			if not provider:
222			return HTTPNotFound()
223			language = self.request.params.get('language', self.request.locale_name)
224			return provider.get_top_display(language=language)
225
226			def _build_providers(self, request):
227			'''
228			:param pyramid.request.Request request:
229			:rtype: :class:`dict`
230			'''
231			# determine targets
232			providers = {}
233			ids = request.params.get('providers.ids', None)
234			if ids:
235			ids = ids.split(',')
236			providers['ids'] = ids
237			subject = self.request.params.get('providers.subject', None)
238			if subject:
239			providers['subject'] = subject
240			return providers
241
242			@view_config(
243			route_name='skosprovider.cs',
244			request_method='GET',
245			accept='application/json',
246			renderer='skosjson'
247			)
248			def get_concepts(self):
249			qb = QueryBuilder(self.request)
250			query = qb()
251			kwargs = {"language": qb.language, "providers": self._build_providers(self.request)}
252			kwargs.update(self._get_sort_params())
253			if qb.no_result:
254			concepts = []
255			else:
256			concepts = self.skos_registry.find(query, **kwargs)
257			# Flatten it all
258			concepts = list(itertools.chain.from_iterable([c['concepts'] for c in concepts]))
259
260			if qb.postprocess:
261			concepts = self._postprocess_wildcards(concepts, qb.label)
262
263			return self._page_results(concepts)
264
265			@view_config(
266			route_name='skosprovider.conceptscheme.cs',
267			request_method='GET',
268			accept='application/json',
269			renderer='skosjson'
270			)
271			def get_conceptscheme_concepts(self):
272			scheme_id = self.request.matchdict['scheme_id']
273			provider = self.skos_registry.get_provider(scheme_id)
274			if not provider:
275			return HTTPNotFound()
276			qb = QueryBuilder(self.request)
277			query = qb()
278			kwargs = {"language": qb.language}
279			kwargs.update(self._get_sort_params())
280			if qb.no_result:
281			concepts = []
282			else:
283			concepts = provider.find(query, **kwargs)
284
285			if qb.postprocess:
286			concepts = self._postprocess_wildcards(concepts, qb.label)
287
288			return self._page_results(concepts)
289
290			@staticmethod
291			def _postprocess_wildcards(concepts, label):
292			# We need to refine results further
293			if label.startswith('') and not label.endswith(''):
294			concepts = [
295			c for c in concepts if c['label'].lower().endswith(label[1:].lower())
296			]
297			elif label.endswith('') and not label.startswith(''):
298			concepts = [
299			c for c in concepts if c['label'].lower().startswith(label[:-1].lower())
300			]
301			return concepts
302
303			def _get_sort_params(self):
304			sort = self.request.params.get('sort', None)
305			# Result sorting
306			if sort:
307			sort_order = 'desc' if sort[0:1] == '-' else 'asc'
308			sort = sort[1:] if sort[0:1] in ['-', '+'] else sort
309			sort = sort.strip().lower() # dojo store does not encode '+'
310			return {"sort": sort, "sort_order": sort_order}
311			return {}
312
313			def _page_results(self, concepts):
314			# Result paging
315			paging_data = False
316			if 'Range' in self.request.headers:
317			paging_data = parse_range_header(self.request.headers['Range'])
318			count = len(concepts)
319			if not paging_data:
320			paging_data = {
321			'start': 0,
322			'finish': count - 1 if count > 0 else 0,
323			'number': count
324			}
325			cslice = concepts[paging_data['start']:paging_data['finish']+1]
326			if len(cslice):
327			cslice[0]['@context'] = self.request.route_url('skosprovider.context')
328			self.request.response.headers['Content-Range'] = \
329			'items %d-%d/%d' % (
330			paging_data['start'], paging_data['finish'], count
331			)
332			return cslice
333
334			@view_config(
335			route_name='skosprovider.c',
336			request_method='GET',
337			accept='application/json',
338			renderer='skosjson'
339			)
340			@view_config(
341			route_name='skosprovider.c',
342			request_method='GET',
343			renderer='skosjsonld',
344			accept='application/ld+json'
345			)
346			@view_config(
347			route_name='skosprovider.c.jsonld',
348			request_method='GET',
349			renderer='skosjsonld'
350			)
351			def get_concept(self):
352			scheme_id = self.request.matchdict['scheme_id']
353			concept_id = self.request.matchdict['c_id']
354			provider = self.skos_registry.get_provider(scheme_id)
355			if not provider:
356			return HTTPNotFound()
357			concept = provider.get_by_id(concept_id)
358			if not concept:
359			return HTTPNotFound()
360			return concept
361
362			@view_config(
363			route_name='skosprovider.c.display_children',
364			request_method='GET',
365			accept='application/json',
366			renderer='skosjson'
367			)
368			def get_concept_display_children(self):
369			scheme_id = self.request.matchdict['scheme_id']
370			concept_id = self.request.matchdict['c_id']
371			provider = self.skos_registry.get_provider(scheme_id)
372			if not provider:
373			return HTTPNotFound()
374			language = self.request.params.get('language', self.request.locale_name)
375			children = provider.get_children_display(concept_id, language=language)
376			if children is False:
377			return HTTPNotFound()
378			return children
379
380			@view_config(
381			route_name='skosprovider.c.expand',
382			request_method='GET',
383			accept='application/json',
384			renderer='skosjson'
385			)
386			def get_expand(self):
387			scheme_id = self.request.matchdict['scheme_id']
388			concept_id = self.request.matchdict['c_id']
389			provider = self.skos_registry.get_provider(scheme_id)
390			if not provider:
391			return HTTPNotFound()
392			expanded = provider.expand(concept_id)
393			if not expanded:
394			return HTTPNotFound()
395			return expanded
396

OnroerendErfgoed / pyramid_skosprovider

Pull Request — develop (#103)

ProviderView.get_conceptschemes() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like