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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

46 statements  

1# dialects/oracle/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 __future__ import annotations 

10 

11from typing import Any 

12from typing import TYPE_CHECKING 

13from typing import TypeVar 

14 

15from ... import types as sqltypes 

16from ...sql.base import _NoArg 

17from ...sql.base import NO_ARG 

18from ...sql.sqltypes import _T_JSON 

19 

20if TYPE_CHECKING: 

21 from ...engine.interfaces import Dialect 

22 from ...sql.type_api import _BindProcessorType 

23 from ...sql.type_api import _LiteralProcessorType 

24 

25_T = TypeVar("_T", bound=Any) 

26 

27 

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

29 """Oracle JSON type. 

30 

31 .. versionadded:: 2.1 

32 

33 Oracle Database supports JSON storage and querying for character and BLOB 

34 datatypes in Oracle 12c, and supports a dedicated JSON data type as of 

35 Oracle 21c. SQLAlchemy supports both of these scenarios when using the 

36 oracledb DBAPI. This type is used implicitly whenever the base 

37 :class:`_types.JSON` datatype is used against an Oracle backend, or may be 

38 constructed directly for access to Oracle-specific parameters such as 

39 :paramref:`_oracle.JSON.use_blob`. 

40 

41 Index operations are adapted to render using the ``JSON_QUERY`` and 

42 ``JSON_VALUE`` functions at the database level. 

43 

44 **Platform Support** - When using Oracle Database versions prior to 21c, 

45 BLOB is used as the storage format. In 21c or later, the native JSON 

46 datatype is used. This can be overridden using the 

47 :paramref:`_oracle.JSON.use_blob` parameter. 

48 

49 **Serialization / Deserialization** - JSON serialization of bound 

50 parameters uses Python ``json.dumps()`` by default rather than oracledb's 

51 native serializer, in order to support the 

52 :paramref:`_sqltypes.JSON.none_as_null` feature. The default serializer 

53 does **not** accept Python ``Decimal`` objects; to use a custom serializer, 

54 pass :paramref:`_sa.create_engine.json_serializer` to 

55 :func:`_sa.create_engine`. 

56 

57 When using the native JSON datatype (21c+), deserialization uses oracledb's 

58 native deserializer by default, which is required for JSON values larger 

59 than 32767 bytes. However, this deserializer returns all numeric values as 

60 ``Decimal`` since Oracle Database stores JSON numbers using its internal 

61 NUMBER type. To receive standard Python numeric types, pass 

62 ``json_deserializer=json.loads`` via 

63 :paramref:`_sa.create_engine.json_deserializer`; note that this limits 

64 maximum JSON value size to 32767 bytes. When using BLOB storage, 

65 SQLAlchemy deserializes using ``json.loads()`` directly rather than the 

66 oracledb deserializer. 

67 

68 **CHECK Constraint with BLOB** - When using BLOB storage, either on Oracle 

69 Database versions prior to 21c or via the :paramref:`_oracle.JSON.use_blob` 

70 parameter, the oracledb driver documentation recommends adding a 

71 ``<colname> IS JSON`` check constraint to indicate to the driver that the 

72 column stores JSON data. This constraint is **not** automatically 

73 generated by :class:`_oracle.JSON` and is not required by SQLAlchemy's 

74 implementation in order to read JSON data from the column. If desired, it 

75 can be added explicitly using :class:`_schema.CheckConstraint`. 

76 

77 .. seealso:: 

78 

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

80 cross-platform JSON datatype. 

81 

82 """ 

83 

84 use_blob: bool | _NoArg 

85 

86 def __init__( 

87 self, none_as_null: bool = False, use_blob: bool | _NoArg = NO_ARG 

88 ): 

89 """Construct a :class:`_oracle.JSON` type. 

90 

91 :param none_as_null=False: if True, persist the value ``None`` as a SQL 

92 NULL value, not the JSON encoding of ``null``. See the notes at 

93 :paramref:`_sqltypes.JSON.none_as_null` for complete background on 

94 this option. 

95 

96 :param use_blob: A boolean parameter indicating if the type should be 

97 rendered in DDL using BLOB instead of JSON. Normally, JSON or BLOB 

98 is chosen automatically based on the version of Oracle in use 

99 (21c or greater for JSON). If the parameter is left at its default 

100 value of the ``NO_ARG`` constant, this automatic selection is used. 

101 However when ``True``, the BLOB datatype will be used unconditionally, 

102 and if ``False``, JSON will be used unconditionally (including on 

103 backends older than 21c, which will raise an error by the server. 

104 This may be used to assert that only JSON-supporting backends 

105 should be used). 

106 

107 """ 

108 

109 super().__init__(none_as_null=none_as_null) 

110 self.use_blob = use_blob 

111 

112 

113class _FormatTypeMixin: 

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

115 raise NotImplementedError() 

116 

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

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

119 

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

121 value = self._format_value(value) 

122 if super_proc: 

123 value = super_proc(value) 

124 return value 

125 

126 return process 

127 

128 def literal_processor( 

129 self, dialect: Dialect 

130 ) -> _LiteralProcessorType[Any]: 

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

132 

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

134 value = self._format_value(value) 

135 if super_proc: 

136 value = super_proc(value) 

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

138 

139 return process 

140 

141 

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

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

144 if isinstance(value, int): 

145 return f"$[{value}]" 

146 else: 

147 return f'$."{value}"' 

148 

149 

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

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

152 return "$%s" % ( 

153 "".join( 

154 f"[{elem}]" if isinstance(elem, int) else f'."{elem}"' 

155 for elem in value 

156 ) 

157 )