Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/psycopg2/extensions.py: 59%

1"""psycopg extensions to the DBAPI-2.0 

2 

3This module holds all the extensions to the DBAPI-2.0 provided by psycopg. 

4 

5- `connection` -- the new-type inheritable connection class 

6- `cursor` -- the new-type inheritable cursor class 

7- `lobject` -- the new-type inheritable large object class 

8- `adapt()` -- exposes the PEP-246_ compatible adapting mechanism used 

9 by psycopg to adapt Python types to PostgreSQL ones 

10 

11.. _PEP-246: https://www.python.org/dev/peps/pep-0246/ 

12""" 

13# psycopg/extensions.py - DBAPI-2.0 extensions specific to psycopg 

14# 

15# Copyright (C) 2003-2019 Federico Di Gregorio <fog@debian.org> 

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

17# 

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

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

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

21# (at your option) any later version. 

22# 

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

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

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

26# and distribute linked combinations including the two. 

27# 

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

29# all of the code used other than OpenSSL. 

30# 

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

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

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

34# License for more details. 

35 

36import re as _re 

37 

38from psycopg2._psycopg import ( # noqa 

39 BINARYARRAY, BOOLEAN, BOOLEANARRAY, BYTES, BYTESARRAY, DATE, DATEARRAY, 

40 DATETIMEARRAY, DECIMAL, DECIMALARRAY, FLOAT, FLOATARRAY, INTEGER, 

41 INTEGERARRAY, INTERVAL, INTERVALARRAY, LONGINTEGER, LONGINTEGERARRAY, 

42 ROWIDARRAY, STRINGARRAY, TIME, TIMEARRAY, UNICODE, UNICODEARRAY, 

43 AsIs, Binary, Boolean, Float, Int, QuotedString, ) 

44 

45from psycopg2._psycopg import ( # noqa 

46 PYDATE, PYDATETIME, PYDATETIMETZ, PYINTERVAL, PYTIME, PYDATEARRAY, 

47 PYDATETIMEARRAY, PYDATETIMETZARRAY, PYINTERVALARRAY, PYTIMEARRAY, 

48 DateFromPy, TimeFromPy, TimestampFromPy, IntervalFromPy, ) 

49 

50from psycopg2._psycopg import ( # noqa 

51 adapt, adapters, encodings, connection, cursor, 

52 lobject, Xid, libpq_version, parse_dsn, quote_ident, 

53 string_types, binary_types, new_type, new_array_type, register_type, 

54 ISQLQuote, Notify, Diagnostics, Column, ConnectionInfo, 

55 QueryCanceledError, TransactionRollbackError, 

56 set_wait_callback, get_wait_callback, encrypt_password, ) 

57 

58 

59"""Isolation level values.""" 

60ISOLATION_LEVEL_AUTOCOMMIT = 0 

61ISOLATION_LEVEL_READ_UNCOMMITTED = 4 

62ISOLATION_LEVEL_READ_COMMITTED = 1 

63ISOLATION_LEVEL_REPEATABLE_READ = 2 

64ISOLATION_LEVEL_SERIALIZABLE = 3 

65ISOLATION_LEVEL_DEFAULT = None 

66 

67 

68"""psycopg connection status values.""" 

69STATUS_SETUP = 0 

70STATUS_READY = 1 

71STATUS_BEGIN = 2 

72STATUS_SYNC = 3 # currently unused 

73STATUS_ASYNC = 4 # currently unused 

74STATUS_PREPARED = 5 

75 

76# This is a useful mnemonic to check if the connection is in a transaction 

77STATUS_IN_TRANSACTION = STATUS_BEGIN 

78 

79 

80"""psycopg asynchronous connection polling values""" 

81POLL_OK = 0 

82POLL_READ = 1 

83POLL_WRITE = 2 

84POLL_ERROR = 3 

85 

86 

87"""Backend transaction status values.""" 

88TRANSACTION_STATUS_IDLE = 0 

89TRANSACTION_STATUS_ACTIVE = 1 

90TRANSACTION_STATUS_INTRANS = 2 

91TRANSACTION_STATUS_INERROR = 3 

92TRANSACTION_STATUS_UNKNOWN = 4 

93 

94 

95def register_adapter(typ, callable): 

96 """Register 'callable' as an ISQLQuote adapter for type 'typ'.""" 

97 adapters[(typ, ISQLQuote)] = callable 

98 

99 

100# The SQL_IN class is the official adapter for tuples starting from 2.0.6. 

101class SQL_IN: 

102 """Adapt any iterable to an SQL quotable object.""" 

103 def __init__(self, seq): 

104 self._seq = seq 

105 self._conn = None 

106 

107 def prepare(self, conn): 

108 self._conn = conn 

109 

110 def getquoted(self): 

111 # this is the important line: note how every object in the 

112 # list is adapted and then how getquoted() is called on it 

113 pobjs = [adapt(o) for o in self._seq] 

114 if self._conn is not None: 

115 for obj in pobjs: 

116 if hasattr(obj, 'prepare'): 

117 obj.prepare(self._conn) 

118 qobjs = [o.getquoted() for o in pobjs] 

119 return b'(' + b', '.join(qobjs) + b')' 

120 

121 def __str__(self): 

122 return str(self.getquoted()) 

123 

124 

125class NoneAdapter: 

126 """Adapt None to NULL. 

127 

128 This adapter is not used normally as a fast path in mogrify uses NULL, 

129 but it makes easier to adapt composite types. 

130 """ 

131 def __init__(self, obj): 

132 pass 

133 

134 def getquoted(self, _null=b"NULL"): 

135 return _null 

136 

137 

138def make_dsn(dsn=None, **kwargs): 

139 """Convert a set of keywords into a connection strings.""" 

140 if dsn is None and not kwargs: 

141 return '' 

142 

143 # If no kwarg is specified don't mung the dsn, but verify it 

144 if not kwargs: 

145 parse_dsn(dsn) 

146 return dsn 

147 

148 # Override the dsn with the parameters 

149 if 'database' in kwargs: 

150 if 'dbname' in kwargs: 

151 raise TypeError( 

152 "you can't specify both 'database' and 'dbname' arguments") 

153 kwargs['dbname'] = kwargs.pop('database') 

154 

155 # Drop the None arguments 

156 kwargs = {k: v for (k, v) in kwargs.items() if v is not None} 

157 

158 if dsn is not None: 

159 tmp = parse_dsn(dsn) 

160 tmp.update(kwargs) 

161 kwargs = tmp 

162 

163 dsn = " ".join(["{}={}".format(k, _param_escape(str(v))) 

164 for (k, v) in kwargs.items()]) 

165 

166 # verify that the returned dsn is valid 

167 parse_dsn(dsn) 

168 

169 return dsn 

170 

171 

172def _param_escape(s, 

173 re_escape=_re.compile(r"([\\'])"), 

174 re_space=_re.compile(r'\s')): 

175 """ 

176 Apply the escaping rule required by PQconnectdb 

177 """ 

178 if not s: 

179 return "''" 

180 

181 s = re_escape.sub(r'\\\1', s) 

182 if re_space.search(s): 

183 s = "'" + s + "'" 

184 

185 return s 

186 

187 

188# Create default json typecasters for PostgreSQL 9.2 oids 

189from psycopg2._json import register_default_json, register_default_jsonb # noqa 

190 

191try: 

192 JSON, JSONARRAY = register_default_json() 

193 JSONB, JSONBARRAY = register_default_jsonb() 

194except ImportError: 

195 pass 

196 

197del register_default_json, register_default_jsonb 

198 

199 

200# Create default Range typecasters 

201from psycopg2. _range import Range # noqa 

202del Range 

203 

204 

205# Add the "cleaned" version of the encodings to the key. 

206# When the encoding is set its name is cleaned up from - and _ and turned 

207# uppercase, so an encoding not respecting these rules wouldn't be found in the 

208# encodings keys and would raise an exception with the unicode typecaster 

209for k, v in list(encodings.items()): 

210 k = k.replace('_', '').replace('-', '').upper() 

211 encodings[k] = v 

212 

213del k, v