atramhasis.views.crud.AtramhasisCrud.get_provider() - Code Metrics - Inspection of "Add Create, Update, Delete provider. (#744, PR:#81..." - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 591d1d...018914 )

unknown

created 2023-05-02 09:04 UTC

AtramhasisCrud.get_provider() A

↳ Parent: atramhasis.views.crud

Complexity

Conditions

Size

Total Lines	8
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	1
dl	0
loc	8
rs	10
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.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 mappers
from atramhasis import utils
from atramhasis.audit import audit
from atramhasis.cache import invalidate_scheme_cache
from atramhasis.data.datamanagers import ProviderDataManager
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.data.models import IDGenerationStrategy
from atramhasis.utils import db_provider_to_skosprovider
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='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, 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='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, 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='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"])

    @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):
        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)

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

OnroerendErfgoed / atramhasis

Push — develop ( 591d1d...018914 )

AtramhasisCrud.get_provider() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like