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
137
138
139
140
141
|
# 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.
Custom serializers and deserializers are specified at the dialect level,
that is using :func:`.create_engine`. The reason for this is that when
using psycopg2, the DBAPI only allows serializers at the per-cursor
or per-connection level. E.g.::
engine = create_engine("postgresql://scott:tiger@localhost/test",
json_serializer=my_serialize_fn,
json_deserializer=my_deserialize_fn
)
When using the psycopg2 dialect, the json_deserializer is registered
against the database using ``psycopg2.extras.register_default_json``.
.. versionadded:: 0.9
"""
__visit_name__ = 'JSON'
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):
json_serializer = dialect._json_serializer or json.dumps
if util.py2k:
encoding = dialect.encoding
def process(value):
return json_serializer(value).encode(encoding)
else:
def process(value):
return json_serializer(value)
return process
def result_processor(self, dialect, coltype):
json_deserializer = dialect._json_deserializer or json.loads
if util.py2k:
encoding = dialect.encoding
def process(value):
return json_deserializer(value.decode(encoding))
else:
def process(value):
return json_deserializer(value)
return process
ischema_names['json'] = JSON
|
