Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1# connectors/mxodbc.py 

2# Copyright (C) 2005-2020 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: http://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 http://www.egenix.com/ 

19 

20""" 

21 

22import re 

23import sys 

24import warnings 

25 

26from . import Connector 

27 

28 

29class MxODBCConnector(Connector): 

30 driver = "mxodbc" 

31 

32 supports_sane_multi_rowcount = False 

33 supports_unicode_statements = True 

34 supports_unicode_binds = True 

35 

36 supports_native_decimal = True 

37 

38 @classmethod 

39 def dbapi(cls): 

40 # this classmethod will normally be replaced by an instance 

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

42 cls._load_mx_exceptions() 

43 platform = sys.platform 

44 if platform == "win32": 

45 from mx.ODBC import Windows as Module 

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

47 elif "linux" in platform: 

48 from mx.ODBC import unixODBC as Module 

49 elif platform == "darwin": 

50 from mx.ODBC import iODBC as Module 

51 else: 

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

53 return Module 

54 

55 @classmethod 

56 def _load_mx_exceptions(cls): 

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

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

59 to avoid requiring all SQLAlchemy users to install mxODBC. 

60 """ 

61 global InterfaceError, ProgrammingError 

62 from mx.ODBC import InterfaceError 

63 from mx.ODBC import ProgrammingError 

64 

65 def on_connect(self): 

66 def connect(conn): 

67 conn.stringformat = self.dbapi.MIXED_STRINGFORMAT 

68 conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT 

69 conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT 

70 conn.errorhandler = self._error_handler() 

71 

72 return connect 

73 

74 def _error_handler(self): 

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

76 emit Python standard warnings. 

77 """ 

78 from mx.ODBC.Error import Warning as MxOdbcWarning 

79 

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

81 if issubclass(errorclass, MxOdbcWarning): 

82 errorclass.__bases__ = (Warning,) 

83 warnings.warn( 

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

85 ) 

86 else: 

87 raise errorclass(errorvalue) 

88 

89 return error_handler 

90 

91 def create_connect_args(self, url): 

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

93 

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

95 

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

97 clear_auto_commit=1, errorhandler=None) 

98 

99 This method translates the values in the provided uri 

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

101 

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

103 not be populated. 

104 

105 """ 

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

107 opts.update(url.query) 

108 args = opts.pop("host") 

109 opts.pop("port", None) 

110 opts.pop("database", None) 

111 return (args,), opts 

112 

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

114 # TODO: eGenix recommends checking connection.closed here 

115 # Does that detect dropped connections ? 

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

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

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

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

120 else: 

121 return False 

122 

123 def _get_server_version_info(self, connection): 

124 # eGenix suggests using conn.dbms_version instead 

125 # of what we're doing here 

126 dbapi_con = connection.connection 

127 version = [] 

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

129 # 18 == pyodbc.SQL_DBMS_VER 

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

131 try: 

132 version.append(int(n)) 

133 except ValueError: 

134 version.append(n) 

135 return tuple(version) 

136 

137 def _get_direct(self, context): 

138 if context: 

139 native_odbc_execute = context.execution_options.get( 

140 "native_odbc_execute", "auto" 

141 ) 

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

143 # compatible especially with SQL Server 

144 return False if native_odbc_execute is True else True 

145 else: 

146 return True 

147 

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

149 cursor.executemany( 

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

151 ) 

152 

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

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