Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/connectors/pyodbc.py: 23%

1# connectors/pyodbc.py 

2# Copyright (C) 2005-2023 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 

8import re 

9 

10from . import Connector 

11from .. import util 

12 

13 

14class PyODBCConnector(Connector): 

15 driver = "pyodbc" 

16 

17 # this is no longer False for pyodbc in general 

18 supports_sane_rowcount_returning = True 

19 supports_sane_multi_rowcount = False 

20 

21 supports_unicode_statements = True 

22 supports_unicode_binds = True 

23 

24 supports_native_decimal = True 

25 default_paramstyle = "named" 

26 

27 use_setinputsizes = False 

28 

29 # for non-DSN connections, this *may* be used to 

30 # hold the desired driver name 

31 pyodbc_driver_name = None 

32 

33 def __init__( 

34 self, supports_unicode_binds=None, use_setinputsizes=False, **kw 

35 ): 

36 super(PyODBCConnector, self).__init__(**kw) 

37 if supports_unicode_binds is not None: 

38 self.supports_unicode_binds = supports_unicode_binds 

39 self.use_setinputsizes = use_setinputsizes 

40 

41 @classmethod 

42 def dbapi(cls): 

43 return __import__("pyodbc") 

44 

45 def create_connect_args(self, url): 

46 opts = url.translate_connect_args(username="user") 

47 opts.update(url.query) 

48 

49 keys = opts 

50 

51 query = url.query 

52 

53 connect_args = {} 

54 for param in ("ansi", "unicode_results", "autocommit"): 

55 if param in keys: 

56 connect_args[param] = util.asbool(keys.pop(param)) 

57 

58 if "odbc_connect" in keys: 

59 connectors = [util.unquote_plus(keys.pop("odbc_connect"))] 

60 else: 

61 

62 def check_quote(token): 

63 if ";" in str(token) or str(token).startswith("{"): 

64 token = "{%s}" % token.replace("}", "}}") 

65 return token 

66 

67 keys = dict((k, check_quote(v)) for k, v in keys.items()) 

68 

69 dsn_connection = "dsn" in keys or ( 

70 "host" in keys and "database" not in keys 

71 ) 

72 if dsn_connection: 

73 connectors = [ 

74 "dsn=%s" % (keys.pop("host", "") or keys.pop("dsn", "")) 

75 ] 

76 else: 

77 port = "" 

78 if "port" in keys and "port" not in query: 

79 port = ",%d" % int(keys.pop("port")) 

80 

81 connectors = [] 

82 driver = keys.pop("driver", self.pyodbc_driver_name) 

83 if driver is None and keys: 

84 # note if keys is empty, this is a totally blank URL 

85 util.warn( 

86 "No driver name specified; " 

87 "this is expected by PyODBC when using " 

88 "DSN-less connections" 

89 ) 

90 else: 

91 connectors.append("DRIVER={%s}" % driver) 

92 

93 connectors.extend( 

94 [ 

95 "Server=%s%s" % (keys.pop("host", ""), port), 

96 "Database=%s" % keys.pop("database", ""), 

97 ] 

98 ) 

99 

100 user = keys.pop("user", None) 

101 if user: 

102 connectors.append("UID=%s" % user) 

103 pwd = keys.pop("password", "") 

104 if pwd: 

105 connectors.append("PWD=%s" % pwd) 

106 else: 

107 authentication = keys.pop("authentication", None) 

108 if authentication: 

109 connectors.append("Authentication=%s" % authentication) 

110 else: 

111 connectors.append("Trusted_Connection=Yes") 

112 

113 # if set to 'Yes', the ODBC layer will try to automagically 

114 # convert textual data from your database encoding to your 

115 # client encoding. This should obviously be set to 'No' if 

116 # you query a cp1253 encoded database from a latin1 client... 

117 if "odbc_autotranslate" in keys: 

118 connectors.append( 

119 "AutoTranslate=%s" % keys.pop("odbc_autotranslate") 

120 ) 

121 

122 connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()]) 

123 

124 return [[";".join(connectors)], connect_args] 

125 

126 def is_disconnect(self, e, connection, cursor): 

127 if isinstance(e, self.dbapi.ProgrammingError): 

128 return "The cursor's connection has been closed." in str( 

129 e 

130 ) or "Attempt to use a closed connection." in str(e) 

131 else: 

132 return False 

133 

134 def _dbapi_version(self): 

135 if not self.dbapi: 

136 return () 

137 return self._parse_dbapi_version(self.dbapi.version) 

138 

139 def _parse_dbapi_version(self, vers): 

140 m = re.match(r"(?:py.*-)?([\d\.]+)(?:-(\w+))?", vers) 

141 if not m: 

142 return () 

143 vers = tuple([int(x) for x in m.group(1).split(".")]) 

144 if m.group(2): 

145 vers += (m.group(2),) 

146 return vers 

147 

148 def _get_server_version_info(self, connection, allow_chars=True): 

149 # NOTE: this function is not reliable, particularly when 

150 # freetds is in use. Implement database-specific server version 

151 # queries. 

152 dbapi_con = connection.connection 

153 version = [] 

154 r = re.compile(r"[.\-]") 

155 for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)): 

156 try: 

157 version.append(int(n)) 

158 except ValueError: 

159 if allow_chars: 

160 version.append(n) 

161 return tuple(version) 

162 

163 def do_set_input_sizes(self, cursor, list_of_tuples, context): 

164 # the rules for these types seems a little strange, as you can pass 

165 # non-tuples as well as tuples, however it seems to assume "0" 

166 # for the subsequent values if you don't pass a tuple which fails 

167 # for types such as pyodbc.SQL_WLONGVARCHAR, which is the datatype 

168 # that ticket #5649 is targeting. 

169 

170 # NOTE: as of #6058, this won't be called if the use_setinputsizes flag 

171 # is False, or if no types were specified in list_of_tuples 

172 

173 cursor.setinputsizes( 

174 [ 

175 (dbtype, None, None) 

176 if not isinstance(dbtype, tuple) 

177 else dbtype 

178 for key, dbtype, sqltype in list_of_tuples 

179 ] 

180 ) 

181 

182 def set_isolation_level(self, connection, level): 

183 # adjust for ConnectionFairy being present 

184 # allows attribute set e.g. "connection.autocommit = True" 

185 # to work properly 

186 if hasattr(connection, "dbapi_connection"): 

187 connection = connection.dbapi_connection 

188 

189 if level == "AUTOCOMMIT": 

190 connection.autocommit = True 

191 else: 

192 connection.autocommit = False 

193 super(PyODBCConnector, self).set_isolation_level(connection, level)