1# dialects/sqlite/json.py 
    2# Copyright (C) 2005-2025 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    .. _JSON1: https://www.sqlite.org/json1.html 
    37 
    38    """ 
    39 
    40 
    41# Note: these objects currently match exactly those of MySQL, however since 
    42# these are not generalizable to all JSON implementations, remain separately 
    43# implemented for each dialect. 
    44class _FormatTypeMixin: 
    45    def _format_value(self, value): 
    46        raise NotImplementedError() 
    47 
    48    def bind_processor(self, dialect): 
    49        super_proc = self.string_bind_processor(dialect) 
    50 
    51        def process(value): 
    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(self, dialect): 
    60        super_proc = self.string_literal_processor(dialect) 
    61 
    62        def process(value): 
    63            value = self._format_value(value) 
    64            if super_proc: 
    65                value = super_proc(value) 
    66            return value 
    67 
    68        return process 
    69 
    70 
    71class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType): 
    72    def _format_value(self, value): 
    73        if isinstance(value, int): 
    74            value = "$[%s]" % value 
    75        else: 
    76            value = '$."%s"' % value 
    77        return value 
    78 
    79 
    80class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType): 
    81    def _format_value(self, value): 
    82        return "$%s" % ( 
    83            "".join( 
    84                [ 
    85                    "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem 
    86                    for elem in value 
    87                ] 
    88            ) 
    89        )