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/pyodbc.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 

8import re 

9 

10from . import Connector 

11from .. import util 

12 

13 

14class PyODBCConnector(Connector): 

15 driver = "pyodbc" 

16 

17 # this is no longer False for pyodbc in general 

18 supports_sane_rowcount_returning = True 

19 supports_sane_multi_rowcount = False 

20 

21 supports_unicode_statements = True 

22 supports_unicode_binds = True 

23 

24 supports_native_decimal = True 

25 default_paramstyle = "named" 

26 

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

28 # hold the desired driver name 

29 pyodbc_driver_name = None 

30 

31 def __init__(self, supports_unicode_binds=None, **kw): 

32 super(PyODBCConnector, self).__init__(**kw) 

33 if supports_unicode_binds is not None: 

34 self.supports_unicode_binds = supports_unicode_binds 

35 

36 @classmethod 

37 def dbapi(cls): 

38 return __import__("pyodbc") 

39 

40 def create_connect_args(self, url): 

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

42 opts.update(url.query) 

43 

44 keys = opts 

45 

46 query = url.query 

47 

48 connect_args = {} 

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

50 if param in keys: 

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

52 

53 if "odbc_connect" in keys: 

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

55 else: 

56 

57 def check_quote(token): 

58 if ";" in str(token): 

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

60 return token 

61 

62 keys = dict((k, check_quote(v)) for k, v in keys.items()) 

63 

64 dsn_connection = "dsn" in keys or ( 

65 "host" in keys and "database" not in keys 

66 ) 

67 if dsn_connection: 

68 connectors = [ 

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

70 ] 

71 else: 

72 port = "" 

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

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

75 

76 connectors = [] 

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

78 if driver is None and keys: 

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

80 util.warn( 

81 "No driver name specified; " 

82 "this is expected by PyODBC when using " 

83 "DSN-less connections" 

84 ) 

85 else: 

86 connectors.append("DRIVER={%s}" % driver) 

87 

88 connectors.extend( 

89 [ 

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

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

92 ] 

93 ) 

94 

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

96 if user: 

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

98 connectors.append("PWD=%s" % keys.pop("password", "")) 

99 else: 

100 connectors.append("Trusted_Connection=Yes") 

101 

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

103 # convert textual data from your database encoding to your 

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

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

106 if "odbc_autotranslate" in keys: 

107 connectors.append( 

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

109 ) 

110 

111 connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()]) 

112 

113 return [[";".join(connectors)], connect_args] 

114 

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

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

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

118 e 

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

120 else: 

121 return False 

122 

123 # def initialize(self, connection): 

124 # super(PyODBCConnector, self).initialize(connection) 

125 

126 def _dbapi_version(self): 

127 if not self.dbapi: 

128 return () 

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

130 

131 def _parse_dbapi_version(self, vers): 

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

133 if not m: 

134 return () 

135 vers = tuple([int(x) for x in m.group(1).split(".")]) 

136 if m.group(2): 

137 vers += (m.group(2),) 

138 return vers 

139 

140 def _get_server_version_info(self, connection, allow_chars=True): 

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

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

143 # queries. 

144 dbapi_con = connection.connection 

145 version = [] 

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

147 for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)): 

148 try: 

149 version.append(int(n)) 

150 except ValueError: 

151 if allow_chars: 

152 version.append(n) 

153 return tuple(version) 

154 

155 def set_isolation_level(self, connection, level): 

156 # adjust for ConnectionFairy being present 

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

158 # to work properly 

159 if hasattr(connection, "connection"): 

160 connection = connection.connection 

161 

162 if level == "AUTOCOMMIT": 

163 connection.autocommit = True 

164 else: 

165 connection.autocommit = False 

166 super(PyODBCConnector, self).set_isolation_level(connection, level)