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
|
# dialects/mssql/json.py
# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from ... import types as sqltypes
# technically, all the dialect-specific datatypes that don't have any special
# behaviors would be private with names like _MSJson. However, we haven't been
# doing this for mysql.JSON or sqlite.JSON which both have JSON / JSONIndexType
# / JSONPathType in their json.py files, so keep consistent with that
# sub-convention for now. A future change can update them all to be
# package-private at once.
class JSON(sqltypes.JSON):
"""MSSQL JSON type.
MSSQL supports JSON-formatted data as of SQL Server 2016.
The :class:`_mssql.JSON` datatype at the DDL level will represent the
datatype as ``NVARCHAR(max)``, but provides for JSON-level comparison
functions as well as Python coercion behavior.
:class:`_mssql.JSON` is used automatically whenever the base
:class:`_types.JSON` datatype is used against a SQL Server backend.
.. seealso::
:class:`_types.JSON` - main documentation for the generic
cross-platform JSON datatype.
The :class:`_mssql.JSON` type supports persistence of JSON values
as well as the core index operations provided by :class:`_types.JSON`
datatype, by adapting the operations to render the ``JSON_VALUE``
or ``JSON_QUERY`` functions at the database level.
The SQL Server :class:`_mssql.JSON` type necessarily makes use of the
``JSON_QUERY`` and ``JSON_VALUE`` functions when querying for elements
of a JSON object. These two functions have a major restriction in that
they are **mutually exclusive** based on the type of object to be returned.
The ``JSON_QUERY`` function **only** returns a JSON dictionary or list,
but not an individual string, numeric, or boolean element; the
``JSON_VALUE`` function **only** returns an individual string, numeric,
or boolean element. **both functions either return NULL or raise
an error if they are not used against the correct expected value**.
To handle this awkward requirement, indexed access rules are as follows:
1. When extracting a sub element from a JSON that is itself a JSON
dictionary or list, the :meth:`_types.JSON.Comparator.as_json` accessor
should be used::
stmt = select(
data_table.c.data["some key"].as_json()
).where(
data_table.c.data["some key"].as_json() == {"sub": "structure"}
)
2. When extracting a sub element from a JSON that is a plain boolean,
string, integer, or float, use the appropriate method among
:meth:`_types.JSON.Comparator.as_boolean`,
:meth:`_types.JSON.Comparator.as_string`,
:meth:`_types.JSON.Comparator.as_integer`,
:meth:`_types.JSON.Comparator.as_float`::
stmt = select(
data_table.c.data["some key"].as_string()
).where(
data_table.c.data["some key"].as_string() == "some string"
)
.. versionadded:: 1.4
"""
# note there was a result processor here that was looking for "number",
# but none of the tests seem to exercise it.
# Note: these objects currently match exactly those of MySQL, however since
# these are not generalizable to all JSON implementations, remain separately
# implemented for each dialect.
class _FormatTypeMixin:
def _format_value(self, value):
raise NotImplementedError()
def bind_processor(self, dialect):
super_proc = self.string_bind_processor(dialect)
def process(value):
value = self._format_value(value)
if super_proc:
value = super_proc(value)
return value
return process
def literal_processor(self, dialect):
super_proc = self.string_literal_processor(dialect)
def process(value):
value = self._format_value(value)
if super_proc:
value = super_proc(value)
return value
return process
class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType):
def _format_value(self, value):
if isinstance(value, int):
value = "$[%s]" % value
else:
value = '$."%s"' % value
return value
class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType):
def _format_value(self, value):
return "$%s" % (
"".join(
[
"[%s]" % elem if isinstance(elem, int) else '."%s"' % elem
for elem in value
]
)
)
|
