3011

pending completion

Build # 3011

Build Type

Pull #807

travis-ci-com

Committed by

web-flow

Commit Message

Merge 37d05c795 into e6d5c7f9c

Pull Request Pull Request #807: Bugfix/739 path param regex

Run Details

23 of 23 new or added lines in 3 files covered. (100.0%)

1482 of 1534 relevant lines covered (96.61%)

4.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.03

/atramhasis/views/crud.py

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

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

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

LOG = logging.getLogger(__name__)


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

    def __init__(self, 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"])

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

OnroerendErfgoed / atramhasis / 3011

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous