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

91.06

/flask_combo_jsonapi/data_layers/filtering/alchemy.py

"""Helper to create sqlalchemy filters according to filter querystring parameter"""
from typing import Any, List, Tuple

from marshmallow_jsonapi.fields import Relationship
from sqlalchemy import and_, or_, not_, sql
from sqlalchemy.orm import aliased

from flask_combo_jsonapi.data_layers.shared import deserialize_field, create_filters_or_sorts
from flask_combo_jsonapi.exceptions import InvalidFilters, PluginMethodNotImplementedError
from flask_combo_jsonapi.schema import get_relationships, get_model_field
from flask_combo_jsonapi.utils import SPLIT_REL

Filter = sql.elements.BinaryExpression
Join = List[Any]

FilterAndJoins = Tuple[
    Filter,
    List[Join],
]


def create_filters(model, filter_info, resource):
    """Apply filters from filters information to base query

    :param DeclarativeMeta model: the model of the node
    :param dict filter_info: current node filter information
    :param Resource resource: the resource
    """
    return create_filters_or_sorts(model, filter_info, resource, Node)


class Node(object):
    """Helper to recursively create filters with sqlalchemy according to filter querystring parameter"""

    def __init__(self, model, filter_, resource, schema):
        """Initialize an instance of a filter node

        :param Model model: an sqlalchemy model
        :param dict filter_: filters information of the current node and deeper nodes
        :param Resource resource: the base resource to apply filters on
        :param Schema schema: the serializer of the resource
        """
        self.model = model
        self.filter_ = filter_
        self.resource = resource
        self.schema = schema

    def create_filter(self, marshmallow_field, model_column, operator, value):
        """
        Create sqlalchemy filter
        :param marshmallow_field:
        :param model_column: column sqlalchemy
        :param operator:
        :param value:
        :return:
        """
        """
        Custom sqlachemy filtering logic can be created in a marshmallow field for any operator
        To implement a new filtering logic (override existing or create a new one)
        create a method inside a field following this pattern:
        `_<your_op_name>_sql_filter_`. Each filtering method has to accept these params: 
        * marshmallow_field - marshmallow field instance
        * model_column - sqlalchemy column instance
        * value - filtering value
        * operator - your operator, for example: "eq", "in", "ilike_str_array", ...
        """
        try:
            f = getattr(marshmallow_field, f'_{operator}_sql_filter_')
        except AttributeError:
            pass
        else:
            return f(
                marshmallow_field=marshmallow_field,
                model_column=model_column,
                value=value,
                operator=operator,
            )
        # Here we have to deserialize and validate fields, that are used in filtering,
        # so the Enum fields are loaded correctly
        value = deserialize_field(marshmallow_field, value)
        return getattr(model_column, self.operator)(value)

    def resolve(self) -> FilterAndJoins:
        """Create filter for a particular node of the filter tree"""
        if self.resource and hasattr(self.resource, 'plugins'):
            for i_plugin in self.resource.plugins:
                try:
                    res = i_plugin.before_data_layers_filtering_alchemy_nested_resolve(self)
                    if res is not None:
                        return res
                except PluginMethodNotImplementedError:
                    pass

        if all(map(
                lambda op: op not in self.filter_,
                ('or', 'and', 'not'),
        )):
            value = self.value

            if isinstance(value, dict):
                return self._relationship_filtering(value)

            if SPLIT_REL in self.filter_.get('name', ''):
                value = {
                    'name': SPLIT_REL.join(self.filter_['name'].split(SPLIT_REL)[1:]),
                    'op': self.filter_['op'],
                    'val': value,
                }
                return self._relationship_filtering(value)

            marshmallow_field = self.schema._declared_fields[self.name]
            if isinstance(marshmallow_field, Relationship):
                value = {
                    'name': marshmallow_field.id_field,
                    'op': self.filter_['op'],
                    'val': value,
                }
                return self._relationship_filtering(value)

            return self.create_filter(
                marshmallow_field=marshmallow_field,
                model_column=self.column,
                operator=self.filter_['op'],
                value=value,
            ), []

        if 'or' in self.filter_:
            return self._create_filters(type_filter='or')
        if 'and' in self.filter_:
            return self._create_filters(type_filter='and')
        if 'not' in self.filter_:
            filter, joins = Node(self.model, self.filter_['not'], self.resource, self.schema).resolve()
            return not_(filter), joins

    def _relationship_filtering(self, value):
        alias = aliased(self.related_model)
        joins = [[alias, self.column]]
        node = Node(alias, value, self.resource, self.related_schema)
        filters, new_joins = node.resolve()
        joins.extend(new_joins)
        return filters, joins

    def _create_filters(self, type_filter: str) -> FilterAndJoins:
        """
        Создаём  фильтр or или and
        :param type_filter: 'or' или 'and'
        :return:
        """
        nodes = [Node(self.model, filter, self.resource, self.schema).resolve() for filter in self.filter_[type_filter]]
        joins = []
        for i_node in nodes:
            joins.extend(i_node[1])
        op = and_ if type_filter == 'and' else or_
        return op(*[i_node[0] for i_node in nodes]), joins

    @property
    def name(self):
        """Return the name of the node or raise a BadRequest exception

        :return str: the name of the field to filter on
        """
        name = self.filter_.get('name')

        if name is None:
            raise InvalidFilters("Can't find name of a filter")

        if SPLIT_REL in name:
            name = name.split(SPLIT_REL)[0]

        if name not in self.schema._declared_fields:
            raise InvalidFilters("{} has no attribute {}".format(self.schema.__name__, name))

        return name

    @property
    def op(self):
        """Return the operator of the node

        :return str: the operator to use in the filter
        """
        try:
            return self.filter_['op']
        except KeyError:
            raise InvalidFilters("Can't find op of a filter")

    @property
    def column(self):
        """Get the column object
        """
        field = self.name

        model_field = get_model_field(self.schema, field)

        try:
            return getattr(self.model, model_field)
        except AttributeError:
            raise InvalidFilters("{} has no attribute {}".format(self.model.__name__, model_field))

    @property
    def operator(self):
        """Get the function operator from his name

        :return callable: a callable to make operation on a column
        """
        operators = (self.op, self.op + '_', '__' + self.op + '__')

        for op in operators:
            if hasattr(self.column, op):
                return op

        raise InvalidFilters("{} has no operator {}".format(self.column.key, self.op))

    @property
    def value(self):
        """Get the value to filter on

        :return: the value to filter on
        """
        if self.filter_.get('field') is not None:
            try:
                result = getattr(self.model, self.filter_['field'])
            except AttributeError:
                raise InvalidFilters("{} has no attribute {}".format(self.model.__name__, self.filter_['field']))
            else:
                return result
        else:
            if 'val' not in self.filter_:
                raise InvalidFilters("Can't find value or field in a filter")

            return self.filter_['val']

    @property
    def related_model(self):
        """Get the related model of a relationship field

        :return DeclarativeMeta: the related model
        """
        relationship_field = self.name

        if relationship_field not in get_relationships(self.schema):
            raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))

        return getattr(self.model, get_model_field(self.schema, relationship_field)).property.mapper.class_

    @property
    def related_schema(self):
        """Get the related schema of a relationship field

        :return Schema: the related schema
        """
        relationship_field = self.name

        if relationship_field not in get_relationships(self.schema):
            raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))

        return self.schema._declared_fields[relationship_field].schema.__class__

1	"""Helper to create sqlalchemy filters according to filter querystring parameter"""
2	from typing import Any, List, Tuple	1✔
3
4	from marshmallow_jsonapi.fields import Relationship	1✔
5	from sqlalchemy import and_, or_, not_, sql	1✔
6	from sqlalchemy.orm import aliased	1✔
7
8	from flask_combo_jsonapi.data_layers.shared import deserialize_field, create_filters_or_sorts	1✔
9	from flask_combo_jsonapi.exceptions import InvalidFilters, PluginMethodNotImplementedError	1✔
10	from flask_combo_jsonapi.schema import get_relationships, get_model_field	1✔
11	from flask_combo_jsonapi.utils import SPLIT_REL	1✔
12
13	Filter = sql.elements.BinaryExpression	1✔
14	Join = List[Any]	1✔
15
16	FilterAndJoins = Tuple[	1✔
17	Filter,
18	List[Join],
19	]
20
21
22	def create_filters(model, filter_info, resource):	1✔
23	"""Apply filters from filters information to base query
24
25	:param DeclarativeMeta model: the model of the node
26	:param dict filter_info: current node filter information
27	:param Resource resource: the resource
28	"""
29	return create_filters_or_sorts(model, filter_info, resource, Node)	1✔
30
31
32	class Node(object):	1✔
33	"""Helper to recursively create filters with sqlalchemy according to filter querystring parameter"""
34
35	def __init__(self, model, filter_, resource, schema):	1✔
36	"""Initialize an instance of a filter node
37
38	:param Model model: an sqlalchemy model
39	:param dict filter_: filters information of the current node and deeper nodes
40	:param Resource resource: the base resource to apply filters on
41	:param Schema schema: the serializer of the resource
42	"""
43	self.model = model	1✔
44	self.filter_ = filter_	1✔
45	self.resource = resource	1✔
46	self.schema = schema	1✔
47
48	def create_filter(self, marshmallow_field, model_column, operator, value):	1✔
49	"""
50	Create sqlalchemy filter
51	:param marshmallow_field:
52	:param model_column: column sqlalchemy
53	:param operator:
54	:param value:
55	:return:
56	"""
57	"""
58	Custom sqlachemy filtering logic can be created in a marshmallow field for any operator
59	To implement a new filtering logic (override existing or create a new one)
60	create a method inside a field following this pattern:
61	`_<your_op_name>_sql_filter_`. Each filtering method has to accept these params:
62	* marshmallow_field - marshmallow field instance
63	* model_column - sqlalchemy column instance
64	* value - filtering value
65	* operator - your operator, for example: "eq", "in", "ilike_str_array", ...
66	"""
67	try:	1✔
68	f = getattr(marshmallow_field, f'_{operator}_sql_filter_')	1✔
69	except AttributeError:	1✔
70	pass	1✔
71	else:
72	return f(	×
73	marshmallow_field=marshmallow_field,
74	model_column=model_column,
75	value=value,
76	operator=operator,
77	)
78	# Here we have to deserialize and validate fields, that are used in filtering,
79	# so the Enum fields are loaded correctly
80	value = deserialize_field(marshmallow_field, value)	1✔
81	return getattr(model_column, self.operator)(value)	1✔
82
83	def resolve(self) -> FilterAndJoins:	1✔
84	"""Create filter for a particular node of the filter tree"""
85	if self.resource and hasattr(self.resource, 'plugins'):	1✔
86	for i_plugin in self.resource.plugins:	1✔
87	try:	×
88	res = i_plugin.before_data_layers_filtering_alchemy_nested_resolve(self)	×
89	if res is not None:	×
90	return res	×
91	except PluginMethodNotImplementedError:	×
92	pass	×
93
94	if all(map(	1✔
95	lambda op: op not in self.filter_,
96	('or', 'and', 'not'),
97	)):
98	value = self.value	1✔
99
100	if isinstance(value, dict):	1✔
101	return self._relationship_filtering(value)	1✔
102
103	if SPLIT_REL in self.filter_.get('name', ''):	1✔
104	value = {	1✔
105	'name': SPLIT_REL.join(self.filter_['name'].split(SPLIT_REL)[1:]),
106	'op': self.filter_['op'],
107	'val': value,
108	}
109	return self._relationship_filtering(value)	1✔
110
111	marshmallow_field = self.schema._declared_fields[self.name]	1✔
112	if isinstance(marshmallow_field, Relationship):	1✔
113	value = {	1✔
114	'name': marshmallow_field.id_field,
115	'op': self.filter_['op'],
116	'val': value,
117	}
118	return self._relationship_filtering(value)	1✔
119
120	return self.create_filter(	1✔
121	marshmallow_field=marshmallow_field,
122	model_column=self.column,
123	operator=self.filter_['op'],
124	value=value,
125	), []
126
127	if 'or' in self.filter_:	1✔
128	return self._create_filters(type_filter='or')	1✔
129	if 'and' in self.filter_:	1✔
130	return self._create_filters(type_filter='and')	1✔
131	if 'not' in self.filter_:	1✔
132	filter, joins = Node(self.model, self.filter_['not'], self.resource, self.schema).resolve()	1✔
133	return not_(filter), joins	×
134
135	def _relationship_filtering(self, value):	1✔
136	alias = aliased(self.related_model)	1✔
137	joins = [[alias, self.column]]	1✔
138	node = Node(alias, value, self.resource, self.related_schema)	1✔
139	filters, new_joins = node.resolve()	1✔
140	joins.extend(new_joins)	1✔
141	return filters, joins	1✔
142
143	def _create_filters(self, type_filter: str) -> FilterAndJoins:	1✔
144	"""
145	Создаём фильтр or или and
146	:param type_filter: 'or' или 'and'
147	:return:
148	"""
149	nodes = [Node(self.model, filter, self.resource, self.schema).resolve() for filter in self.filter_[type_filter]]	1✔
150	joins = []	1✔
151	for i_node in nodes:	1✔
152	joins.extend(i_node[1])	1✔
153	op = and_ if type_filter == 'and' else or_	1✔
154	return op(*[i_node[0] for i_node in nodes]), joins	1✔
155
156	@property	1✔
157	def name(self):	1✔
158	"""Return the name of the node or raise a BadRequest exception
159
160	:return str: the name of the field to filter on
161	"""
162	name = self.filter_.get('name')	1✔
163
164	if name is None:	1✔
165	raise InvalidFilters("Can't find name of a filter")	1✔
166
167	if SPLIT_REL in name:	1✔
168	name = name.split(SPLIT_REL)[0]	1✔
169
170	if name not in self.schema._declared_fields:	1✔
171	raise InvalidFilters("{} has no attribute {}".format(self.schema.__name__, name))	1✔
172
173	return name	1✔
174
175	@property	1✔
176	def op(self):	1✔
177	"""Return the operator of the node
178
179	:return str: the operator to use in the filter
180	"""
181	try:	1✔
182	return self.filter_['op']	1✔
183	except KeyError:	×
184	raise InvalidFilters("Can't find op of a filter")	×
185
186	@property	1✔
187	def column(self):	1✔
188	"""Get the column object
189	"""
190	field = self.name	1✔
191
192	model_field = get_model_field(self.schema, field)	1✔
193
194	try:	1✔
195	return getattr(self.model, model_field)	1✔
196	except AttributeError:	1✔
197	raise InvalidFilters("{} has no attribute {}".format(self.model.__name__, model_field))	1✔
198
199	@property	1✔
200	def operator(self):	1✔
201	"""Get the function operator from his name
202
203	:return callable: a callable to make operation on a column
204	"""
205	operators = (self.op, self.op + '_', '__' + self.op + '__')	1✔
206
207	for op in operators:	1✔
208	if hasattr(self.column, op):	1✔
209	return op	1✔
210
211	raise InvalidFilters("{} has no operator {}".format(self.column.key, self.op))	1✔
212
213	@property	1✔
214	def value(self):	1✔
215	"""Get the value to filter on
216
217	:return: the value to filter on
218	"""
219	if self.filter_.get('field') is not None:	1✔
220	try:	1✔
221	result = getattr(self.model, self.filter_['field'])	1✔
222	except AttributeError:	1✔
223	raise InvalidFilters("{} has no attribute {}".format(self.model.__name__, self.filter_['field']))	1✔
224	else:
225	return result	×
226	else:
227	if 'val' not in self.filter_:	1✔
228	raise InvalidFilters("Can't find value or field in a filter")	1✔
229
230	return self.filter_['val']	1✔
231
232	@property	1✔
233	def related_model(self):	1✔
234	"""Get the related model of a relationship field
235
236	:return DeclarativeMeta: the related model
237	"""
238	relationship_field = self.name	1✔
239
240	if relationship_field not in get_relationships(self.schema):	1✔
241	raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))	1✔
242
243	return getattr(self.model, get_model_field(self.schema, relationship_field)).property.mapper.class_	1✔
244
245	@property	1✔
246	def related_schema(self):	1✔
247	"""Get the related schema of a relationship field
248
249	:return Schema: the related schema
250	"""
251	relationship_field = self.name	1✔
252
253	if relationship_field not in get_relationships(self.schema):	1✔
254	raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))	1✔
255
256	return self.schema._declared_fields[relationship_field].schema.__class__	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