codekingpro/portable-devtools

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