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
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# postgresql/json.py
# Copyright (C) 2005-2013 the SQLAlchemy authors and contributors <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
from __future__ import absolute_import
import json
from .base import ischema_names
from ... import types as sqltypes
from ...sql.operators import custom_op
from ... import util
__all__ = ('JSON', )
class JSON(sqltypes.TypeEngine):
"""Represent the Postgresql JSON type.
The :class:`.JSON` type stores arbitrary JSON format data, e.g.::
data_table = Table('data_table', metadata,
Column('id', Integer, primary_key=True),
Column('data', JSON)
)
with engine.connect() as conn:
conn.execute(
data_table.insert(),
data = {"key1": "value1", "key2": "value2"}
)
:class:`.JSON` provides several operations:
* Index operations::
data_table.c.data['some key']
* Index operations returning text (required for text comparison or casting)::
data_table.c.data.astext['some key'] == 'some value'
* Path index operations::
data_table.c.data[('key_1', 'key_2', ..., 'key_n')]
* Path index operations returning text (required for text comparison or casting)::
data_table.c.data.astext[('key_1', 'key_2', ..., 'key_n')] == 'some value'
The :class:`.JSON` type, when used with the SQLAlchemy ORM, does not detect
in-place mutations to the structure. In order to detect these, the
:mod:`sqlalchemy.ext.mutable` extension must be used. This extension will
allow "in-place" changes to the datastructure to produce events which
will be detected by the unit of work. See the example at :class:`.HSTORE`
for a simple example involving a dictionary.
.. versionadded:: 0.9
"""
__visit_name__ = 'JSON'
def __init__(self, json_serializer=None, json_deserializer=None):
if json_serializer:
self.json_serializer = json_serializer
else:
self.json_serializer = json.dumps
if json_deserializer:
self.json_deserializer = json_deserializer
else:
self.json_deserializer = json.loads
class comparator_factory(sqltypes.Concatenable.Comparator):
"""Define comparison operations for :class:`.JSON`."""
class _astext(object):
def __init__(self, parent):
self.parent = parent
def __getitem__(self, other):
return self.parent.expr._get_item(other, True)
def _get_item(self, other, astext):
if hasattr(other, '__iter__') and \
not isinstance(other, util.string_types):
op = "#>"
other = "{%s}" % (", ".join(util.text_type(elem) for elem in other))
else:
op = "->"
if astext:
op += ">"
# ops: ->, ->>, #>, #>>
return self.expr.op(op, precedence=5)(other)
def __getitem__(self, other):
"""Get the value at a given key."""
return self._get_item(other, False)
@property
def astext(self):
return self._astext(self)
def _adapt_expression(self, op, other_comparator):
if isinstance(op, custom_op):
if op.opstring == '->':
return op, sqltypes.Text
return sqltypes.Concatenable.Comparator.\
_adapt_expression(self, op, other_comparator)
def bind_processor(self, dialect):
if util.py2k:
encoding = dialect.encoding
def process(value):
return self.json_serializer(value).encode(encoding)
else:
def process(value):
return self.json_serializer(value)
return process
def result_processor(self, dialect, coltype):
if util.py2k:
encoding = dialect.encoding
def process(value):
return self.json_deserializer(value.decode(encoding))
else:
def process(value):
return self.json_deserializer(value)
return process
ischema_names['json'] = JSON
|
