Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/mssql/mssqlpython.py: 39%

1# dialects/mssql/mssqlpython.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# mypy: ignore-errors 

8 

9""" 

10.. dialect:: mssql+mssqlpython 

11 :name: mssqlpython 

12 :dbapi: mssql-python 

13 :connectstring: mssql+mssqlpython://<username>:<password>@<host>:<port>/<dbname> 

14 :url: https://github.com/microsoft/mssql-python 

15 

16mssql-python is a driver for Microsoft SQL Server produced by Microsoft. 

17 

18.. versionadded:: 2.1.0b2 

19 

20 

21The driver is generally similar to pyodbc in most aspects as it is based 

22on the same ODBC framework. 

23 

24Connection Strings 

25------------------ 

26 

27Examples of connecting with the mssql-python driver:: 

28 

29 from sqlalchemy import create_engine 

30 

31 # Basic connection 

32 engine = create_engine( 

33 "mssql+mssqlpython://user:password@hostname/database" 

34 ) 

35 

36 # With Windows Authentication 

37 engine = create_engine( 

38 "mssql+mssqlpython://hostname/database?authentication=ActiveDirectoryIntegrated" 

39 ) 

40 

41""" # noqa 

42 

43from __future__ import annotations 

44 

45import re 

46from typing import Any 

47from typing import Dict 

48from typing import List 

49from typing import Optional 

50from typing import TYPE_CHECKING 

51from typing import Union 

52 

53from .base import MSDialect 

54from .pyodbc import _ms_numeric_pyodbc 

55from ... import util 

56from ...sql import sqltypes 

57 

58if TYPE_CHECKING: 

59 from ... import pool 

60 from ...engine import interfaces 

61 from ...engine.interfaces import ConnectArgsType 

62 from ...engine.interfaces import DBAPIModule 

63 from ...engine.interfaces import IsolationLevel 

64 from ...engine.interfaces import URL 

65 

66 

67class _MSNumeric_mssqlpython(_ms_numeric_pyodbc, sqltypes.Numeric): 

68 pass 

69 

70 

71class _MSFloat_mssqlpython(_ms_numeric_pyodbc, sqltypes.Float): 

72 pass 

73 

74 

75class MSDialect_mssqlpython(MSDialect): 

76 driver = "mssqlpython" 

77 supports_statement_cache = True 

78 

79 supports_sane_rowcount_returning = True 

80 supports_sane_multi_rowcount = True 

81 supports_native_uuid = True 

82 scope_identity_must_be_embedded = True 

83 

84 supports_native_decimal = True 

85 

86 _disconnect_driver_errors = frozenset( 

87 { 

88 # mssql-python converts the originating SQLSTATE into these 

89 # values and exposes them on ``Exception.driver_error``. 

90 "Disconnect error", # 01002 

91 "Client unable to establish connection", # 08001 

92 "Connection not open", # 08003 

93 "Connection failure during transaction", # 08007 

94 "Communication link failure", # 08S01 

95 # mssql-python does not map these SQLSTATE values to a 

96 # dedicated driver error yet. 

97 "An error occurred with SQLSTATE code: 08S02", 

98 "An error occurred with SQLSTATE code: 10054", 

99 "Connection timeout expired", # HYT01 

100 "Function sequence error", # HY010 

101 } 

102 ) 

103 

104 colspecs = util.update_copy( 

105 MSDialect.colspecs, 

106 { 

107 sqltypes.Numeric: _MSNumeric_mssqlpython, 

108 sqltypes.Float: _MSFloat_mssqlpython, 

109 }, 

110 ) 

111 

112 def __init__(self, enable_pooling=False, **kw): 

113 super().__init__(**kw) 

114 if not enable_pooling and self.dbapi is not None: 

115 self.loaded_dbapi.pooling(enabled=False) 

116 

117 # used by pyodbc _ms_numeric_pyodbc class. mssql-python 1.15 

118 # binds every Decimal as SQL_NUMERIC; prior versions bound Decimal 

119 # values within the MONEY range as VARCHAR, so that very large and 

120 # very small values needed to be passed as strings as well. On 

121 # 1.15 those strings are instead coerced to the NUMERIC type of 

122 # the Decimal values alongside them, which overflows. 

123 # a version which can't be determined is assumed to be an old one 

124 version = self._dbapi_version_or_none 

125 self._need_decimal_fix = bool(self.dbapi) and ( 

126 version is None or version < (1, 15) 

127 ) 

128 

129 @classmethod 

130 def import_dbapi(cls) -> DBAPIModule: 

131 return __import__("mssql_python") 

132 

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

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

135 opts.update(url.query) 

136 

137 keys = opts 

138 

139 query = url.query 

140 

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

142 connectors: List[str] 

143 

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

145 if ( 

146 ";" in str(token) 

147 or "}" in str(token) 

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

149 ): 

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

151 return token 

152 

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

154 

155 port = "" 

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

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

158 

159 connectors = [] 

160 

161 connectors.extend( 

162 [ 

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

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

165 ] 

166 ) 

167 

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

169 if user: 

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

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

172 if pwd: 

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

174 else: 

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

176 if authentication: 

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

178 

179 connectors.extend( 

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

181 ) 

182 

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

184 

185 def is_disconnect( 

186 self, 

187 e: Exception, 

188 connection: Optional[ 

189 Union[pool.PoolProxiedConnection, interfaces.DBAPIConnection] 

190 ], 

191 cursor: Optional[interfaces.DBAPICursor], 

192 ) -> bool: 

193 if not isinstance(e, self.loaded_dbapi.Error): 

194 return False 

195 

196 if getattr(e, "driver_error", None) in self._disconnect_driver_errors: 

197 return True 

198 

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

200 return ( 

201 "The cursor's connection has been closed." in str(e) 

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

203 or "Driver Error: Operation cannot be performed" in str(e) 

204 ) 

205 

206 if isinstance(e, self.loaded_dbapi.InterfaceError): 

207 return bool( 

208 re.search(r"Cannot .* on (?:a )?closed connection", str(e)) 

209 ) 

210 

211 return False 

212 

213 def retrieve_dbapi_version( 

214 self, dbapi: interfaces.DBAPIModule 

215 ) -> util.VersionInfo: 

216 return util.parse_version_string(getattr(dbapi, "__version__", None)) 

217 

218 def _get_server_version_info(self, connection): 

219 vers = connection.exec_driver_sql("select @@version").scalar() 

220 m = re.match(r"Microsoft .*? - (\d+)\.(\d+)\.(\d+)\.(\d+)", vers) 

221 if m: 

222 return tuple(int(x) for x in m.group(1, 2, 3, 4)) 

223 else: 

224 return None 

225 

226 def get_isolation_level_values( 

227 self, dbapi_connection: interfaces.DBAPIConnection 

228 ) -> List[IsolationLevel]: 

229 return [ 

230 *super().get_isolation_level_values(dbapi_connection), 

231 "AUTOCOMMIT", 

232 ] 

233 

234 def set_isolation_level( 

235 self, 

236 dbapi_connection: interfaces.DBAPIConnection, 

237 level: IsolationLevel, 

238 ) -> None: 

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) 

249 

250 

251dialect = MSDialect_mssqlpython