codekingpro/portable-devtools

1# dialects/mysql/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
7from __future__ import annotations
8
9from typing import Any
10from typing import TYPE_CHECKING
11
12from ... import types as sqltypes
13
14if TYPE_CHECKING:
15    from ...engine.interfaces import Dialect
16    from ...sql.type_api import _BindProcessorType
17    from ...sql.type_api import _LiteralProcessorType
18
19
20class JSON(sqltypes.JSON):
21    """MySQL JSON type.
22
23    MySQL supports JSON as of version 5.7.
24    MariaDB supports JSON (as an alias for LONGTEXT) as of version 10.2.
25
26    :class:`_mysql.JSON` is used automatically whenever the base
27    :class:`_types.JSON` datatype is used against a MySQL or MariaDB backend.
28
29    .. seealso::
30
31        :class:`_types.JSON` - main documentation for the generic
32        cross-platform JSON datatype.
33
34    The :class:`.mysql.JSON` type supports persistence of JSON values
35    as well as the core index operations provided by :class:`_types.JSON`
36    datatype, by adapting the operations to render the ``JSON_EXTRACT``
37    function at the database level.
38
39    """
40
41    pass
42
43
44class _FormatTypeMixin:
45    def _format_value(self, value: Any) -> str:
46        raise NotImplementedError()
47
48    def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]:
49        super_proc = self.string_bind_processor(dialect)  # type: ignore[attr-defined]  # noqa: E501
50
51        def process(value: Any) -> Any:
52            value = self._format_value(value)
53            if super_proc:
54                value = super_proc(value)
55            return value
56
57        return process
58
59    def literal_processor(
60        self, dialect: Dialect
61    ) -> _LiteralProcessorType[Any]:
62        super_proc = self.string_literal_processor(dialect)  # type: ignore[attr-defined]  # noqa: E501
63
64        def process(value: Any) -> str:
65            value = self._format_value(value)
66            if super_proc:
67                value = super_proc(value)
68            return value  # type: ignore[no-any-return]
69
70        return process
71
72
73class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType):
74    def _format_value(self, value: Any) -> str:
75        if isinstance(value, int):
76            formatted_value = "$[%s]" % value
77        else:
78            formatted_value = '$."%s"' % value
79        return formatted_value
80
81
82class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType):
83    def _format_value(self, value: Any) -> str:
84        return "$%s" % (
85            "".join(
86                [
87                    "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem
88                    for elem in value
89                ]
90            )
91        )
92