atramhasis.views.crud.AtramhasisCrud.get_concept() - Code Metrics - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

atramhasis.views.crud.AtramhasisCrud.get_concept() A
last analyzed 2024-08-12 13:03 UTC

↳ Parent: atramhasis.views.crud

Complexity

Conditions

Size

Total Lines	21
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	21
rs	9.7
c	0
b	0
f	0
cc	4
nop	1

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

import colander
import transaction
from pyramid.httpexceptions import HTTPMethodNotAllowed
from pyramid.httpexceptions import HTTPNoContent
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 import utils
from atramhasis.audit import audit
from atramhasis.cache import invalidate_scheme_cache
from atramhasis.data.models import IDGenerationStrategy
from atramhasis.errors import ConceptNotFoundException
from atramhasis.errors import ConceptSchemeNotFoundException
from atramhasis.errors import SkosRegistryNotFoundException
from atramhasis.errors import ValidationError
from atramhasis.json_processors import provider
from atramhasis.mappers import map_concept
from atramhasis.mappers import map_conceptscheme
from atramhasis.protected_resources import protected_operation
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, request):
        self.request = request
        self.skos_manager = self.request.data_managers['skos_manager']
        self.conceptscheme_manager = self.request.data_managers['conceptscheme_manager']
        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, validate_id_generation):
        from atramhasis.validators import (
            Concept as ConceptSchema,
            concept_schema_validator
        )

        concept_schema = ConceptSchema(
            validator=concept_schema_validator
        ).bind(
            request=self.request,
            provider=provider,
            validate_id_generation=validate_id_generation
        )
        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='skosprovider.conceptscheme', permission='view', request_method='GET')
    def get_conceptscheme(self):
        # is the same as pyramid_skosprovider but wrapped with the audit decorator
        return ProviderView(self.request).get_conceptscheme()

    @view_config(route_name='skosprovider.conceptscheme', permission='edit', request_method='PUT')
    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

    @audit
    @view_config(route_name='skosprovider.c', permission='view', request_method='GET')
    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='skosprovider.conceptscheme.cs', permission='edit', request_method='POST')
    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, validate_id_generation=True
        )
        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["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='skosprovider.c', permission='edit', request_method='PUT')
    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, validate_id_generation=False
        )
        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='skosprovider.c', permission='delete', request_method='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"])

    @view_config(
        route_name='atramhasis.providers',
        permission='add-provider',
        request_method='POST',
        openapi=True
    )
    def add_provider(self):
        db_provider = provider.create_provider(
            json_data=self.request.openapi_validated.body,
            session=self.request.db,
            skos_registry=self.request.skos_registry,
        )
        self.request.response.status_code = 201
        return utils.db_provider_to_skosprovider(db_provider)

    @view_config(
        route_name='atramhasis.provider',
        permission='edit-provider',
        request_method='PUT',
        openapi=True
    )
    def update_provider(self):
        try:
            db_provider = provider.update_provider(
                provider_id=self.request.matchdict["id"],
                json_data=self.request.openapi_validated.body,
                session=self.request.db,
            )
            return utils.db_provider_to_skosprovider(db_provider)
        except NoResultFound:
            db_provider = provider.create_provider(
                json_data=self.request.openapi_validated.body,
                session=self.request.db,
                skos_registry=self.request.skos_registry,
            )
            self.request.response.status_code = 201
            return utils.db_provider_to_skosprovider(db_provider)

    @view_config(
        route_name='atramhasis.provider',
        permission='delete-provider',
        request_method='DELETE',
        openapi=True
    )
    def delete_provider(self):
        provider.delete_provider(
            provider_id=self.request.matchdict["id"],
            session=self.request.db,
        )
        return HTTPNoContent()


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

OnroerendErfgoed / atramhasis

atramhasis.views.crud.AtramhasisCrud.get_concept() A last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

atramhasis.views.crud.AtramhasisCrud.get_concept() A
last analyzed 2024-08-12 13:03 UTC