-
Notifications
You must be signed in to change notification settings - Fork 5
/
query.py
123 lines (97 loc) · 3.71 KB
/
query.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
# This file is a part of the AnyBlok / Pyramid project
#
# Copyright (C) 2018 Jean-Sebastien SUZANNE <jssuzanne@anybox.fr>
#
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file,You can
# obtain one at http://mozilla.org/MPL/2.0/.
from anyblok import Declarations
from sqlalchemy import or_, and_
from dateutil.parser import parse
def condition_filter(query, conditions, objects):
if 'or' in conditions:
query, compiled = sub_condition_filter(
query, conditions['or'], objects)
return query, or_(*compiled)
elif 'and' in conditions:
query, compiled = sub_condition_filter(
query, conditions['and'], objects)
return query, and_(*compiled)
elif 'not' in conditions:
query, compiled = sub_condition_filter(
query, [conditions['not']], objects)
return query, ~ compiled[0]
return condition_filter_leaf(query, conditions, objects)
def sub_condition_filter(query, conditions, objects):
compileds = []
for condition in conditions:
query, compiled = condition_filter(query, condition, objects)
compileds.append(compiled)
return query, compileds
def get_value_for(query, conditions, key_value, key_condition, key_adapter,
objects):
if key_value in conditions:
value = conditions[key_value]
if key_adapter in conditions:
value = adapt_value(conditions[key_adapter], value)
return query, value
if key_condition in conditions:
value = conditions[key_condition].split('.')
entry = objects[value[0]]
keys = value[1:]
if len(keys) > 1:
return query, get_value_for_relationship(query, entry, keys)
return query, getattr(entry, keys[0])
return query, None
def get_value_for_relationship(query, entry, keys):
res = getattr(entry, keys[0], None)
if len(keys) == 1:
return query, res
if hasattr(entry, '__registry_name__'):
Model = query.anyblok.get(entry.__registry_name__)
if entry is Model:
query = query.join(res)
return query, get_value_for_relationship(query, res, keys[1:])
def adapt_value(adapter, value):
if adapter == 'datetime':
return parse(value)
def condition_filter_leaf(query, conditions, objects): # noqa
query, left = get_value_for(
query, conditions, 'left_value', 'left_condition', 'left_adapter',
objects)
query, right = get_value_for(
query, conditions, 'right_value', 'right_condition', 'right_adapter',
objects)
operator = conditions.get('operator')
if operator == '==':
return query, left == right
if operator == '!=':
return query, left != right
if operator == 'in':
return query, left.in_(right)
if operator == 'not in':
return query, left.notin(right)
if operator == '<':
return query, left < right
if operator == '<=':
return query, left <= right
if operator == '>':
return query, left > right
if operator == '>=':
return query, left >= right
if operator == 'like':
return query, left.like(right)
if operator == 'not like':
return query, ~ left.like(right)
if operator == 'ilike':
return query, left.ilike(right)
if operator == 'not ilike':
return query, ~ left.ilike(right)
return query, None
@Declarations.register(Declarations.Core)
class Query:
def condition_filter(self, conditions, objects):
query, compiled = condition_filter(self, conditions, objects)
if compiled is None:
return self
return query.filter(compiled)