codekingpro/portable-devtools

1# dialects/mysql/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
7# mypy: ignore-errors
8
9from ... import types as sqltypes
10
11
12class JSON(sqltypes.JSON):
13    """MySQL JSON type.
14
15    MySQL supports JSON as of version 5.7.
16    MariaDB supports JSON (as an alias for LONGTEXT) as of version 10.2.
17
18    :class:`_mysql.JSON` is used automatically whenever the base
19    :class:`_types.JSON` datatype is used against a MySQL or MariaDB backend.
20
21    .. seealso::
22
23        :class:`_types.JSON` - main documentation for the generic
24        cross-platform JSON datatype.
25
26    The :class:`.mysql.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 at the database level.
30
31    """
32
33    pass
34
35
36class _FormatTypeMixin:
37    def _format_value(self, value):
38        raise NotImplementedError()
39
40    def bind_processor(self, dialect):
41        super_proc = self.string_bind_processor(dialect)
42
43        def process(value):
44            value = self._format_value(value)
45            if super_proc:
46                value = super_proc(value)
47            return value
48
49        return process
50
51    def literal_processor(self, dialect):
52        super_proc = self.string_literal_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
63class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType):
64    def _format_value(self, value):
65        if isinstance(value, int):
66            value = "$[%s]" % value
67        else:
68            value = '$."%s"' % value
69        return value
70
71
72class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType):
73    def _format_value(self, value):
74        return "$%s" % (
75            "".join(
76                [
77                    "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem
78                    for elem in value
79                ]
80            )
81        )
82