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

1# dialects/mssql/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# technically, all the dialect-specific datatypes that don't have any special 

12# behaviors would be private with names like _MSJson. However, we haven't been 

13# doing this for mysql.JSON or sqlite.JSON which both have JSON / JSONIndexType 

14# / JSONPathType in their json.py files, so keep consistent with that 

15# sub-convention for now. A future change can update them all to be 

16# package-private at once. 

17 

18 

19class JSON(sqltypes.JSON): 

20 """MSSQL JSON type. 

21 

22 MSSQL supports JSON-formatted data as of SQL Server 2016. 

23 

24 The :class:`_mssql.JSON` datatype at the DDL level will represent the 

25 datatype as ``NVARCHAR(max)``, but provides for JSON-level comparison 

26 functions as well as Python coercion behavior. 

27 

28 :class:`_mssql.JSON` is used automatically whenever the base 

29 :class:`_types.JSON` datatype is used against a SQL Server backend. 

30 

31 .. seealso:: 

32 

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

34 cross-platform JSON datatype. 

35 

36 The :class:`_mssql.JSON` type supports persistence of JSON values 

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

38 datatype, by adapting the operations to render the ``JSON_VALUE`` 

39 or ``JSON_QUERY`` functions at the database level. 

40 

41 The SQL Server :class:`_mssql.JSON` type necessarily makes use of the 

42 ``JSON_QUERY`` and ``JSON_VALUE`` functions when querying for elements 

43 of a JSON object. These two functions have a major restriction in that 

44 they are **mutually exclusive** based on the type of object to be returned. 

45 The ``JSON_QUERY`` function **only** returns a JSON dictionary or list, 

46 but not an individual string, numeric, or boolean element; the 

47 ``JSON_VALUE`` function **only** returns an individual string, numeric, 

48 or boolean element. **both functions either return NULL or raise 

49 an error if they are not used against the correct expected value**. 

50 

51 To handle this awkward requirement, indexed access rules are as follows: 

52 

53 1. When extracting a sub element from a JSON that is itself a JSON 

54 dictionary or list, the :meth:`_types.JSON.Comparator.as_json` accessor 

55 should be used:: 

56 

57 stmt = select(data_table.c.data["some key"].as_json()).where( 

58 data_table.c.data["some key"].as_json() == {"sub": "structure"} 

59 ) 

60 

61 2. When extracting a sub element from a JSON that is a plain boolean, 

62 string, integer, or float, use the appropriate method among 

63 :meth:`_types.JSON.Comparator.as_boolean`, 

64 :meth:`_types.JSON.Comparator.as_string`, 

65 :meth:`_types.JSON.Comparator.as_integer`, 

66 :meth:`_types.JSON.Comparator.as_float`:: 

67 

68 stmt = select(data_table.c.data["some key"].as_string()).where( 

69 data_table.c.data["some key"].as_string() == "some string" 

70 ) 

71 

72 .. versionadded:: 1.4 

73 

74 

75 """ 

76 

77 # note there was a result processor here that was looking for "number", 

78 # but none of the tests seem to exercise it. 

79 

80 

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

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

83# implemented for each dialect. 

84class _FormatTypeMixin: 

85 def _format_value(self, value): 

86 raise NotImplementedError() 

87 

88 def bind_processor(self, dialect): 

89 super_proc = self.string_bind_processor(dialect) 

90 

91 def process(value): 

92 value = self._format_value(value) 

93 if super_proc: 

94 value = super_proc(value) 

95 return value 

96 

97 return process 

98 

99 def literal_processor(self, dialect): 

100 super_proc = self.string_literal_processor(dialect) 

101 

102 def process(value): 

103 value = self._format_value(value) 

104 if super_proc: 

105 value = super_proc(value) 

106 return value 

107 

108 return process 

109 

110 

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

112 def _format_value(self, value): 

113 if isinstance(value, int): 

114 value = "$[%s]" % value 

115 else: 

116 value = '$."%s"' % value 

117 return value 

118 

119 

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

121 def _format_value(self, value): 

122 return "$%s" % ( 

123 "".join( 

124 [ 

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

126 for elem in value 

127 ] 

128 ) 

129 )