Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy_jsonfield/jsonfield.py: 62%

1# Copyright 2017-2022 Alexey Stepanov aka penguinolog 

2 

3# Licensed under the Apache License, Version 2.0 (the "License"); you may 

4# not use this file except in compliance with the License. You may obtain 

5# a copy of the License at 

6 

7# http://www.apache.org/licenses/LICENSE-2.0 

8 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

12# License for the specific language governing permissions and limitations 

13# under the License. 

14 

15"""JSONField implementation for SQLAlchemy.""" 

16 

17from __future__ import annotations 

18 

19# Standard Library 

20import json 

21import types 

22import typing 

23 

24# External Dependencies 

25import sqlalchemy.ext.mutable 

26import sqlalchemy.types 

27 

28if typing.TYPE_CHECKING: 

29 # External Dependencies 

30 from sqlalchemy.engine.default import DefaultDialect # noqa: F401 

31 from sqlalchemy.sql.type_api import TypeEngine # noqa: F401 

32 

33__all__ = ("JSONField", "mutable_json_field") 

34 

35 

36# noinspection PyAbstractClass 

37class JSONField(sqlalchemy.types.TypeDecorator): # type: ignore[misc] # pylint: disable=abstract-method 

38 """Represent an immutable structure as a json-encoded string or json. 

39 

40 Usage:: 

41 

42 JSONField(enforce_string=True|False, enforce_unicode=True|False) 

43 

44 """ 

45 

46 def process_literal_param(self, value: typing.Any, dialect: DefaultDialect) -> typing.Any: 

47 """Re-use of process_bind_param. 

48 

49 :return: encoded value if required 

50 :rtype: typing.Union[str, typing.Any] 

51 """ 

52 return self.process_bind_param(value, dialect) 

53 

54 impl = sqlalchemy.types.TypeEngine # Special placeholder 

55 

56 def __init__( # pylint: disable=keyword-arg-before-vararg 

57 self, 

58 enforce_string: bool = False, 

59 enforce_unicode: bool = False, 

60 json: types.ModuleType | typing.Any = json, # pylint: disable=redefined-outer-name 

61 json_type: TypeEngine = sqlalchemy.JSON, 

62 *args: typing.Any, 

63 **kwargs: typing.Any, 

64 ) -> None: 

65 """JSONField. 

66 

67 :param enforce_string: enforce String(UnicodeText) type usage 

68 :type enforce_string: bool 

69 :param enforce_unicode: do not encode non-ascii data 

70 :type enforce_unicode: bool 

71 :param json: JSON encoding/decoding library. By default: standard json package. 

72 :param json_type: the sqlalchemy/dialect class that will be used to render the DB JSON type. 

73 By default: sqlalchemy.JSON 

74 :param args: extra baseclass arguments 

75 :type args: typing.Any 

76 :param kwargs: extra baseclass keyworded arguments 

77 :type kwargs: typing.Any 

78 """ 

79 self.__enforce_string = enforce_string 

80 self.__enforce_unicode = enforce_unicode 

81 self.__json_codec = json 

82 self.__json_type = json_type 

83 super().__init__(*args, **kwargs) 

84 

85 def __use_json(self, dialect: DefaultDialect) -> bool: 

86 """Helper to determine, which encoder to use. 

87 

88 :return: use engine-based json encoder 

89 :rtype: bool 

90 """ 

91 return hasattr(dialect, "_json_serializer") and not self.__enforce_string 

92 

93 def load_dialect_impl(self, dialect: DefaultDialect) -> TypeEngine: 

94 """Select impl by dialect. 

95 

96 :return: dialect implementation depends on decoding method 

97 :rtype: TypeEngine 

98 """ 

99 if self.__use_json(dialect): 

100 return dialect.type_descriptor(self.__json_type) 

101 return dialect.type_descriptor(sqlalchemy.UnicodeText) 

102 

103 def process_bind_param(self, value: typing.Any, dialect: DefaultDialect) -> str | typing.Any: 

104 """Encode data, if required. 

105 

106 :return: encoded value if required 

107 :rtype: typing.Union[str, typing.Any] 

108 """ 

109 if self.__use_json(dialect) or value is None: 

110 return value 

111 

112 return self.__json_codec.dumps(value, ensure_ascii=not self.__enforce_unicode) 

113 

114 def process_result_value(self, value: str | typing.Any, dialect: DefaultDialect) -> typing.Any: 

115 """Decode data, if required. 

116 

117 :return: decoded result value if required 

118 :rtype: typing.Any 

119 """ 

120 if self.__use_json(dialect) or value is None: 

121 return value 

122 

123 return self.__json_codec.loads(value) 

124 

125 

126def mutable_json_field( # pylint: disable=keyword-arg-before-vararg, redefined-outer-name 

127 enforce_string: bool = False, 

128 enforce_unicode: bool = False, 

129 json: types.ModuleType | typing.Any = json, 

130 *args: typing.Any, 

131 **kwargs: typing.Any, 

132) -> JSONField: 

133 """Mutable JSONField creator. 

134 

135 :param enforce_string: enforce String(UnicodeText) type usage 

136 :type enforce_string: bool 

137 :param enforce_unicode: do not encode non-ascii data 

138 :type enforce_unicode: bool 

139 :param json: JSON encoding/decoding library. 

140 By default: standard json package. 

141 :param args: extra baseclass arguments 

142 :type args: typing.Any 

143 :param kwargs: extra baseclass keyworded arguments 

144 :type kwargs: typing.Any 

145 :return: Mutable JSONField via MutableDict.as_mutable 

146 :rtype: JSONField 

147 """ 

148 return sqlalchemy.ext.mutable.MutableDict.as_mutable( # type: ignore[no-any-return] 

149 JSONField( # type: ignore[misc] 

150 enforce_string=enforce_string, 

151 enforce_unicode=enforce_unicode, 

152 json=json, 

153 *args, # noqa: B026 

154 **kwargs, 

155 ) 

156 )