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

64.94

/flask_combo_jsonapi/data_layers/sorting/alchemy.py

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

from sqlalchemy import sql
from sqlalchemy.orm import aliased

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


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

SortAndJoins = Tuple[
    Sort,
    List[Join],
]

def create_sorts(model, sort_info, resource):
    """Apply sorts from sorts information to base query

    :param DeclarativeMeta model: the model of the node
    :param list sort_info: current node sort information
    :param Resource resource: the resource
    """
    return create_filters_or_sorts(model, sort_info, resource, Node)


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

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

        :param Model model: an sqlalchemy model
        :param dict sort_: sorts 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.sort_ = sort_
        self.resource = resource
        self.schema = schema

    @classmethod
    def create_sort(cls, marshmallow_field, model_column, order):
        """
        Create sqlalchemy sort
        :param marshmallow_field:
        :param model_column: column sqlalchemy
        :param str order: desc | asc (or custom)
        :return:
        """
        """
        Custom sqlachemy sorting logic can be created in a marshmallow field for any field
        You can override existing ('asc', 'desc') or create new - then follow this pattern:
        `_<custom_sort_name>_sql_sort_`. This method has to accept following params:
        * marshmallow_field - marshmallow field instance
        * model_column - sqlalchemy column instance
        """
        try:
            f = getattr(marshmallow_field, f'_{order}_sql_sort_')
        except AttributeError:
            pass
        else:
            return f(
                marshmallow_field=marshmallow_field,
                model_column=model_column,
            )
        return getattr(model_column, order)()

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

        field = self.sort_.get('field', '')
        if not hasattr(self.model, field) and SPLIT_REL not in field:
            raise InvalidSort("{} has no attribute {}".format(self.model.__name__, field))

        if SPLIT_REL in field:
            value = {
                'field': SPLIT_REL.join(field.split(SPLIT_REL)[1:]),
                'order': self.sort_['order']
            }
            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

        return self.create_sort(
            marshmallow_field=self.schema._declared_fields[self.name],
            model_column=self.column,
            order=self.sort_['order']
        ), []

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

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

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

        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 column(self):
        """Get the column object

        :param DeclarativeMeta model: the model
        :param str field: the field
        :return InstrumentedAttribute: the column to filter on
        """
        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 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 sortings according to filter querystring parameter"""
2	from typing import Any, List, Tuple	1✔
3
4	from sqlalchemy import sql	1✔
5	from sqlalchemy.orm import aliased	1✔
6
7	from flask_combo_jsonapi.data_layers.shared import create_filters_or_sorts	1✔
8	from flask_combo_jsonapi.exceptions import InvalidFilters, PluginMethodNotImplementedError, InvalidSort	1✔
9	from flask_combo_jsonapi.schema import get_relationships, get_model_field	1✔
10	from flask_combo_jsonapi.utils import SPLIT_REL	1✔
11
12
13	Sort = sql.elements.BinaryExpression	1✔
14	Join = List[Any]	1✔
15
16	SortAndJoins = Tuple[	1✔
17	Sort,
18	List[Join],
19	]
20
21	def create_sorts(model, sort_info, resource):	1✔
22	"""Apply sorts from sorts information to base query
23
24	:param DeclarativeMeta model: the model of the node
25	:param list sort_info: current node sort information
26	:param Resource resource: the resource
27	"""
28	return create_filters_or_sorts(model, sort_info, resource, Node)	1✔
29
30
31	class Node(object):	1✔
32	"""Helper to recursively create sorts with sqlalchemy according to sort querystring parameter"""
33
34	def __init__(self, model, sort_, resource, schema):	1✔
35	"""Initialize an instance of a filter node
36
37	:param Model model: an sqlalchemy model
38	:param dict sort_: sorts information of the current node and deeper nodes
39	:param Resource resource: the base resource to apply filters on
40	:param Schema schema: the serializer of the resource
41	"""
42	self.model = model	1✔
43	self.sort_ = sort_	1✔
44	self.resource = resource	1✔
45	self.schema = schema	1✔
46
47	@classmethod	1✔
48	def create_sort(cls, marshmallow_field, model_column, order):	1✔
49	"""
50	Create sqlalchemy sort
51	:param marshmallow_field:
52	:param model_column: column sqlalchemy
53	:param str order: desc \| asc (or custom)
54	:return:
55	"""
56	"""
57	Custom sqlachemy sorting logic can be created in a marshmallow field for any field
58	You can override existing ('asc', 'desc') or create new - then follow this pattern:
59	`_<custom_sort_name>_sql_sort_`. This method has to accept following params:
60	* marshmallow_field - marshmallow field instance
61	* model_column - sqlalchemy column instance
62	"""
63	try:	1✔
64	f = getattr(marshmallow_field, f'_{order}_sql_sort_')	1✔
65	except AttributeError:	1✔
66	pass	1✔
67	else:
68	return f(	×
69	marshmallow_field=marshmallow_field,
70	model_column=model_column,
71	)
72	return getattr(model_column, order)()	1✔
73
74	def resolve(self) -> SortAndJoins:	1✔
75	"""Create sort for a particular node of the sort tree"""
76	if hasattr(self.resource, 'plugins'):	1✔
77	for i_plugin in self.resource.plugins:	1✔
78	try:	×
79	res = i_plugin.before_data_layers_sorting_alchemy_nested_resolve(self)	×
80	if res is not None:	×
81	return res	×
82	except PluginMethodNotImplementedError:	×
83	pass	×
84
85	field = self.sort_.get('field', '')	1✔
86	if not hasattr(self.model, field) and SPLIT_REL not in field:	1✔
87	raise InvalidSort("{} has no attribute {}".format(self.model.__name__, field))	1✔
88
89	if SPLIT_REL in field:	1✔
90	value = {	×
91	'field': SPLIT_REL.join(field.split(SPLIT_REL)[1:]),
92	'order': self.sort_['order']
93	}
94	alias = aliased(self.related_model)	×
95	joins = [[alias, self.column]]	×
96	node = Node(alias, value, self.resource, self.related_schema)	×
97	filters, new_joins = node.resolve()	×
98	joins.extend(new_joins)	×
99	return filters, joins	×
100
101	return self.create_sort(	1✔
102	marshmallow_field=self.schema._declared_fields[self.name],
103	model_column=self.column,
104	order=self.sort_['order']
105	), []
106
107	@property	1✔
108	def name(self):	1✔
109	"""Return the name of the node or raise a BadRequest exception
110
111	:return str: the name of the sort to sort on
112	"""
113	name = self.sort_.get('field')	1✔
114
115	if name is None:	1✔
116	raise InvalidFilters("Can't find name of a sort")	×
117
118	if SPLIT_REL in name:	1✔
119	name = name.split(SPLIT_REL)[0]	×
120
121	if name not in self.schema._declared_fields:	1✔
122	raise InvalidFilters("{} has no attribute {}".format(self.schema.__name__, name))	×
123
124	return name	1✔
125
126	@property	1✔
127	def column(self):	1✔
128	"""Get the column object
129
130	:param DeclarativeMeta model: the model
131	:param str field: the field
132	:return InstrumentedAttribute: the column to filter on
133	"""
134	field = self.name	1✔
135
136	model_field = get_model_field(self.schema, field)	1✔
137
138	try:	1✔
139	return getattr(self.model, model_field)	1✔
140	except AttributeError:	×
141	raise InvalidFilters("{} has no attribute {}".format(self.model.__name__, model_field))	×
142
143	@property	1✔
144	def related_model(self):	1✔
145	"""Get the related model of a relationship field
146
147	:return DeclarativeMeta: the related model
148	"""
149	relationship_field = self.name	×
150
151	if relationship_field not in get_relationships(self.schema):	×
152	raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))	×
153
154	return getattr(self.model, get_model_field(self.schema, relationship_field)).property.mapper.class_	×
155
156	@property	1✔
157	def related_schema(self):	1✔
158	"""Get the related schema of a relationship field
159
160	:return Schema: the related schema
161	"""
162	relationship_field = self.name	×
163
164	if relationship_field not in get_relationships(self.schema):	×
165	raise InvalidFilters("{} has no relationship attribute {}".format(self.schema.__name__, relationship_field))	×
166
167	return self.schema._declared_fields[relationship_field].schema.__class__	×

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