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

1# connectors/mxodbc.py 

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

8""" 

9Provide a SQLALchemy connector for the eGenix mxODBC commercial 

10Python adapter for ODBC. This is not a free product, but eGenix 

11provides SQLAlchemy with a license for use in continuous integration 

12testing. 

13 

14This has been tested for use with mxODBC 3.1.2 on SQL Server 2005 

15and 2008, using the SQL Server Native driver. However, it is 

16possible for this to be used on other database platforms. 

17 

18For more info on mxODBC, see https://www.egenix.com/ 

19 

20.. deprecated:: 1.4 The mxODBC DBAPI is deprecated and will be removed 

21 in a future version. Please use one of the supported DBAPIs to 

22 connect to mssql. 

23 

24""" 

25 

26import re 

27import sys 

28import warnings 

29 

30from . import Connector 

31from ..util import warn_deprecated 

32 

33 

34class MxODBCConnector(Connector): 

35 driver = "mxodbc" 

36 

37 supports_sane_multi_rowcount = False 

38 supports_unicode_statements = True 

39 supports_unicode_binds = True 

40 

41 supports_native_decimal = True 

42 

43 @classmethod 

44 def dbapi(cls): 

45 # this classmethod will normally be replaced by an instance 

46 # attribute of the same name, so this is normally only called once. 

47 cls._load_mx_exceptions() 

48 platform = sys.platform 

49 if platform == "win32": 

50 from mx.ODBC import Windows as Module 

51 # this can be the string "linux2", and possibly others 

52 elif "linux" in platform: 

53 from mx.ODBC import unixODBC as Module 

54 elif platform == "darwin": 

55 from mx.ODBC import iODBC as Module 

56 else: 

57 raise ImportError("Unrecognized platform for mxODBC import") 

58 

59 warn_deprecated( 

60 "The mxODBC DBAPI is deprecated and will be removed" 

61 "in a future version. Please use one of the supported DBAPIs to" 

62 "connect to mssql.", 

63 version="1.4", 

64 ) 

65 return Module 

66 

67 @classmethod 

68 def _load_mx_exceptions(cls): 

69 """Import mxODBC exception classes into the module namespace, 

70 as if they had been imported normally. This is done here 

71 to avoid requiring all SQLAlchemy users to install mxODBC. 

72 """ 

73 global InterfaceError, ProgrammingError 

74 from mx.ODBC import InterfaceError 

75 from mx.ODBC import ProgrammingError 

76 

77 def on_connect(self): 

78 def connect(conn): 

79 conn.stringformat = self.dbapi.MIXED_STRINGFORMAT 

80 conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT 

81 conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT 

82 conn.errorhandler = self._error_handler() 

83 

84 return connect 

85 

86 def _error_handler(self): 

87 """Return a handler that adjusts mxODBC's raised Warnings to 

88 emit Python standard warnings. 

89 """ 

90 from mx.ODBC.Error import Warning as MxOdbcWarning 

91 

92 def error_handler(connection, cursor, errorclass, errorvalue): 

93 if issubclass(errorclass, MxOdbcWarning): 

94 errorclass.__bases__ = (Warning,) 

95 warnings.warn( 

96 message=str(errorvalue), category=errorclass, stacklevel=2 

97 ) 

98 else: 

99 raise errorclass(errorvalue) 

100 

101 return error_handler 

102 

103 def create_connect_args(self, url): 

104 r"""Return a tuple of \*args, \**kwargs for creating a connection. 

105 

106 The mxODBC 3.x connection constructor looks like this: 

107 

108 connect(dsn, user='', password='', 

109 clear_auto_commit=1, errorhandler=None) 

110 

111 This method translates the values in the provided URI 

112 into args and kwargs needed to instantiate an mxODBC Connection. 

113 

114 The arg 'errorhandler' is not used by SQLAlchemy and will 

115 not be populated. 

116 

117 """ 

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

119 opts.update(url.query) 

120 args = opts.pop("host") 

121 opts.pop("port", None) 

122 opts.pop("database", None) 

123 return (args,), opts 

124 

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

126 # TODO: eGenix recommends checking connection.closed here 

127 # Does that detect dropped connections ? 

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

129 return "connection already closed" in str(e) 

130 elif isinstance(e, self.dbapi.Error): 

131 return "[08S01]" in str(e) 

132 else: 

133 return False 

134 

135 def _get_server_version_info(self, connection): 

136 # eGenix suggests using conn.dbms_version instead 

137 # of what we're doing here 

138 dbapi_con = connection.connection 

139 version = [] 

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

141 # 18 == pyodbc.SQL_DBMS_VER 

142 for n in r.split(dbapi_con.getinfo(18)[1]): 

143 try: 

144 version.append(int(n)) 

145 except ValueError: 

146 version.append(n) 

147 return tuple(version) 

148 

149 def _get_direct(self, context): 

150 if context: 

151 native_odbc_execute = context.execution_options.get( 

152 "native_odbc_execute", "auto" 

153 ) 

154 # default to direct=True in all cases, is more generally 

155 # compatible especially with SQL Server 

156 return False if native_odbc_execute is True else True 

157 else: 

158 return True 

159 

160 def do_executemany(self, cursor, statement, parameters, context=None): 

161 cursor.executemany( 

162 statement, parameters, direct=self._get_direct(context) 

163 ) 

164 

165 def do_execute(self, cursor, statement, parameters, context=None): 

166 cursor.execute(statement, parameters, direct=self._get_direct(context))