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-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 

18 

19from . import Connector 

20from .. import ExecutionContext 

21from .. import pool 

22from .. import util 

23from ..engine import ConnectArgsType 

24from ..engine import Connection 

25from ..engine import interfaces 

26from ..engine import URL 

27from ..sql.type_api import TypeEngine 

28 

29if typing.TYPE_CHECKING: 

30 from ..engine.interfaces import DBAPIModule 

31 from ..engine.interfaces import IsolationLevel 

32 

33 

34class PyODBCConnector(Connector): 

35 driver = "pyodbc" 

36 

37 # this is no longer False for pyodbc in general 

38 supports_sane_rowcount_returning = True 

39 supports_sane_multi_rowcount = False 

40 

41 supports_native_decimal = True 

42 default_paramstyle = "named" 

43 

44 fast_executemany = False 

45 

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

47 # hold the desired driver name 

48 pyodbc_driver_name: Optional[str] = None 

49 

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

51 super().__init__(**kw) 

52 if use_setinputsizes: 

53 self.bind_typing = interfaces.BindTyping.SETINPUTSIZES 

54 

55 @classmethod 

56 def import_dbapi(cls) -> DBAPIModule: 

57 return __import__("pyodbc") 

58 

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

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

61 opts.update(url.query) 

62 

63 keys = opts 

64 

65 query = url.query 

66 

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

68 connectors: List[str] 

69 

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

71 if param in keys: 

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

73 

74 if "odbc_connect" in keys: 

75 # (potential breaking change for issue #11250) 

76 connectors = [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 retrieve_dbapi_version( 

168 self, dbapi: interfaces.DBAPIModule 

169 ) -> util.VersionInfo: 

170 return util.parse_version_string(dbapi.version) 

171 

172 def _get_server_version_info( 

173 self, connection: Connection 

174 ) -> interfaces.ServerVersionInfoType: 

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

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

177 # queries. 

178 dbapi_con = connection.connection.dbapi_connection 

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

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

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

182 try: 

183 version += (int(n),) 

184 except ValueError: 

185 pass 

186 return tuple(version) 

187 

188 def do_set_input_sizes( 

189 self, 

190 cursor: interfaces.DBAPICursor, 

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

192 context: ExecutionContext, 

193 ) -> None: 

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

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

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

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

198 # that ticket #5649 is targeting. 

199 

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

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

202 # specified in list_of_tuples 

203 

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

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

206 # fast_executemany=True 

207 

208 if ( 

209 context.execute_style is interfaces.ExecuteStyle.EXECUTEMANY 

210 and self.fast_executemany 

211 ): 

212 return 

213 

214 cursor.setinputsizes( 

215 [ 

216 ( 

217 (dbtype, None, None) 

218 if not isinstance(dbtype, tuple) 

219 else dbtype 

220 ) 

221 for key, dbtype, sqltype in list_of_tuples 

222 ] 

223 ) 

224 

225 def get_isolation_level_values( 

226 self, dbapi_conn: interfaces.DBAPIConnection 

227 ) -> List[IsolationLevel]: 

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

229 

230 def set_isolation_level( 

231 self, 

232 dbapi_connection: interfaces.DBAPIConnection, 

233 level: IsolationLevel, 

234 ) -> None: 

235 # adjust for ConnectionFairy being present 

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

237 # to work properly 

238 

239 if level == "AUTOCOMMIT": 

240 dbapi_connection.autocommit = True 

241 else: 

242 dbapi_connection.autocommit = False 

243 super().set_isolation_level(dbapi_connection, level) 

244 

245 def detect_autocommit_setting( 

246 self, dbapi_conn: interfaces.DBAPIConnection 

247 ) -> bool: 

248 return bool(dbapi_conn.autocommit)