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# mysql/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 

8r""" 

9 

10 

11.. dialect:: mysql+pyodbc 

12 :name: PyODBC 

13 :dbapi: pyodbc 

14 :connectstring: mysql+pyodbc://<username>:<password>@<dsnname> 

15 :url: http://pypi.python.org/pypi/pyodbc/ 

16 

17 .. note:: The PyODBC for MySQL dialect is not well supported, and 

18 is subject to unresolved character encoding issues 

19 which exist within the current ODBC drivers available. 

20 (see http://code.google.com/p/pyodbc/issues/detail?id=25). 

21 Other dialects for MySQL are recommended. 

22 

23Pass through exact pyodbc connection string:: 

24 

25 import urllib 

26 connection_string = ( 

27 'DRIVER=MySQL ODBC 8.0 ANSI Driver;' 

28 'SERVER=localhost;' 

29 'PORT=3307;' 

30 'DATABASE=mydb;' 

31 'UID=root;' 

32 'PWD=(whatever);' 

33 'charset=utf8mb4;' 

34 ) 

35 params = urllib.parse.quote_plus(connection_string) 

36 connection_uri = "mysql+pyodbc:///?odbc_connect=%s" % params 

37 

38""" # noqa 

39 

40import re 

41 

42from .base import MySQLDialect 

43from .base import MySQLExecutionContext 

44from .types import TIME 

45from ... import util 

46from ...connectors.pyodbc import PyODBCConnector 

47from ...sql.sqltypes import Time 

48 

49 

50class _pyodbcTIME(TIME): 

51 def result_processor(self, dialect, coltype): 

52 def process(value): 

53 # pyodbc returns a datetime.time object; no need to convert 

54 return value 

55 

56 return process 

57 

58 

59class MySQLExecutionContext_pyodbc(MySQLExecutionContext): 

60 def get_lastrowid(self): 

61 cursor = self.create_cursor() 

62 cursor.execute("SELECT LAST_INSERT_ID()") 

63 lastrowid = cursor.fetchone()[0] 

64 cursor.close() 

65 return lastrowid 

66 

67 

68class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect): 

69 colspecs = util.update_copy(MySQLDialect.colspecs, {Time: _pyodbcTIME}) 

70 supports_unicode_statements = False 

71 execution_ctx_cls = MySQLExecutionContext_pyodbc 

72 

73 pyodbc_driver_name = "MySQL" 

74 

75 def __init__(self, **kw): 

76 # deal with http://code.google.com/p/pyodbc/issues/detail?id=25 

77 kw.setdefault("convert_unicode", True) 

78 super(MySQLDialect_pyodbc, self).__init__(**kw) 

79 

80 def _detect_charset(self, connection): 

81 """Sniff out the character set in use for connection results.""" 

82 

83 # Prefer 'character_set_results' for the current connection over the 

84 # value in the driver. SET NAMES or individual variable SETs will 

85 # change the charset without updating the driver's view of the world. 

86 # 

87 # If it's decided that issuing that sort of SQL leaves you SOL, then 

88 # this can prefer the driver value. 

89 rs = connection.execute("SHOW VARIABLES LIKE 'character_set%%'") 

90 opts = {row[0]: row[1] for row in self._compat_fetchall(rs)} 

91 for key in ("character_set_connection", "character_set"): 

92 if opts.get(key, None): 

93 return opts[key] 

94 

95 util.warn( 

96 "Could not detect the connection character set. " 

97 "Assuming latin1." 

98 ) 

99 return "latin1" 

100 

101 def _extract_error_code(self, exception): 

102 m = re.compile(r"\((\d+)\)").search(str(exception.args)) 

103 c = m.group(1) 

104 if c: 

105 return int(c) 

106 else: 

107 return None 

108 

109 

110dialect = MySQLDialect_pyodbc