Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/pyodbc.py: 42%

1# mysql/pyodbc.py 

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

8r""" 

9 

10 

11.. dialect:: mysql+pyodbc 

12 :name: PyODBC 

13 :dbapi: pyodbc 

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

15 :url: https://pypi.org/project/pyodbc/ 

16 

17.. note:: 

18 

19 The PyODBC for MySQL dialect is **not tested as part of 

20 SQLAlchemy's continuous integration**. 

21 The recommended MySQL dialects are mysqlclient and PyMySQL. 

22 However, if you want to use the mysql+pyodbc dialect and require 

23 full support for ``utf8mb4`` characters (including supplementary 

24 characters like emoji) be sure to use a current release of 

25 MySQL Connector/ODBC and specify the "ANSI" (**not** "Unicode") 

26 version of the driver in your DSN or connection string. 

27 

28Pass through exact pyodbc connection string:: 

29 

30 import urllib 

31 connection_string = ( 

32 'DRIVER=MySQL ODBC 8.0 ANSI Driver;' 

33 'SERVER=localhost;' 

34 'PORT=3307;' 

35 'DATABASE=mydb;' 

36 'UID=root;' 

37 'PWD=(whatever);' 

38 'charset=utf8mb4;' 

39 ) 

40 params = urllib.parse.quote_plus(connection_string) 

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

42 

43""" # noqa 

44 

45import re 

46 

47from .base import MySQLDialect 

48from .base import MySQLExecutionContext 

49from .types import TIME 

50from ... import exc 

51from ... import util 

52from ...connectors.pyodbc import PyODBCConnector 

53from ...sql.sqltypes import Time 

54 

55 

56class _pyodbcTIME(TIME): 

57 def result_processor(self, dialect, coltype): 

58 def process(value): 

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

60 return value 

61 

62 return process 

63 

64 

65class MySQLExecutionContext_pyodbc(MySQLExecutionContext): 

66 def get_lastrowid(self): 

67 cursor = self.create_cursor() 

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

69 lastrowid = cursor.fetchone()[0] 

70 cursor.close() 

71 return lastrowid 

72 

73 

74class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect): 

75 supports_statement_cache = True 

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

77 supports_unicode_statements = True 

78 execution_ctx_cls = MySQLExecutionContext_pyodbc 

79 

80 pyodbc_driver_name = "MySQL" 

81 

82 def _detect_charset(self, connection): 

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

84 

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

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

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

88 # 

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

90 # this can prefer the driver value. 

91 

92 # set this to None as _fetch_setting attempts to use it (None is OK) 

93 self._connection_charset = None 

94 try: 

95 value = self._fetch_setting(connection, "character_set_client") 

96 if value: 

97 return value 

98 except exc.DBAPIError: 

99 pass 

100 

101 util.warn( 

102 "Could not detect the connection character set. " 

103 "Assuming latin1." 

104 ) 

105 return "latin1" 

106 

107 def _get_server_version_info(self, connection): 

108 return MySQLDialect._get_server_version_info(self, connection) 

109 

110 def _extract_error_code(self, exception): 

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

112 c = m.group(1) 

113 if c: 

114 return int(c) 

115 else: 

116 return None 

117 

118 def on_connect(self): 

119 super_ = super(MySQLDialect_pyodbc, self).on_connect() 

120 

121 def on_connect(conn): 

122 if super_ is not None: 

123 super_(conn) 

124 

125 # declare Unicode encoding for pyodbc as per 

126 # https://github.com/mkleehammer/pyodbc/wiki/Unicode 

127 pyodbc_SQL_CHAR = 1 # pyodbc.SQL_CHAR 

128 pyodbc_SQL_WCHAR = -8 # pyodbc.SQL_WCHAR 

129 conn.setdecoding(pyodbc_SQL_CHAR, encoding="utf-8") 

130 conn.setdecoding(pyodbc_SQL_WCHAR, encoding="utf-8") 

131 conn.setencoding(encoding="utf-8") 

132 

133 return on_connect 

134 

135 

136dialect = MySQLDialect_pyodbc