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

1# dialects/sqlite/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 """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 )