_is_integer() - Code Metrics - Inspection of "Merge pull request #30 from OnroerendErfgoed/DEV_0..." - OnroerendErfgoed/augeias - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7299df...ddf84c )

unknown

created 2017-10-23 15:59 UTC

_is_integer() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	6
rs	9.4285

import uuid

from pyramid.response import Response
from pyramid.httpexceptions import HTTPLengthRequired, HTTPBadRequest, HTTPNotFound
from pyramid.view import view_config

from augeias.stores.error import NotFoundException


@view_config(context=NotFoundException, renderer='json')
def failed_not_found(exc, request):
    request.response.status_int = 404
    return {'message': exc.value}


class ValidationFailure(Exception):
    def __init__(self, msg):
        self.msg = msg


@view_config(context=ValidationFailure, renderer='json')
def failed_validation(exc, request):
    request.response.status_int = 400
    return {'message': 'Failed validation: %s' % exc.msg}


class AugeiasView(object):

    def __init__(self, request):
        self.request = request

    @view_config(route_name='home', renderer='json')
    def my_view(self):
        return {'project': 'augeias'}

    @view_config(route_name='list_collections', permission='view')
    def list_collections(self):
        res = Response(content_type='application/json', status=200)
        res.json_body = [c for c in self.request.registry.collections]
        return res

    @view_config(route_name='update_object', permission='edit')
    def update_object(self):
        '''

        Update an object in the data store.
        The input object data may be:
        - an object stream (Content-Type: application/octet-stream),
        - or a store location (host url, collection_key, container_key and object_key) 
            within the same Augeias instance (Content-Type: application/json). 
        '''
        object_data = _get_object_data(self.request)
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        object_key = self.request.matchdict['object_key']
        if len(object_key) < 3:
            raise ValidationFailure('The object key must be 3 characters long')
        collection.object_store.update_object(container_key, object_key, object_data)
        res = Response(content_type='application/json', status=200)
        res.json_body = {
            'container_key': container_key,
            'object_key': object_key,
            'uri': collection.uri_generator.generate_object_uri(
                collection=collection.name,
                container=container_key,
                object=object_key)

        }
        return res

    @view_config(route_name='create_object_and_id', permission='edit')
    def create_object_and_id(self):
        '''create an object in the data store and generate an id'''
        object_data = _get_object_data(self.request)
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        object_key = str(uuid.uuid4())
        collection.object_store.update_object(container_key, object_key, object_data)
        res = Response(content_type='application/json', status=201)
        res.json_body = {
            'container_key': container_key,
            'object_key': object_key,
            'uri': collection.uri_generator.generate_object_uri(
                collection=collection.name,
                container=container_key,
                object=object_key)

        }
        return res

    @view_config(route_name='delete_object', permission='edit')
    def delete_object(self):
        '''delete an object from the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        object_key = self.request.matchdict['object_key']
        collection.object_store.delete_object(container_key, object_key)
        res = Response(content_type='application/json', status=200)
        res.json_body = {
            'container_key': container_key,
            'object_key': object_key,
            'uri': collection.uri_generator.generate_object_uri(
                collection=collection.name,
                container=container_key,
                object=object_key)
        }
        return res

    @view_config(route_name='get_object', permission='view')
    def get_object(self):
        '''retrieve an object from the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        object_key = self.request.matchdict['object_key']
        object_data = collection.object_store.get_object(container_key, object_key)
        content_type = collection.object_store.get_object_info(container_key, object_key)['mime']
        res = Response(content_type=content_type, status=200)
        res.body = object_data
        return res

    @view_config(route_name='get_object_info', permission='view')
    def get_object_info(self):
        '''retrieve object info (mimetype, size, time last modification) from the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']

        object_key = self.request.matchdict['object_key']
        res = Response(content_type='application/json', status=200)
        res.json_body = collection.object_store.get_object_info(container_key, object_key)
        return res

    @view_config(route_name='list_object_keys_for_container', permission='view')
    def list_object_keys_for_container(self):
        '''list all object keys for a container in the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        res = Response(content_type='application/json', status=200)
        res.json_body = collection.object_store.list_object_keys_for_container(container_key)
        return res

    @view_config(route_name='create_container', permission='edit')

    def create_container(self):
        '''create a new container in the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        collection.object_store.create_container(container_key)
        res = Response(content_type='application/json', status=200)
        res.json_body = {
            'container_key': container_key,
            'uri': collection.uri_generator.generate_container_uri(
                collection=collection.name,
                container=container_key)
        }
        return res

    @view_config(route_name='create_container_and_id', permission='edit')

    def create_container_and_id(self):
        '''create a new container in the data store and generate an id'''
        collection = _retrieve_collection(self.request)
        container_key = str(uuid.uuid4())
        collection.object_store.create_container(container_key)
        res = Response(content_type='application/json', status=201)
        res.json_body = {
            'container_key': container_key,
            'uri': collection.uri_generator.generate_container_uri(
                collection=collection.name,
                container=container_key)
        }
        return res

    @view_config(route_name='delete_container', permission='edit')
    def delete_container(self):
        '''delete a container in the data store'''
        collection = _retrieve_collection(self.request)
        container_key = self.request.matchdict['container_key']
        collection.object_store.delete_container(container_key)
        res = Response(content_type='application/json', status=200)
        res.json_body = {
            'container_key': container_key,
            'uri': collection.uri_generator.generate_container_uri(
                collection=collection.name,
                container=container_key)
        }
        return res


# HELPERS


def _is_integer(s):
    try:
        int(s)  # python 2.7 'auto-promotes' int to long if required
        return True
    except ValueError:
        return False


def _get_object_data(request):
    if request.content_type == 'application/json':
        return _get_object_data_from_json_body(request)
    else:
        return _get_object_data_from_stream(request)


def _get_object_data_from_stream(request):
    if 'Content-Length' not in request.headers or not _is_integer(request.headers['Content-Length']):
        raise HTTPLengthRequired
    content_length = int(request.headers['Content-Length'])
    object_data = request.body_file
    if content_length == 0:
        raise HTTPBadRequest('body is empty')
    return object_data


def _get_object_data_from_json_body(request):
    '''
    The json body contains the location of the input object. 
    The body must include be the host url, collection_key, container_key and object_key.
    '''
    json_data = _get_json_from_request(request)
    required_keys = ['host_url', 'collection_key', 'container_key', 'object_key']
    missing_keys = set(required_keys) - set(json_data.keys())
    if missing_keys:
        raise HTTPBadRequest('\n'.join('{} is Required.'.format(key) for key in missing_keys))
    if request.host_url != json_data['host_url']:
        raise ValidationFailure('Host must be equal to the current host url {}.'.format(request.host))
    collection = request.registry.collections.get(json_data['collection_key'])
    if not collection:
        raise HTTPBadRequest('Collection {} was not found'.format(json_data['collection_key']))
    try:
        object_data = collection.object_store.get_object(json_data['container_key'], json_data['object_key'])
    except NotFoundException:
        raise HTTPBadRequest('Container - object ({0} - {1}) combination was not found in Collection {2}'.format(
            json_data['container_key'], json_data['object_key'], json_data['collection_key']))
    return object_data


def _get_json_from_request(request):
    try:
        return request.json_body
    except AttributeError as e:
        raise HTTPBadRequest(detail="Request has no json body. \n%s" % e)  # pragma: no cover
    except ValueError as e:
        raise HTTPBadRequest(detail="Request has incorrect json body. \n%s" % e)


def _retrieve_collection(request):
    collection_name = request.matchdict['collection_key']
    if collection_name in request.registry.collections:
        collection = request.registry.collections[collection_name]
    else:
        raise HTTPNotFound('collection not found')
    return collection


1		import uuid
2
3		from pyramid.response import Response
4		from pyramid.httpexceptions import HTTPLengthRequired, HTTPBadRequest, HTTPNotFound
5		from pyramid.view import view_config
6
7		from augeias.stores.error import NotFoundException
8
9
10		@view_config(context=NotFoundException, renderer='json')
11		def failed_not_found(exc, request):
12		request.response.status_int = 404
13		return {'message': exc.value}
14
15
16		class ValidationFailure(Exception):
17		def __init__(self, msg):
18		self.msg = msg
19
20
21		@view_config(context=ValidationFailure, renderer='json')
22		def failed_validation(exc, request):
23		request.response.status_int = 400
24		return {'message': 'Failed validation: %s' % exc.msg}
25
26
27		class AugeiasView(object):
28
29		def __init__(self, request):
30		self.request = request
31
32		@view_config(route_name='home', renderer='json')
33		def my_view(self):
34		return {'project': 'augeias'}
35
36		@view_config(route_name='list_collections', permission='view')
37		def list_collections(self):
38		res = Response(content_type='application/json', status=200)
39		res.json_body = [c for c in self.request.registry.collections]
40		return res
41
42		@view_config(route_name='update_object', permission='edit')
43		def update_object(self):
44	View Code Duplication	'''
		0 ignored issues – show Duplication introduced 2017-08-21 14:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
45		Update an object in the data store.
46		The input object data may be:
47		- an object stream (Content-Type: application/octet-stream),
48		- or a store location (host url, collection_key, container_key and object_key)
49		within the same Augeias instance (Content-Type: application/json).
50		'''
51		object_data = _get_object_data(self.request)
52		collection = _retrieve_collection(self.request)
53		container_key = self.request.matchdict['container_key']
54		object_key = self.request.matchdict['object_key']
55		if len(object_key) < 3:
56		raise ValidationFailure('The object key must be 3 characters long')
57		collection.object_store.update_object(container_key, object_key, object_data)
58		res = Response(content_type='application/json', status=200)
59		res.json_body = {
60		'container_key': container_key,
61		'object_key': object_key,
62		'uri': collection.uri_generator.generate_object_uri(
63		collection=collection.name,
64		container=container_key,
65	View Code Duplication	object=object_key)
		0 ignored issues – show Duplication introduced 2017-08-21 14:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
66		}
67		return res
68
69		@view_config(route_name='create_object_and_id', permission='edit')
70		def create_object_and_id(self):
71		'''create an object in the data store and generate an id'''
72		object_data = _get_object_data(self.request)
73		collection = _retrieve_collection(self.request)
74		container_key = self.request.matchdict['container_key']
75		object_key = str(uuid.uuid4())
76		collection.object_store.update_object(container_key, object_key, object_data)
77		res = Response(content_type='application/json', status=201)
78		res.json_body = {
79		'container_key': container_key,
80		'object_key': object_key,
81		'uri': collection.uri_generator.generate_object_uri(
82		collection=collection.name,
83		container=container_key,
84	View Code Duplication	object=object_key)
		0 ignored issues – show Duplication introduced 2017-08-21 14:54 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
85		}
86		return res
87
88		@view_config(route_name='delete_object', permission='edit')
89		def delete_object(self):
90		'''delete an object from the data store'''
91		collection = _retrieve_collection(self.request)
92		container_key = self.request.matchdict['container_key']
93		object_key = self.request.matchdict['object_key']
94		collection.object_store.delete_object(container_key, object_key)
95		res = Response(content_type='application/json', status=200)
96		res.json_body = {
97		'container_key': container_key,
98		'object_key': object_key,
99		'uri': collection.uri_generator.generate_object_uri(
100		collection=collection.name,
101		container=container_key,
102		object=object_key)
103		}
104		return res
105
106		@view_config(route_name='get_object', permission='view')
107		def get_object(self):
108		'''retrieve an object from the data store'''
109		collection = _retrieve_collection(self.request)
110		container_key = self.request.matchdict['container_key']
111		object_key = self.request.matchdict['object_key']
112		object_data = collection.object_store.get_object(container_key, object_key)
113		content_type = collection.object_store.get_object_info(container_key, object_key)['mime']
114		res = Response(content_type=content_type, status=200)
115		res.body = object_data
116		return res
117
118		@view_config(route_name='get_object_info', permission='view')
119		def get_object_info(self):
120		'''retrieve object info (mimetype, size, time last modification) from the data store'''
121		collection = _retrieve_collection(self.request)
122	View Code Duplication	container_key = self.request.matchdict['container_key']
		0 ignored issues – show Duplication introduced 2017-08-17 09:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
123		object_key = self.request.matchdict['object_key']
124		res = Response(content_type='application/json', status=200)
125		res.json_body = collection.object_store.get_object_info(container_key, object_key)
126		return res
127
128		@view_config(route_name='list_object_keys_for_container', permission='view')
129		def list_object_keys_for_container(self):
130		'''list all object keys for a container in the data store'''
131		collection = _retrieve_collection(self.request)
132		container_key = self.request.matchdict['container_key']
133		res = Response(content_type='application/json', status=200)
134		res.json_body = collection.object_store.list_object_keys_for_container(container_key)
135		return res
136
137	View Code Duplication	@view_config(route_name='create_container', permission='edit')
		0 ignored issues – show Duplication introduced 2017-08-17 09:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
138		def create_container(self):
139		'''create a new container in the data store'''
140		collection = _retrieve_collection(self.request)
141		container_key = self.request.matchdict['container_key']
142		collection.object_store.create_container(container_key)
143		res = Response(content_type='application/json', status=200)
144		res.json_body = {
145		'container_key': container_key,
146		'uri': collection.uri_generator.generate_container_uri(
147		collection=collection.name,
148		container=container_key)
149		}
150		return res
151
152	View Code Duplication	@view_config(route_name='create_container_and_id', permission='edit')
		0 ignored issues – show Duplication introduced 2017-08-09 08:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
153		def create_container_and_id(self):
154		'''create a new container in the data store and generate an id'''
155		collection = _retrieve_collection(self.request)
156		container_key = str(uuid.uuid4())
157		collection.object_store.create_container(container_key)
158		res = Response(content_type='application/json', status=201)
159		res.json_body = {
160		'container_key': container_key,
161		'uri': collection.uri_generator.generate_container_uri(
162		collection=collection.name,
163		container=container_key)
164		}
165		return res
166
167		@view_config(route_name='delete_container', permission='edit')
168		def delete_container(self):
169		'''delete a container in the data store'''
170		collection = _retrieve_collection(self.request)
171		container_key = self.request.matchdict['container_key']
172		collection.object_store.delete_container(container_key)
173		res = Response(content_type='application/json', status=200)
174		res.json_body = {
175		'container_key': container_key,
176		'uri': collection.uri_generator.generate_container_uri(
177		collection=collection.name,
178		container=container_key)
179		}
180		return res
181
182
183		# HELPERS
184
185
186		def _is_integer(s):
187		try:
188		int(s) # python 2.7 'auto-promotes' int to long if required
189		return True
190		except ValueError:
191		return False
192
193
194		def _get_object_data(request):
195		if request.content_type == 'application/json':
196		return _get_object_data_from_json_body(request)
197		else:
198		return _get_object_data_from_stream(request)
199
200
201		def _get_object_data_from_stream(request):
202		if 'Content-Length' not in request.headers or not _is_integer(request.headers['Content-Length']):
203		raise HTTPLengthRequired
204		content_length = int(request.headers['Content-Length'])
205		object_data = request.body_file
206		if content_length == 0:
207		raise HTTPBadRequest('body is empty')
208		return object_data
209
210
211		def _get_object_data_from_json_body(request):
212		'''
213		The json body contains the location of the input object.
214		The body must include be the host url, collection_key, container_key and object_key.
215		'''
216		json_data = _get_json_from_request(request)
217		required_keys = ['host_url', 'collection_key', 'container_key', 'object_key']
218		missing_keys = set(required_keys) - set(json_data.keys())
219		if missing_keys:
220		raise HTTPBadRequest('\n'.join('{} is Required.'.format(key) for key in missing_keys))
221		if request.host_url != json_data['host_url']:
222		raise ValidationFailure('Host must be equal to the current host url {}.'.format(request.host))
223		collection = request.registry.collections.get(json_data['collection_key'])
224		if not collection:
225		raise HTTPBadRequest('Collection {} was not found'.format(json_data['collection_key']))
226		try:
227		object_data = collection.object_store.get_object(json_data['container_key'], json_data['object_key'])
228		except NotFoundException:
229		raise HTTPBadRequest('Container - object ({0} - {1}) combination was not found in Collection {2}'.format(
230		json_data['container_key'], json_data['object_key'], json_data['collection_key']))
231		return object_data
232
233
234		def _get_json_from_request(request):
235		try:
236		return request.json_body
237		except AttributeError as e:
238		raise HTTPBadRequest(detail="Request has no json body. \n%s" % e) # pragma: no cover
239		except ValueError as e:
240		raise HTTPBadRequest(detail="Request has incorrect json body. \n%s" % e)
241
242
243		def _retrieve_collection(request):
244		collection_name = request.matchdict['collection_key']
245		if collection_name in request.registry.collections:
246		collection = request.registry.collections[collection_name]
247		else:
248		raise HTTPNotFound('collection not found')
249		return collection
250

OnroerendErfgoed / augeias

Push — master ( 7299df...ddf84c )

_is_integer() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like