Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/sqlite/json.py: 38%

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 

7from __future__ import annotations 

8 

9from typing import Any 

10from typing import TYPE_CHECKING 

11 

12from ... import types as sqltypes 

13from ...sql.sqltypes import _T_JSON 

14 

15if TYPE_CHECKING: 

16 from ...engine.interfaces import Dialect 

17 from ...sql.type_api import _BindProcessorType 

18 from ...sql.type_api import _LiteralProcessorType 

19 

20 

21class JSON(sqltypes.JSON[_T_JSON]): 

22 """SQLite JSON type. 

23 

24 SQLite supports JSON as of version 3.9 through its JSON1_ extension. Note 

25 that JSON1_ is a 

26 `loadable extension <https://www.sqlite.org/loadext.html>`_ and as such 

27 may not be available, or may require run-time loading. 

28 

29 :class:`_sqlite.JSON` is used automatically whenever the base 

30 :class:`_types.JSON` datatype is used against a SQLite backend. 

31 

32 .. seealso:: 

33 

34 :class:`_types.JSON` - main documentation for the generic 

35 cross-platform JSON datatype. 

36 

37 The :class:`_sqlite.JSON` type supports persistence of JSON values 

38 as well as the core index operations provided by :class:`_types.JSON` 

39 datatype, by adapting the operations to render the ``JSON_EXTRACT`` 

40 function wrapped in the ``JSON_QUOTE`` function at the database level. 

41 Extracted values are quoted in order to ensure that the results are 

42 always JSON string values. 

43 

44 

45 .. _JSON1: https://www.sqlite.org/json1.html 

46 

47 """ 

48 

49 

50# Note: these objects currently match exactly those of MySQL, however since 

51# these are not generalizable to all JSON implementations, remain separately 

52# implemented for each dialect. 

53class _FormatTypeMixin: 

54 def _format_value(self, value: Any) -> str: 

55 raise NotImplementedError() 

56 

57 def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]: 

58 super_proc = self.string_bind_processor(dialect) # type: ignore[attr-defined] # noqa: E501 

59 

60 def process(value: Any) -> Any: 

61 value = self._format_value(value) 

62 if super_proc: 

63 value = super_proc(value) 

64 return value 

65 

66 return process 

67 

68 def literal_processor( 

69 self, dialect: Dialect 

70 ) -> _LiteralProcessorType[Any]: 

71 super_proc = self.string_literal_processor(dialect) # type: ignore[attr-defined] # noqa: E501 

72 

73 def process(value: Any) -> str: 

74 value = self._format_value(value) 

75 if super_proc: 

76 value = super_proc(value) 

77 return value # type: ignore[no-any-return] 

78 

79 return process 

80 

81 

82class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType): 

83 def _format_value(self, value: Any) -> str: 

84 if isinstance(value, int): 

85 formatted_value = "$[%s]" % value 

86 else: 

87 formatted_value = '$."%s"' % value 

88 return formatted_value 

89 

90 

91class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType): 

92 def _format_value(self, value: Any) -> str: 

93 return "$%s" % ( 

94 "".join( 

95 [ 

96 "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem 

97 for elem in value 

98 ] 

99 ) 

100 )