atramhasis.views.crud.AtramhasisCrud.get_providers() - Code Metrics - Inspection of "Add GET provider endpoints. (#795)" - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( da51e8...70dbf1 )

unknown

created 2023-03-09 12:38 UTC

AtramhasisCrud.get_providers() A

↳ Parent: atramhasis.views.crud

Complexity

Conditions

Size

Total Lines	13
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	11
nop	1
dl	0
loc	13
rs	9.85
c	0
b	0
f	0

"""
Module containing views related to the REST service.
"""
import logging
import time

import colander
import transaction
from pyramid.httpexceptions import HTTPMethodNotAllowed
from pyramid.view import view_config
from pyramid.view import view_defaults
from pyramid_skosprovider.views import ProviderView
from skosprovider_sqlalchemy.models import Collection
from skosprovider_sqlalchemy.models import Concept
from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import NoResultFound

from atramhasis.audit import audit
from atramhasis.cache import invalidate_scheme_cache
from atramhasis.errors import ConceptNotFoundException
from atramhasis.errors import ConceptSchemeNotFoundException
from atramhasis.errors import SkosRegistryNotFoundException
from atramhasis.errors import ValidationError
from atramhasis.mappers import map_concept
from atramhasis.mappers import map_conceptscheme
from atramhasis.protected_resources import protected_operation
from atramhasis.skos import IDGenerationStrategy
from atramhasis.utils import from_thing
from atramhasis.utils import internal_providers_only

LOG = logging.getLogger(__name__)


@view_defaults(accept='application/json', renderer='skosrenderer_verbose')
class AtramhasisCrud:
    """
    This object groups CRUD REST views part of the private user interface.
    """

    def __init__(self, context, request):
        self.request = request
        self.skos_manager = self.request.data_managers['skos_manager']
        self.conceptscheme_manager = self.request.data_managers['conceptscheme_manager']
        self.context = context
        self.logged_in = request.authenticated_userid
        if 'scheme_id' in self.request.matchdict:
            self.scheme_id = self.request.matchdict['scheme_id']
            if hasattr(request, 'skos_registry') and request.skos_registry is not None:
                self.skos_registry = self.request.skos_registry
            else:
                raise SkosRegistryNotFoundException()
            self.provider = self.skos_registry.get_provider(self.scheme_id)
            if not self.provider:
                raise ConceptSchemeNotFoundException(self.scheme_id)

    def _get_json_body(self):
        json_body = self.request.json_body
        if 'c_id' in self.request.matchdict and 'id' not in json_body:
            json_body['id'] = self.request.matchdict['c_id']
        return json_body

    def _validate_concept(self, json_concept, provider):
        from atramhasis.validators import (
            Concept as ConceptSchema,
            concept_schema_validator
        )

        concept_schema = ConceptSchema(
            validator=concept_schema_validator
        ).bind(
            request=self.request,
            provider=provider
        )
        try:
            return concept_schema.deserialize(json_concept)
        except colander.Invalid as e:
            raise ValidationError(
                'Concept could not be validated',
                e.asdict()
            )

    def _validate_conceptscheme(self, json_conceptscheme):
        from atramhasis.validators import (
            ConceptScheme as ConceptSchemeSchema,
            conceptscheme_schema_validator
        )

        conceptscheme_schema = ConceptSchemeSchema(
            validator=conceptscheme_schema_validator
        ).bind(
            request=self.request
        )
        try:
            return conceptscheme_schema.deserialize(json_conceptscheme)
        except colander.Invalid as e:  # pragma no cover
            # I doubt this code will ever be reached, keeping it here just in case
            raise ValidationError(
                'Conceptscheme could not be validated',
                e.asdict()
            )

    @audit
    @view_config(route_name='atramhasis.get_conceptscheme', permission='view')
    def get_conceptscheme(self):
        if self.request.method == 'DELETE':
            raise HTTPMethodNotAllowed
        # is the same as the pyramid_skosprovider get_conceptscheme function, but wrapped with the audit function
        return ProviderView(self.request).get_conceptscheme()

    @view_config(route_name='atramhasis.edit_conceptscheme', permission='edit')
    def edit_conceptscheme(self):
        """
        Edit an existing concept

        :raises atramhasis.errors.ValidationError: If the provided json can't be validated
        """
        validated_json_conceptscheme = self._validate_conceptscheme(self._get_json_body())
        conceptscheme = self.conceptscheme_manager.get(self.provider.conceptscheme_id)
        conceptscheme = map_conceptscheme(conceptscheme, validated_json_conceptscheme)
        conceptscheme = self.conceptscheme_manager.save(conceptscheme)
        self.request.response.status = '200'
        return conceptscheme

    @view_config(route_name='atramhasis.get_conceptschemes', permission='view')
    def get_conceptschemes(self):
        if self.request.method == 'POST':
            raise HTTPMethodNotAllowed
        # is the same as the pyramid_skosprovider get_conceptscheme function, method not allowed included
        from pyramid_skosprovider.views import ProviderView
        return ProviderView(self.request).get_conceptschemes()

    @audit
    @view_config(route_name='atramhasis.get_concept', permission='view')
    def get_concept(self):
        """
        Get an existing concept

        :raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
        """
        c_id = self.request.matchdict['c_id']
        if isinstance(self.provider, SQLAlchemyProvider):
            try:
                concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
            except NoResultFound:
                raise ConceptNotFoundException(c_id)
        else:
            concept = self.provider.get_by_id(c_id)
            if not concept:
                raise ConceptNotFoundException(c_id)

        self.request.response.status = '200'
        return concept

    @internal_providers_only
    @view_config(route_name='atramhasis.add_concept', permission='edit')
    def add_concept(self):
        """
        Add a new concept to a conceptscheme

        :raises atramhasis.errors.ValidationError: If the provided json can't be validated
        """
        validated_json_concept = self._validate_concept(self._get_json_body(), self.provider)
        exc = None
        id_generation_strategy = IDGenerationStrategy.NUMERIC
        for _ in range(5):
            try:
                if validated_json_concept['type'] == 'concept':
                    concept = Concept()
                else:
                    concept = Collection()

                id_generation_strategy = self.provider.metadata.get(
                    "atramhasis.id_generation_strategy", IDGenerationStrategy.NUMERIC
                )
                if id_generation_strategy == IDGenerationStrategy.MANUAL:
                    concept.concept_id = validated_json_concept["concept_id"]
                else:
                    concept.concept_id = self.skos_manager.get_next_cid(
                        self.provider.conceptscheme_id, id_generation_strategy
                    )

                concept.conceptscheme_id = self.provider.conceptscheme_id
                concept.uri = self.provider.uri_generator.generate(id=concept.concept_id)
                map_concept(concept, validated_json_concept, self.skos_manager)
                concept = self.skos_manager.save(concept)
                break
            except IntegrityError as exc:
                if id_generation_strategy == IDGenerationStrategy.MANUAL:
                    # Technically the concept_id is not 100% guaranteed to be the cause
                    # of an IntegrityError, so log trace just in case.
                    LOG.exception("Integrity error")

                    concept_id = validated_json_concept["concept_id"]
                    raise ValidationError(
                        "Integrity error", [{"concept_id": f"{concept_id} already exists."}]
                    )
                # There is a small chance that another concept gets added at the same
                # time. There is nothing wrong with the request, so we try again.
                transaction.abort()
                time.sleep(0.05)
        else:
            raise Exception(
                f"Could not save new concept due to IntegrityErrors. {exc}"
            )

        invalidate_scheme_cache(self.scheme_id)

        self.request.response.status = '201'
        self.request.response.location = self.request.route_path(
            'skosprovider.c', scheme_id=self.scheme_id, c_id=concept.concept_id)
        return from_thing(concept)

    @internal_providers_only
    @view_config(route_name='atramhasis.edit_concept', permission='edit')
    def edit_concept(self):
        """
        Edit an existing concept

        :raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
        :raises atramhasis.errors.ValidationError: If the provided json can't be validated
        """
        c_id = self.request.matchdict['c_id']
        validated_json_concept = self._validate_concept(self._get_json_body(), self.provider)
        try:
            concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
        except NoResultFound:
            raise ConceptNotFoundException(c_id)
        concept = map_concept(concept, validated_json_concept, self.skos_manager)

        invalidate_scheme_cache(self.scheme_id)

        self.request.response.status = '200'
        return from_thing(concept)

    @internal_providers_only
    @protected_operation
    @view_config(route_name='atramhasis.delete_concept', permission='delete')
    def delete_concept(self):
        """
        Delete an existing concept

        :raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
        """
        c_id = self.request.matchdict['c_id']
        try:
            concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
        except NoResultFound:
            raise ConceptNotFoundException(c_id)
        result = from_thing(concept)
        self.skos_manager.delete_thing(concept)

        invalidate_scheme_cache(self.scheme_id)

        self.request.response.status = '200'
        return result

    @view_config(
        route_name='atramhasis.providers',
        permission='view',
        request_method='GET',
        openapi=True,
    )
    def get_providers(self):
        query_params = self.request.openapi_validated.parameters.query
        if 'subject' in query_params:
            filters = {'subject': query_params['subject']}
        else:
            filters = {}
        return self.request.skos_registry.get_providers(**filters)

    @view_config(
        route_name='atramhasis.provider',
        permission='view',
        request_method='GET',
        openapi=True
    )
    def get_provider(self):
        return self.request.skos_registry.get_provider(self.request.matchdict["id"])


1			"""
2			Module containing views related to the REST service.
3			"""
4			import logging
5			import time
6
7			import colander
8			import transaction
9			from pyramid.httpexceptions import HTTPMethodNotAllowed
10			from pyramid.view import view_config
11			from pyramid.view import view_defaults
12			from pyramid_skosprovider.views import ProviderView
13			from skosprovider_sqlalchemy.models import Collection
14			from skosprovider_sqlalchemy.models import Concept
15			from skosprovider_sqlalchemy.providers import SQLAlchemyProvider
16			from sqlalchemy.exc import IntegrityError
17			from sqlalchemy.exc import NoResultFound
18
19			from atramhasis.audit import audit
20			from atramhasis.cache import invalidate_scheme_cache
21			from atramhasis.errors import ConceptNotFoundException
22			from atramhasis.errors import ConceptSchemeNotFoundException
23			from atramhasis.errors import SkosRegistryNotFoundException
24			from atramhasis.errors import ValidationError
25			from atramhasis.mappers import map_concept
26			from atramhasis.mappers import map_conceptscheme
27			from atramhasis.protected_resources import protected_operation
28			from atramhasis.skos import IDGenerationStrategy
29			from atramhasis.utils import from_thing
30			from atramhasis.utils import internal_providers_only
31
32			LOG = logging.getLogger(__name__)
33
34
35			@view_defaults(accept='application/json', renderer='skosrenderer_verbose')
36			class AtramhasisCrud:
37			"""
38			This object groups CRUD REST views part of the private user interface.
39			"""
40
41			def __init__(self, context, request):
42			self.request = request
43			self.skos_manager = self.request.data_managers['skos_manager']
44			self.conceptscheme_manager = self.request.data_managers['conceptscheme_manager']
45			self.context = context
46			self.logged_in = request.authenticated_userid
47			if 'scheme_id' in self.request.matchdict:
48			self.scheme_id = self.request.matchdict['scheme_id']
49			if hasattr(request, 'skos_registry') and request.skos_registry is not None:
50			self.skos_registry = self.request.skos_registry
51			else:
52			raise SkosRegistryNotFoundException()
53			self.provider = self.skos_registry.get_provider(self.scheme_id)
54			if not self.provider:
55			raise ConceptSchemeNotFoundException(self.scheme_id)
56
57			def _get_json_body(self):
58			json_body = self.request.json_body
59			if 'c_id' in self.request.matchdict and 'id' not in json_body:
60			json_body['id'] = self.request.matchdict['c_id']
61			return json_body
62
63			def _validate_concept(self, json_concept, provider):
64			from atramhasis.validators import (
65			Concept as ConceptSchema,
66			concept_schema_validator
67			)
68
69			concept_schema = ConceptSchema(
70			validator=concept_schema_validator
71			).bind(
72			request=self.request,
73			provider=provider
74			)
75			try:
76			return concept_schema.deserialize(json_concept)
77			except colander.Invalid as e:
78			raise ValidationError(
79			'Concept could not be validated',
80			e.asdict()
81			)
82
83			def _validate_conceptscheme(self, json_conceptscheme):
84			from atramhasis.validators import (
85			ConceptScheme as ConceptSchemeSchema,
86			conceptscheme_schema_validator
87			)
88
89			conceptscheme_schema = ConceptSchemeSchema(
90			validator=conceptscheme_schema_validator
91			).bind(
92			request=self.request
93			)
94			try:
95			return conceptscheme_schema.deserialize(json_conceptscheme)
96			except colander.Invalid as e: # pragma no cover
97			# I doubt this code will ever be reached, keeping it here just in case
98			raise ValidationError(
99			'Conceptscheme could not be validated',
100			e.asdict()
101			)
102
103			@audit
104			@view_config(route_name='atramhasis.get_conceptscheme', permission='view')
105			def get_conceptscheme(self):
106			if self.request.method == 'DELETE':
107			raise HTTPMethodNotAllowed
108			# is the same as the pyramid_skosprovider get_conceptscheme function, but wrapped with the audit function
109			return ProviderView(self.request).get_conceptscheme()
110
111			@view_config(route_name='atramhasis.edit_conceptscheme', permission='edit')
112			def edit_conceptscheme(self):
113			"""
114			Edit an existing concept
115
116			:raises atramhasis.errors.ValidationError: If the provided json can't be validated
117			"""
118			validated_json_conceptscheme = self._validate_conceptscheme(self._get_json_body())
119			conceptscheme = self.conceptscheme_manager.get(self.provider.conceptscheme_id)
120			conceptscheme = map_conceptscheme(conceptscheme, validated_json_conceptscheme)
121			conceptscheme = self.conceptscheme_manager.save(conceptscheme)
122			self.request.response.status = '200'
123			return conceptscheme
124
125			@view_config(route_name='atramhasis.get_conceptschemes', permission='view')
126			def get_conceptschemes(self):
127			if self.request.method == 'POST':
128			raise HTTPMethodNotAllowed
129			# is the same as the pyramid_skosprovider get_conceptscheme function, method not allowed included
130			from pyramid_skosprovider.views import ProviderView
131			return ProviderView(self.request).get_conceptschemes()
132
133			@audit
134			@view_config(route_name='atramhasis.get_concept', permission='view')
135			def get_concept(self):
136			"""
137			Get an existing concept
138
139			:raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
140			"""
141			c_id = self.request.matchdict['c_id']
142			if isinstance(self.provider, SQLAlchemyProvider):
143			try:
144			concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
145			except NoResultFound:
146			raise ConceptNotFoundException(c_id)
147			else:
148			concept = self.provider.get_by_id(c_id)
149			if not concept:
150			raise ConceptNotFoundException(c_id)
151
152			self.request.response.status = '200'
153			return concept
154
155			@internal_providers_only
156			@view_config(route_name='atramhasis.add_concept', permission='edit')
157			def add_concept(self):
158			"""
159			Add a new concept to a conceptscheme
160
161			:raises atramhasis.errors.ValidationError: If the provided json can't be validated
162			"""
163			validated_json_concept = self._validate_concept(self._get_json_body(), self.provider)
164			exc = None
165			id_generation_strategy = IDGenerationStrategy.NUMERIC
166			for _ in range(5):
167			try:
168			if validated_json_concept['type'] == 'concept':
169			concept = Concept()
170			else:
171			concept = Collection()
172
173			id_generation_strategy = self.provider.metadata.get(
174			"atramhasis.id_generation_strategy", IDGenerationStrategy.NUMERIC
175			)
176			if id_generation_strategy == IDGenerationStrategy.MANUAL:
177			concept.concept_id = validated_json_concept["concept_id"]
178			else:
179			concept.concept_id = self.skos_manager.get_next_cid(
180			self.provider.conceptscheme_id, id_generation_strategy
181			)
182
183			concept.conceptscheme_id = self.provider.conceptscheme_id
184			concept.uri = self.provider.uri_generator.generate(id=concept.concept_id)
185			map_concept(concept, validated_json_concept, self.skos_manager)
186			concept = self.skos_manager.save(concept)
187			break
188			except IntegrityError as exc:
189			if id_generation_strategy == IDGenerationStrategy.MANUAL:
190			# Technically the concept_id is not 100% guaranteed to be the cause
191			# of an IntegrityError, so log trace just in case.
192			LOG.exception("Integrity error")
193
194			concept_id = validated_json_concept["concept_id"]
195			raise ValidationError(
196			"Integrity error", [{"concept_id": f"{concept_id} already exists."}]
197			)
198			# There is a small chance that another concept gets added at the same
199			# time. There is nothing wrong with the request, so we try again.
200			transaction.abort()
201			time.sleep(0.05)
202			else:
203			raise Exception(
204			f"Could not save new concept due to IntegrityErrors. {exc}"
205			)
206
207			invalidate_scheme_cache(self.scheme_id)
208
209			self.request.response.status = '201'
210			self.request.response.location = self.request.route_path(
211			'skosprovider.c', scheme_id=self.scheme_id, c_id=concept.concept_id)
212			return from_thing(concept)
213
214			@internal_providers_only
215			@view_config(route_name='atramhasis.edit_concept', permission='edit')
216			def edit_concept(self):
217			"""
218			Edit an existing concept
219
220			:raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
221			:raises atramhasis.errors.ValidationError: If the provided json can't be validated
222			"""
223			c_id = self.request.matchdict['c_id']
224			validated_json_concept = self._validate_concept(self._get_json_body(), self.provider)
225			try:
226			concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
227			except NoResultFound:
228			raise ConceptNotFoundException(c_id)
229			concept = map_concept(concept, validated_json_concept, self.skos_manager)
230
231			invalidate_scheme_cache(self.scheme_id)
232
233			self.request.response.status = '200'
234			return from_thing(concept)
235
236			@internal_providers_only
237			@protected_operation
238			@view_config(route_name='atramhasis.delete_concept', permission='delete')
239			def delete_concept(self):
240			"""
241			Delete an existing concept
242
243			:raises atramhasis.errors.ConceptNotFoundException: If the concept can't be found
244			"""
245			c_id = self.request.matchdict['c_id']
246			try:
247			concept = self.skos_manager.get_thing(c_id, self.provider.conceptscheme_id)
248			except NoResultFound:
249			raise ConceptNotFoundException(c_id)
250			result = from_thing(concept)
251			self.skos_manager.delete_thing(concept)
252
253			invalidate_scheme_cache(self.scheme_id)
254
255			self.request.response.status = '200'
256			return result
257
258			@view_config(
259			route_name='atramhasis.providers',
260			permission='view',
261			request_method='GET',
262			openapi=True,
263			)
264			def get_providers(self):
265			query_params = self.request.openapi_validated.parameters.query
266			if 'subject' in query_params:
267			filters = {'subject': query_params['subject']}
268			else:
269			filters = {}
270			return self.request.skos_registry.get_providers(**filters)
271
272			@view_config(
273			route_name='atramhasis.provider',
274			permission='view',
275			request_method='GET',
276			openapi=True
277			)
278			def get_provider(self):
279			return self.request.skos_registry.get_provider(self.request.matchdict["id"])
280

OnroerendErfgoed / atramhasis

Push — develop ( da51e8...70dbf1 )

AtramhasisCrud.get_providers() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like