1# dialects/sqlite/json.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7from ... import types as sqltypes
8
9
10class JSON(sqltypes.JSON):
11 """SQLite JSON type.
12
13 SQLite supports JSON as of version 3.9 through its JSON1_ extension. Note
14 that JSON1_ is a
15 `loadable extension <https://www.sqlite.org/loadext.html>`_ and as such
16 may not be available, or may require run-time loading.
17
18 :class:`_sqlite.JSON` is used automatically whenever the base
19 :class:`_types.JSON` datatype is used against a SQLite backend.
20
21 .. seealso::
22
23 :class:`_types.JSON` - main documentation for the generic
24 cross-platform JSON datatype.
25
26 The :class:`_sqlite.JSON` type supports persistence of JSON values
27 as well as the core index operations provided by :class:`_types.JSON`
28 datatype, by adapting the operations to render the ``JSON_EXTRACT``
29 function wrapped in the ``JSON_QUOTE`` function at the database level.
30 Extracted values are quoted in order to ensure that the results are
31 always JSON string values.
32
33
34 .. versionadded:: 1.3
35
36
37 .. _JSON1: https://www.sqlite.org/json1.html
38
39 """
40
41
42# Note: these objects currently match exactly those of MySQL, however since
43# these are not generalizable to all JSON implementations, remain separately
44# implemented for each dialect.
45class _FormatTypeMixin(object):
46 def _format_value(self, value):
47 raise NotImplementedError()
48
49 def bind_processor(self, dialect):
50 super_proc = self.string_bind_processor(dialect)
51
52 def process(value):
53 value = self._format_value(value)
54 if super_proc:
55 value = super_proc(value)
56 return value
57
58 return process
59
60 def literal_processor(self, dialect):
61 super_proc = self.string_literal_processor(dialect)
62
63 def process(value):
64 value = self._format_value(value)
65 if super_proc:
66 value = super_proc(value)
67 return value
68
69 return process
70
71
72class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType):
73 def _format_value(self, value):
74 if isinstance(value, int):
75 value = "$[%s]" % value
76 else:
77 value = '$."%s"' % value
78 return value
79
80
81class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType):
82 def _format_value(self, value):
83 return "$%s" % (
84 "".join(
85 [
86 "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem
87 for elem in value
88 ]
89 )
90 )