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

1# dialects/mysql/pyodbc.py 

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

9r""" 

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 

32 connection_string = ( 

33 "DRIVER=MySQL ODBC 8.0 ANSI Driver;" 

34 "SERVER=localhost;" 

35 "PORT=3307;" 

36 "DATABASE=mydb;" 

37 "UID=root;" 

38 "PWD=(whatever);" 

39 "charset=utf8mb4;" 

40 ) 

41 params = urllib.parse.quote_plus(connection_string) 

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

43 

44""" # noqa 

45from __future__ import annotations 

46 

47import datetime 

48import re 

49from typing import Any 

50from typing import Callable 

51from typing import Optional 

52from typing import Tuple 

53from typing import TYPE_CHECKING 

54from typing import Union 

55 

56from .base import MySQLDialect 

57from .base import MySQLExecutionContext 

58from .types import TIME 

59from ... import exc 

60from ... import util 

61from ...connectors.pyodbc import PyODBCConnector 

62from ...sql.sqltypes import Time 

63 

64if TYPE_CHECKING: 

65 from ...engine import Connection 

66 from ...engine.interfaces import DBAPIConnection 

67 from ...engine.interfaces import Dialect 

68 from ...sql.type_api import _ResultProcessorType 

69 

70 

71class _pyodbcTIME(TIME): 

72 def result_processor( 

73 self, dialect: Dialect, coltype: object 

74 ) -> _ResultProcessorType[datetime.time]: 

75 def process(value: Any) -> Union[datetime.time, None]: 

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

77 return value # type: ignore[no-any-return] 

78 

79 return process 

80 

81 

82class MySQLExecutionContext_pyodbc(MySQLExecutionContext): 

83 def get_lastrowid(self) -> int: 

84 cursor = self.create_cursor() 

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

86 lastrowid = cursor.fetchone()[0] # type: ignore[index] 

87 cursor.close() 

88 return lastrowid # type: ignore[no-any-return] 

89 

90 

91class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect): 

92 supports_statement_cache = True 

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

94 supports_unicode_statements = True 

95 execution_ctx_cls = MySQLExecutionContext_pyodbc 

96 

97 pyodbc_driver_name = "MySQL" 

98 

99 def _detect_charset(self, connection: Connection) -> str: 

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

101 

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

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

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

105 # 

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

107 # this can prefer the driver value. 

108 

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

110 self._connection_charset = None 

111 try: 

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

113 if value: 

114 return value 

115 except exc.DBAPIError: 

116 pass 

117 

118 util.warn( 

119 "Could not detect the connection character set. " 

120 "Assuming latin1." 

121 ) 

122 return "latin1" 

123 

124 def _get_server_version_info( 

125 self, connection: Connection 

126 ) -> Tuple[int, ...]: 

127 return MySQLDialect._get_server_version_info(self, connection) 

128 

129 def _extract_error_code(self, exception: BaseException) -> Optional[int]: 

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

131 if m is None: 

132 return None 

133 c: Optional[str] = m.group(1) 

134 if c: 

135 return int(c) 

136 else: 

137 return None 

138 

139 def on_connect(self) -> Callable[[DBAPIConnection], None]: 

140 super_ = super().on_connect() 

141 

142 def on_connect(conn: DBAPIConnection) -> None: 

143 if super_ is not None: 

144 super_(conn) 

145 

146 # declare Unicode encoding for pyodbc as per 

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

148 pyodbc_SQL_CHAR = 1 # pyodbc.SQL_CHAR 

149 pyodbc_SQL_WCHAR = -8 # pyodbc.SQL_WCHAR 

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

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

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

153 

154 return on_connect 

155 

156 

157dialect = MySQLDialect_pyodbc