forked from getredash/redash
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtypes.py
More file actions
96 lines (70 loc) · 2.57 KB
/
types.py
File metadata and controls
96 lines (70 loc) · 2.57 KB
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
import pytz
from sqlalchemy.types import TypeDecorator
from sqlalchemy.ext.indexable import index_property
from sqlalchemy.ext.mutable import Mutable
from redash.utils import json_dumps, json_loads
from redash.utils.configuration import ConfigurationContainer
from .base import db
class Configuration(TypeDecorator):
impl = db.Text
def process_bind_param(self, value, dialect):
return value.to_json()
def process_result_value(self, value, dialect):
return ConfigurationContainer.from_json(value)
# XXX replace PseudoJSON and MutableDict with real JSON field
class PseudoJSON(TypeDecorator):
impl = db.Text
def process_bind_param(self, value, dialect):
if value is None:
return value
return json_dumps(value)
def process_result_value(self, value, dialect):
if not value:
return value
return json_loads(value)
class MutableDict(Mutable, dict):
@classmethod
def coerce(cls, key, value):
"Convert plain dictionaries to MutableDict."
if not isinstance(value, MutableDict):
if isinstance(value, dict):
return MutableDict(value)
# this call will raise ValueError
return Mutable.coerce(key, value)
else:
return value
def __setitem__(self, key, value):
"Detect dictionary set events and emit change events."
dict.__setitem__(self, key, value)
self.changed()
def __delitem__(self, key):
"Detect dictionary del events and emit change events."
dict.__delitem__(self, key)
self.changed()
class MutableList(Mutable, list):
def append(self, value):
list.append(self, value)
self.changed()
def remove(self, value):
list.remove(self, value)
self.changed()
@classmethod
def coerce(cls, key, value):
if not isinstance(value, MutableList):
if isinstance(value, list):
return MutableList(value)
return Mutable.coerce(key, value)
else:
return value
class json_cast_property(index_property):
"""
A SQLAlchemy index property that is able to cast the
entity attribute as the specified cast type. Useful
for JSON and JSONB colums for easier querying/filtering.
"""
def __init__(self, cast_type, *args, **kwargs):
super(json_cast_property, self).__init__(*args, **kwargs)
self.cast_type = cast_type
def expr(self, model):
expr = super(json_cast_property, self).expr(model)
return expr.astext.cast(self.cast_type)