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

1# connectors/pyodbc.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 

8from __future__ import annotations 

9 

10import re 

11from types import ModuleType 

12import typing 

13from typing import Any 

14from typing import Dict 

15from typing import List 

16from typing import Optional 

17from typing import Tuple 

18from typing import Union 

19from urllib.parse import unquote_plus 

20 

21from . import Connector 

22from .. import ExecutionContext 

23from .. import pool 

24from .. import util 

25from ..engine import ConnectArgsType 

26from ..engine import Connection 

27from ..engine import interfaces 

28from ..engine import URL 

29from ..sql.type_api import TypeEngine 

30 

31if typing.TYPE_CHECKING: 

32 from ..engine.interfaces import IsolationLevel 

33 

34 

35class PyODBCConnector(Connector): 

36 driver = "pyodbc" 

37 

38 # this is no longer False for pyodbc in general 

39 supports_sane_rowcount_returning = True 

40 supports_sane_multi_rowcount = False 

41 

42 supports_native_decimal = True 

43 default_paramstyle = "named" 

44 

45 fast_executemany = False 

46 

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

48 # hold the desired driver name 

49 pyodbc_driver_name: Optional[str] = None 

50 

51 dbapi: ModuleType 

52 

53 def __init__(self, use_setinputsizes: bool = False, **kw: Any): 

54 super().__init__(**kw) 

55 if use_setinputsizes: 

56 self.bind_typing = interfaces.BindTyping.SETINPUTSIZES 

57 

58 @classmethod 

59 def import_dbapi(cls) -> ModuleType: 

60 return __import__("pyodbc") 

61 

62 def create_connect_args(self, url: URL) -> ConnectArgsType: 

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

64 opts.update(url.query) 

65 

66 keys = opts 

67 

68 query = url.query 

69 

70 connect_args: Dict[str, Any] = {} 

71 connectors: List[str] 

72 

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

74 if param in keys: 

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

76 

77 if "odbc_connect" in keys: 

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

79 else: 

80 

81 def check_quote(token: str) -> str: 

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

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

84 return token 

85 

86 keys = {k: check_quote(v) for k, v in keys.items()} 

87 

88 dsn_connection = "dsn" in keys or ( 

89 "host" in keys and "database" not in keys 

90 ) 

91 if dsn_connection: 

92 connectors = [ 

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

94 ] 

95 else: 

96 port = "" 

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

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

99 

100 connectors = [] 

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

102 if driver is None and keys: 

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

104 util.warn( 

105 "No driver name specified; " 

106 "this is expected by PyODBC when using " 

107 "DSN-less connections" 

108 ) 

109 else: 

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

111 

112 connectors.extend( 

113 [ 

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

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

116 ] 

117 ) 

118 

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

120 if user: 

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

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

123 if pwd: 

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

125 else: 

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

127 if authentication: 

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

129 else: 

130 connectors.append("Trusted_Connection=Yes") 

131 

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

133 # convert textual data from your database encoding to your 

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

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

136 if "odbc_autotranslate" in keys: 

137 connectors.append( 

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

139 ) 

140 

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

142 

143 return ((";".join(connectors),), connect_args) 

144 

145 def is_disconnect( 

146 self, 

147 e: Exception, 

148 connection: Optional[ 

149 Union[pool.PoolProxiedConnection, interfaces.DBAPIConnection] 

150 ], 

151 cursor: Optional[interfaces.DBAPICursor], 

152 ) -> bool: 

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

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

155 e 

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

157 else: 

158 return False 

159 

160 def _dbapi_version(self) -> interfaces.VersionInfoType: 

161 if not self.dbapi: 

162 return () 

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

164 

165 def _parse_dbapi_version(self, vers: str) -> interfaces.VersionInfoType: 

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

167 if not m: 

168 return () 

169 vers_tuple: interfaces.VersionInfoType = tuple( 

170 [int(x) for x in m.group(1).split(".")] 

171 ) 

172 if m.group(2): 

173 vers_tuple += (m.group(2),) 

174 return vers_tuple 

175 

176 def _get_server_version_info( 

177 self, connection: Connection 

178 ) -> interfaces.VersionInfoType: 

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

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

181 # queries. 

182 dbapi_con = connection.connection.dbapi_connection 

183 version: Tuple[Union[int, str], ...] = () 

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

185 for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)): # type: ignore[union-attr] # noqa: E501 

186 try: 

187 version += (int(n),) 

188 except ValueError: 

189 pass 

190 return tuple(version) 

191 

192 def do_set_input_sizes( 

193 self, 

194 cursor: interfaces.DBAPICursor, 

195 list_of_tuples: List[Tuple[str, Any, TypeEngine[Any]]], 

196 context: ExecutionContext, 

197 ) -> None: 

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

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

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

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

202 # that ticket #5649 is targeting. 

203 

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

205 # parameter were not passed to the dialect, or if no types were 

206 # specified in list_of_tuples 

207 

208 # as of #8177 for 2.0 we assume use_setinputsizes=True and only 

209 # omit the setinputsizes calls for .executemany() with 

210 # fast_executemany=True 

211 

212 if ( 

213 context.execute_style is interfaces.ExecuteStyle.EXECUTEMANY 

214 and self.fast_executemany 

215 ): 

216 return 

217 

218 cursor.setinputsizes( 

219 [ 

220 ( 

221 (dbtype, None, None) 

222 if not isinstance(dbtype, tuple) 

223 else dbtype 

224 ) 

225 for key, dbtype, sqltype in list_of_tuples 

226 ] 

227 ) 

228 

229 def get_isolation_level_values( 

230 self, dbapi_conn: interfaces.DBAPIConnection 

231 ) -> List[IsolationLevel]: 

232 return [*super().get_isolation_level_values(dbapi_conn), "AUTOCOMMIT"] 

233 

234 def set_isolation_level( 

235 self, 

236 dbapi_connection: interfaces.DBAPIConnection, 

237 level: IsolationLevel, 

238 ) -> None: 

239 # adjust for ConnectionFairy being present 

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

241 # to work properly 

242 

243 if level == "AUTOCOMMIT": 

244 dbapi_connection.autocommit = True 

245 else: 

246 dbapi_connection.autocommit = False 

247 super().set_isolation_level(dbapi_connection, level)