3771624463

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #68 from AdCombo/fix__init_subclass__for_multi_project

Run Details

60 of 60 new or added lines in 1 file covered. (100.0%)

1351 of 1620 relevant lines covered (83.4%)

0.83 hits per line

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

93.88

/flask_combo_jsonapi/querystring.py

"""Helper to deal with querystring parameters according to jsonapi specification"""

import simplejson as json

from flask import current_app

from flask_combo_jsonapi.exceptions import BadRequest, InvalidFilters, InvalidSort, InvalidField, InvalidInclude
from flask_combo_jsonapi.schema import get_model_field, get_relationships, get_schema_from_type
from flask_combo_jsonapi.utils import SPLIT_REL


class QueryStringManager(object):
    """Querystring parser according to jsonapi reference"""

    MANAGED_KEYS = (
        'filter',
        'page',
        'fields',
        'sort',
        'include',
        'q'
    )

    def __init__(self, querystring, schema):
        """Initialization instance

        :param dict querystring: query string dict from request.args
        """
        if not isinstance(querystring, dict):
            raise ValueError('QueryStringManager require a dict-like object querystring parameter')

        self.qs = querystring
        self.schema = schema

    def _get_key_values(self, name):
        """Return a dict containing key / values items for a given key, used for items like filters, page, etc.

        :param str name: name of the querystring parameter
        :return dict: a dict of key / values items
        """
        results = {}

        for key, value in self.qs.items():
            try:
                if not key.startswith(name):
                    continue

                key_start = key.index('[') + 1
                key_end = key.index(']')
                item_key = key[key_start:key_end]

                if ',' in value:
                    item_value = value.split(',')
                else:
                    item_value = value
                results.update({item_key: item_value})
            except Exception:
                raise BadRequest("Parse error", source={'parameter': key})

        return results

    def _simple_filters(self, dict_):
        return [{"name": key, "op": "eq", "val": value}
                for (key, value) in dict_.items()]

    @property
    def querystring(self):
        """Return original querystring but containing only managed keys

        :return dict: dict of managed querystring parameter
        """
        return {key: value for (key, value) in self.qs.items()
                if key.startswith(self.MANAGED_KEYS) or self._get_key_values('filter[')}

    @property
    def filters(self):
        """Return filters from query string.

        :return list: filter information
        """
        results = []
        filters = self.qs.get('filter')
        if filters is not None:
            try:
                results.extend(json.loads(filters))
            except (ValueError, TypeError):
                raise InvalidFilters("Parse error")
        if self._get_key_values('filter['):
            results.extend(self._simple_filters(self._get_key_values('filter[')))
        return results

    @property
    def pagination(self):
        """Return parameters page[size] and page[number) as a dict.
        If missing parmeter `size` then default parameter PAGE_SIZE is used.

        :return dict: a dict of pagination information

        Example with number strategy::

            >>> query_string = {'page[number]': '25', 'page[size]': '10'}
            >>> parsed_query.pagination
            {'number': 25, 'size': 10}
        """
        # check values type
        result = self._get_key_values('page')
        for key, value in result.items():
            if key not in ('number', 'size'):
                raise BadRequest("{} is not a valid parameter of pagination".format(key), source={'parameter': 'page'})
            try:
                result[key] = int(value)
            except ValueError:
                raise BadRequest("Parse error", source={'parameter': 'page[{}]'.format(key)})

        result.setdefault('size', current_app.config.get('PAGE_SIZE', 30))

        if current_app.config.get('ALLOW_DISABLE_PAGINATION', True) is False and result.get('size') == 0:
            raise BadRequest("You are not allowed to disable pagination", source={'parameter': 'page[size]'})

        if current_app.config.get('MAX_PAGE_SIZE') is not None and 'size' in result:
            if int(result['size']) > current_app.config['MAX_PAGE_SIZE']:
                raise BadRequest("Maximum page size is {}".format(current_app.config['MAX_PAGE_SIZE']),
                                 source={'parameter': 'page[size]'})

        return result

    @property
    def fields(self):
        """Return fields wanted by client.

        :return dict: a dict of sparse fieldsets information

        Return value will be a dict containing all fields by resource, for example::

            {
                "user": ['name', 'email'],
            }

        """
        result = self._get_key_values('fields')
        for key, value in result.items():
            if not isinstance(value, list):
                result[key] = [value]

        for key, value in result.items():
            schema = get_schema_from_type(key)
            for obj in value:
                if obj not in schema._declared_fields:
                    raise InvalidField("{} has no attribute {}".format(schema.__name__, obj))

        return result

    @property
    def sorting(self):
        """Return fields to sort by including sort name for SQLAlchemy and row
        sort parameter for other ORMs

        :return list: a list of sorting information

        Example of return value::

            [
                {'field': 'created_at', 'order': 'desc'},
            ]

        """
        if self.qs.get('sort'):
            sorting_results = []
            for sort_field in self.qs['sort'].split(','):
                field = sort_field.replace('-', '')
                if SPLIT_REL not in field:
                    if field not in self.schema._declared_fields:
                        raise InvalidSort("{} has no attribute {}".format(self.schema.__name__, field))
                    if field in get_relationships(self.schema):
                        raise InvalidSort("You can't sort on {} because it is a relationship field".format(field))
                    field = get_model_field(self.schema, field)
                order = 'desc' if sort_field.startswith('-') else 'asc'
                sorting_results.append({'field': field, 'order': order})
            return sorting_results

        return []

    @property
    def include(self):
        """Return fields to include

        :return list: a list of include information
        """
        include_param = self.qs.get('include', [])

        if current_app.config.get('MAX_INCLUDE_DEPTH') is not None:
            for include_path in include_param:
                if len(include_path.split(SPLIT_REL)) > current_app.config['MAX_INCLUDE_DEPTH']:
                    raise InvalidInclude("You can't use include through more than {} relationships"
                                         .format(current_app.config['MAX_INCLUDE_DEPTH']))

        return include_param.split(',') if include_param else []

1	"""Helper to deal with querystring parameters according to jsonapi specification"""
2
3	import simplejson as json	1✔
4
5	from flask import current_app	1✔
6
7	from flask_combo_jsonapi.exceptions import BadRequest, InvalidFilters, InvalidSort, InvalidField, InvalidInclude	1✔
8	from flask_combo_jsonapi.schema import get_model_field, get_relationships, get_schema_from_type	1✔
9	from flask_combo_jsonapi.utils import SPLIT_REL	1✔
10
11
12	class QueryStringManager(object):	1✔
13	"""Querystring parser according to jsonapi reference"""
14
15	MANAGED_KEYS = (	1✔
16	'filter',
17	'page',
18	'fields',
19	'sort',
20	'include',
21	'q'
22	)
23
24	def __init__(self, querystring, schema):	1✔
25	"""Initialization instance
26
27	:param dict querystring: query string dict from request.args
28	"""
29	if not isinstance(querystring, dict):	1✔
30	raise ValueError('QueryStringManager require a dict-like object querystring parameter')	1✔
31
32	self.qs = querystring	1✔
33	self.schema = schema	1✔
34
35	def _get_key_values(self, name):	1✔
36	"""Return a dict containing key / values items for a given key, used for items like filters, page, etc.
37
38	:param str name: name of the querystring parameter
39	:return dict: a dict of key / values items
40	"""
41	results = {}	1✔
42
43	for key, value in self.qs.items():	1✔
44	try:	1✔
45	if not key.startswith(name):	1✔
46	continue	1✔
47
48	key_start = key.index('[') + 1	1✔
49	key_end = key.index(']')	1✔
50	item_key = key[key_start:key_end]	1✔
51
52	if ',' in value:	1✔
53	item_value = value.split(',')	1✔
54	else:
55	item_value = value	1✔
56	results.update({item_key: item_value})	1✔
57	except Exception:	1✔
58	raise BadRequest("Parse error", source={'parameter': key})	1✔
59
60	return results	1✔
61
62	def _simple_filters(self, dict_):	1✔
63	return [{"name": key, "op": "eq", "val": value}	1✔
64	for (key, value) in dict_.items()]
65
66	@property	1✔
67	def querystring(self):	1✔
68	"""Return original querystring but containing only managed keys
69
70	:return dict: dict of managed querystring parameter
71	"""
72	return {key: value for (key, value) in self.qs.items()	1✔
73	if key.startswith(self.MANAGED_KEYS) or self._get_key_values('filter[')}
74
75	@property	1✔
76	def filters(self):	1✔
77	"""Return filters from query string.
78
79	:return list: filter information
80	"""
81	results = []	1✔
82	filters = self.qs.get('filter')	1✔
83	if filters is not None:	1✔
84	try:	1✔
85	results.extend(json.loads(filters))	1✔
86	except (ValueError, TypeError):	1✔
87	raise InvalidFilters("Parse error")	1✔
88	if self._get_key_values('filter['):	1✔
89	results.extend(self._simple_filters(self._get_key_values('filter[')))	1✔
90	return results	1✔
91
92	@property	1✔
93	def pagination(self):	1✔
94	"""Return parameters page[size] and page[number) as a dict.
95	If missing parmeter `size` then default parameter PAGE_SIZE is used.
96
97	:return dict: a dict of pagination information
98
99	Example with number strategy::
100
101	>>> query_string = {'page[number]': '25', 'page[size]': '10'}
102	>>> parsed_query.pagination
103	{'number': 25, 'size': 10}
104	"""
105	# check values type
106	result = self._get_key_values('page')	1✔
107	for key, value in result.items():	1✔
108	if key not in ('number', 'size'):	1✔
109	raise BadRequest("{} is not a valid parameter of pagination".format(key), source={'parameter': 'page'})	1✔
110	try:	1✔
111	result[key] = int(value)	1✔
112	except ValueError:	1✔
113	raise BadRequest("Parse error", source={'parameter': 'page[{}]'.format(key)})	1✔
114
115	result.setdefault('size', current_app.config.get('PAGE_SIZE', 30))	1✔
116
117	if current_app.config.get('ALLOW_DISABLE_PAGINATION', True) is False and result.get('size') == 0:	1✔
118	raise BadRequest("You are not allowed to disable pagination", source={'parameter': 'page[size]'})	×
119
120	if current_app.config.get('MAX_PAGE_SIZE') is not None and 'size' in result:	1✔
121	if int(result['size']) > current_app.config['MAX_PAGE_SIZE']:	×
122	raise BadRequest("Maximum page size is {}".format(current_app.config['MAX_PAGE_SIZE']),	×
123	source={'parameter': 'page[size]'})
124
125	return result	1✔
126
127	@property	1✔
128	def fields(self):	1✔
129	"""Return fields wanted by client.
130
131	:return dict: a dict of sparse fieldsets information
132
133	Return value will be a dict containing all fields by resource, for example::
134
135	{
136	"user": ['name', 'email'],
137	}
138
139	"""
140	result = self._get_key_values('fields')	1✔
141	for key, value in result.items():	1✔
142	if not isinstance(value, list):	1✔
143	result[key] = [value]	1✔
144
145	for key, value in result.items():	1✔
146	schema = get_schema_from_type(key)	1✔
147	for obj in value:	1✔
148	if obj not in schema._declared_fields:	1✔
149	raise InvalidField("{} has no attribute {}".format(schema.__name__, obj))	1✔
150
151	return result	1✔
152
153	@property	1✔
154	def sorting(self):	1✔
155	"""Return fields to sort by including sort name for SQLAlchemy and row
156	sort parameter for other ORMs
157
158	:return list: a list of sorting information
159
160	Example of return value::
161
162	[
163	{'field': 'created_at', 'order': 'desc'},
164	]
165
166	"""
167	if self.qs.get('sort'):	1✔
168	sorting_results = []	1✔
169	for sort_field in self.qs['sort'].split(','):	1✔
170	field = sort_field.replace('-', '')	1✔
171	if SPLIT_REL not in field:	1✔
172	if field not in self.schema._declared_fields:	1✔
173	raise InvalidSort("{} has no attribute {}".format(self.schema.__name__, field))	1✔
174	if field in get_relationships(self.schema):	1✔
175	raise InvalidSort("You can't sort on {} because it is a relationship field".format(field))	1✔
176	field = get_model_field(self.schema, field)	1✔
177	order = 'desc' if sort_field.startswith('-') else 'asc'	1✔
178	sorting_results.append({'field': field, 'order': order})	1✔
179	return sorting_results	1✔
180
181	return []	1✔
182
183	@property	1✔
184	def include(self):	1✔
185	"""Return fields to include
186
187	:return list: a list of include information
188	"""
189	include_param = self.qs.get('include', [])	1✔
190
191	if current_app.config.get('MAX_INCLUDE_DEPTH') is not None:	1✔
192	for include_path in include_param:	×
193	if len(include_path.split(SPLIT_REL)) > current_app.config['MAX_INCLUDE_DEPTH']:	×
194	raise InvalidInclude("You can't use include through more than {} relationships"	×
195	.format(current_app.config['MAX_INCLUDE_DEPTH']))
196
197	return include_param.split(',') if include_param else []	1✔

AdCombo / flask-combo-jsonapi / 3771624463

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