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

1# connectors/pyodbc.py 

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

11import typing 

12from typing import Any 

13from typing import Dict 

14from typing import List 

15from typing import Optional 

16from typing import Tuple 

17from typing import Union 

18from urllib.parse import unquote_plus 

19 

20from . import Connector 

21from .. import ExecutionContext 

22from .. import pool 

23from .. import util 

24from ..engine import ConnectArgsType 

25from ..engine import Connection 

26from ..engine import interfaces 

27from ..engine import URL 

28from ..sql.type_api import TypeEngine 

29 

30if typing.TYPE_CHECKING: 

31 from ..engine.interfaces import DBAPIModule 

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 def __init__(self, use_setinputsizes: bool = False, **kw: Any): 

52 super().__init__(**kw) 

53 if use_setinputsizes: 

54 self.bind_typing = interfaces.BindTyping.SETINPUTSIZES 

55 

56 @classmethod 

57 def import_dbapi(cls) -> DBAPIModule: 

58 return __import__("pyodbc") 

59 

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

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

62 opts.update(url.query) 

63 

64 keys = opts 

65 

66 query = url.query 

67 

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

69 connectors: List[str] 

70 

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

72 if param in keys: 

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

74 

75 if "odbc_connect" in keys: 

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

77 else: 

78 

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

80 if ( 

81 ";" in str(token) 

82 or "}" in str(token) 

83 or str(token).startswith("{") 

84 ): 

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

86 return token 

87 

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

89 

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

91 

92 dsn_connection = "dsn" in keys or ( 

93 "host" in keys and "database" not in keys 

94 ) 

95 if dsn_connection: 

96 connectors = [ 

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

98 ] 

99 else: 

100 port = "" 

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

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

103 

104 connectors = [] 

105 if driver is None and keys: 

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

107 util.warn( 

108 "No driver name specified; " 

109 "this is expected by PyODBC when using " 

110 "DSN-less connections" 

111 ) 

112 else: 

113 connectors.append( 

114 "DRIVER={%s}" % str(driver).replace("}", "}}") 

115 ) 

116 

117 connectors.extend( 

118 [ 

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

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

121 ] 

122 ) 

123 

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

125 if user: 

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

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

128 if pwd: 

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

130 else: 

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

132 if authentication: 

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

134 else: 

135 connectors.append("Trusted_Connection=Yes") 

136 

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

138 # convert textual data from your database encoding to your 

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

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

141 if "odbc_autotranslate" in keys: 

142 connectors.append( 

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

144 ) 

145 

146 connectors.extend( 

147 ["%s=%s" % (check_quote(k), v) for k, v in keys.items()] 

148 ) 

149 

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

151 

152 def is_disconnect( 

153 self, 

154 e: Exception, 

155 connection: Optional[ 

156 Union[pool.PoolProxiedConnection, interfaces.DBAPIConnection] 

157 ], 

158 cursor: Optional[interfaces.DBAPICursor], 

159 ) -> bool: 

160 if isinstance(e, self.loaded_dbapi.ProgrammingError): 

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

162 e 

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

164 else: 

165 return False 

166 

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

168 if not self.dbapi: 

169 return () 

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

171 

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

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

174 if not m: 

175 return () 

176 vers_tuple: interfaces.VersionInfoType = tuple( 

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

178 ) 

179 if m.group(2): 

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

181 return vers_tuple 

182 

183 def _get_server_version_info( 

184 self, connection: Connection 

185 ) -> interfaces.VersionInfoType: 

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

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

188 # queries. 

189 dbapi_con = connection.connection.dbapi_connection 

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

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

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

193 try: 

194 version += (int(n),) 

195 except ValueError: 

196 pass 

197 return tuple(version) 

198 

199 def do_set_input_sizes( 

200 self, 

201 cursor: interfaces.DBAPICursor, 

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

203 context: ExecutionContext, 

204 ) -> None: 

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

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

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

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

209 # that ticket #5649 is targeting. 

210 

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

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

213 # specified in list_of_tuples 

214 

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

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

217 # fast_executemany=True 

218 

219 if ( 

220 context.execute_style is interfaces.ExecuteStyle.EXECUTEMANY 

221 and self.fast_executemany 

222 ): 

223 return 

224 

225 cursor.setinputsizes( 

226 [ 

227 ( 

228 (dbtype, None, None) 

229 if not isinstance(dbtype, tuple) 

230 else dbtype 

231 ) 

232 for key, dbtype, sqltype in list_of_tuples 

233 ] 

234 ) 

235 

236 def get_isolation_level_values( 

237 self, dbapi_conn: interfaces.DBAPIConnection 

238 ) -> List[IsolationLevel]: 

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

240 

241 def set_isolation_level( 

242 self, 

243 dbapi_connection: interfaces.DBAPIConnection, 

244 level: IsolationLevel, 

245 ) -> None: 

246 # adjust for ConnectionFairy being present 

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

248 # to work properly 

249 

250 if level == "AUTOCOMMIT": 

251 dbapi_connection.autocommit = True 

252 else: 

253 dbapi_connection.autocommit = False 

254 super().set_isolation_level(dbapi_connection, level) 

255 

256 def detect_autocommit_setting( 

257 self, dbapi_conn: interfaces.DBAPIConnection 

258 ) -> bool: 

259 return bool(dbapi_conn.autocommit)