Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/psycopg2/_json.py: 55%

1"""Implementation of the JSON adaptation objects 

2 

3This module exists to avoid a circular import problem: pyscopg2.extras depends 

4on psycopg2.extension, so I can't create the default JSON typecasters in 

5extensions importing register_json from extras. 

6""" 

7 

8# psycopg/_json.py - Implementation of the JSON adaptation objects 

9# 

10# Copyright (C) 2012-2019 Daniele Varrazzo <daniele.varrazzo@gmail.com> 

11# Copyright (C) 2020-2021 The Psycopg Team 

12# 

13# psycopg2 is free software: you can redistribute it and/or modify it 

14# under the terms of the GNU Lesser General Public License as published 

15# by the Free Software Foundation, either version 3 of the License, or 

16# (at your option) any later version. 

17# 

18# In addition, as a special exception, the copyright holders give 

19# permission to link this program with the OpenSSL library (or with 

20# modified versions of OpenSSL that use the same license as OpenSSL), 

21# and distribute linked combinations including the two. 

22# 

23# You must obey the GNU Lesser General Public License in all respects for 

24# all of the code used other than OpenSSL. 

25# 

26# psycopg2 is distributed in the hope that it will be useful, but WITHOUT 

27# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

28# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public 

29# License for more details. 

30 

31import json 

32 

33from psycopg2._psycopg import ISQLQuote, QuotedString 

34from psycopg2._psycopg import new_type, new_array_type, register_type 

35 

36 

37# oids from PostgreSQL 9.2 

38JSON_OID = 114 

39JSONARRAY_OID = 199 

40 

41# oids from PostgreSQL 9.4 

42JSONB_OID = 3802 

43JSONBARRAY_OID = 3807 

44 

45 

46class Json: 

47 """ 

48 An `~psycopg2.extensions.ISQLQuote` wrapper to adapt a Python object to 

49 :sql:`json` data type. 

50 

51 `!Json` can be used to wrap any object supported by the provided *dumps* 

52 function. If none is provided, the standard :py:func:`json.dumps()` is 

53 used. 

54 

55 """ 

56 def __init__(self, adapted, dumps=None): 

57 self.adapted = adapted 

58 self._conn = None 

59 self._dumps = dumps or json.dumps 

60 

61 def __conform__(self, proto): 

62 if proto is ISQLQuote: 

63 return self 

64 

65 def dumps(self, obj): 

66 """Serialize *obj* in JSON format. 

67 

68 The default is to call `!json.dumps()` or the *dumps* function 

69 provided in the constructor. You can override this method to create a 

70 customized JSON wrapper. 

71 """ 

72 return self._dumps(obj) 

73 

74 def prepare(self, conn): 

75 self._conn = conn 

76 

77 def getquoted(self): 

78 s = self.dumps(self.adapted) 

79 qs = QuotedString(s) 

80 if self._conn is not None: 

81 qs.prepare(self._conn) 

82 return qs.getquoted() 

83 

84 def __str__(self): 

85 # getquoted is binary 

86 return self.getquoted().decode('ascii', 'replace') 

87 

88 

89def register_json(conn_or_curs=None, globally=False, loads=None, 

90 oid=None, array_oid=None, name='json'): 

91 """Create and register typecasters converting :sql:`json` type to Python objects. 

92 

93 :param conn_or_curs: a connection or cursor used to find the :sql:`json` 

94 and :sql:`json[]` oids; the typecasters are registered in a scope 

95 limited to this object, unless *globally* is set to `!True`. It can be 

96 `!None` if the oids are provided 

97 :param globally: if `!False` register the typecasters only on 

98 *conn_or_curs*, otherwise register them globally 

99 :param loads: the function used to parse the data into a Python object. If 

100 `!None` use `!json.loads()`, where `!json` is the module chosen 

101 according to the Python version (see above) 

102 :param oid: the OID of the :sql:`json` type if known; If not, it will be 

103 queried on *conn_or_curs* 

104 :param array_oid: the OID of the :sql:`json[]` array type if known; 

105 if not, it will be queried on *conn_or_curs* 

106 :param name: the name of the data type to look for in *conn_or_curs* 

107 

108 The connection or cursor passed to the function will be used to query the 

109 database and look for the OID of the :sql:`json` type (or an alternative 

110 type if *name* if provided). No query is performed if *oid* and *array_oid* 

111 are provided. Raise `~psycopg2.ProgrammingError` if the type is not found. 

112 

113 """ 

114 if oid is None: 114 ↛ 115line 114 didn't jump to line 115, because the condition on line 114 was never true

115 oid, array_oid = _get_json_oids(conn_or_curs, name) 

116 

117 JSON, JSONARRAY = _create_json_typecasters( 

118 oid, array_oid, loads=loads, name=name.upper()) 

119 

120 register_type(JSON, not globally and conn_or_curs or None) 

121 

122 if JSONARRAY is not None: 122 ↛ 125line 122 didn't jump to line 125, because the condition on line 122 was never false

123 register_type(JSONARRAY, not globally and conn_or_curs or None) 

124 

125 return JSON, JSONARRAY 

126 

127 

128def register_default_json(conn_or_curs=None, globally=False, loads=None): 

129 """ 

130 Create and register :sql:`json` typecasters for PostgreSQL 9.2 and following. 

131 

132 Since PostgreSQL 9.2 :sql:`json` is a builtin type, hence its oid is known 

133 and fixed. This function allows specifying a customized *loads* function 

134 for the default :sql:`json` type without querying the database. 

135 All the parameters have the same meaning of `register_json()`. 

136 """ 

137 return register_json(conn_or_curs=conn_or_curs, globally=globally, 

138 loads=loads, oid=JSON_OID, array_oid=JSONARRAY_OID) 

139 

140 

141def register_default_jsonb(conn_or_curs=None, globally=False, loads=None): 

142 """ 

143 Create and register :sql:`jsonb` typecasters for PostgreSQL 9.4 and following. 

144 

145 As in `register_default_json()`, the function allows to register a 

146 customized *loads* function for the :sql:`jsonb` type at its known oid for 

147 PostgreSQL 9.4 and following versions. All the parameters have the same 

148 meaning of `register_json()`. 

149 """ 

150 return register_json(conn_or_curs=conn_or_curs, globally=globally, 

151 loads=loads, oid=JSONB_OID, array_oid=JSONBARRAY_OID, name='jsonb') 

152 

153 

154def _create_json_typecasters(oid, array_oid, loads=None, name='JSON'): 

155 """Create typecasters for json data type.""" 

156 if loads is None: 

157 loads = json.loads 

158 

159 def typecast_json(s, cur): 

160 if s is None: 

161 return None 

162 return loads(s) 

163 

164 JSON = new_type((oid, ), name, typecast_json) 

165 if array_oid is not None: 165 ↛ 168line 165 didn't jump to line 168, because the condition on line 165 was never false

166 JSONARRAY = new_array_type((array_oid, ), f"{name}ARRAY", JSON) 

167 else: 

168 JSONARRAY = None 

169 

170 return JSON, JSONARRAY 

171 

172 

173def _get_json_oids(conn_or_curs, name='json'): 

174 # lazy imports 

175 from psycopg2.extensions import STATUS_IN_TRANSACTION 

176 from psycopg2.extras import _solve_conn_curs 

177 

178 conn, curs = _solve_conn_curs(conn_or_curs) 

179 

180 # Store the transaction status of the connection to revert it after use 

181 conn_status = conn.status 

182 

183 # column typarray not available before PG 8.3 

184 typarray = conn.info.server_version >= 80300 and "typarray" or "NULL" 

185 

186 # get the oid for the hstore 

187 curs.execute( 

188 "SELECT t.oid, %s FROM pg_type t WHERE t.typname = %%s;" 

189 % typarray, (name,)) 

190 r = curs.fetchone() 

191 

192 # revert the status of the connection as before the command 

193 if conn_status != STATUS_IN_TRANSACTION and not conn.autocommit: 

194 conn.rollback() 

195 

196 if not r: 

197 raise conn.ProgrammingError(f"{name} data type not found") 

198 

199 return r