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

Passed

Pull Request — develop (#812)

unknown

created 2023-03-22 15:31 UTC

AtramhasisCrud.add_provider() A

↳ Parent: atramhasis.views.crud

Complexity

Conditions

Size

Total Lines	19
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	16
nop	1
dl	0
loc	19
rs	9.6
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.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.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='edit',
        request_method='POST',
        openapi=True
    )
    def add_provider(self):
        json_data = self.request.openapi_validated.body
        db_provider = mappers.map_provider(json_data)
        if not db_provider.id:
            self.request.db.add(db_provider.conceptscheme)
            self.request.db.flush()
            db_provider.id = str(db_provider.conceptscheme.id)

        self.request.db.add(db_provider)
        self.request.db.flush()

        self.request.response.status = 201
        return db_provider_to_skosprovider(db_provider)

    @view_config(
        route_name='atramhasis.provider',
        permission='edit',
        request_method='PUT',
        openapi=True
    )
    def update_provider(self):
        manager = ProviderDataManager(self.request.db)
        db_provider = manager.get_provider_by_id(self.request.matchdict["id"])
        json_data = self.request.openapi_validated.body
        db_provider = mappers.map_provider(json_data, provider=db_provider)
        self.request.db.flush()
        return db_provider_to_skosprovider(db_provider)

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

OnroerendErfgoed / atramhasis

Pull Request — develop (#812)

AtramhasisCrud.add_provider() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like