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

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

↳ Parent: atramhasis.views.crud

Complexity

Conditions

Size

Total Lines	60
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	38
nop	1
dl	0
loc	60
rs	7.568
c	0
b	0
f	0

How to fix Long Method

"""
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.add_concept() B last analyzed 2024-08-12 13:03 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

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