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

1# dialects/mysql/pymysql.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 

8r""" 

9 

10.. dialect:: mysql+pymysql 

11 :name: PyMySQL 

12 :dbapi: pymysql 

13 :connectstring: mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] 

14 :url: https://pymysql.readthedocs.io/ 

15 

16Unicode 

17------- 

18 

19Please see :ref:`mysql_unicode` for current recommendations on unicode 

20handling. 

21 

22.. _pymysql_ssl: 

23 

24SSL Connections 

25------------------ 

26 

27The PyMySQL DBAPI accepts the same SSL arguments as that of MySQLdb, 

28described at :ref:`mysqldb_ssl`. See that section for additional examples. 

29 

30If the server uses an automatically-generated certificate that is self-signed 

31or does not match the host name (as seen from the client), it may also be 

32necessary to indicate ``ssl_check_hostname=false`` in PyMySQL:: 

33 

34 connection_uri = ( 

35 "mysql+pymysql://scott:tiger@192.168.0.134/test" 

36 "?ssl_ca=/home/gord/client-ssl/ca.pem" 

37 "&ssl_cert=/home/gord/client-ssl/client-cert.pem" 

38 "&ssl_key=/home/gord/client-ssl/client-key.pem" 

39 "&ssl_check_hostname=false" 

40 ) 

41 

42MySQL-Python Compatibility 

43-------------------------- 

44 

45The pymysql DBAPI is a pure Python port of the MySQL-python (MySQLdb) driver, 

46and targets 100% compatibility. Most behavioral notes for MySQL-python apply 

47to the pymysql driver as well. 

48 

49""" # noqa 

50from __future__ import annotations 

51 

52from typing import Any 

53from typing import Dict 

54from typing import Optional 

55from typing import TYPE_CHECKING 

56from typing import Union 

57 

58from .mysqldb import MySQLDialect_mysqldb 

59from ...util import langhelpers 

60from ...util.typing import Literal 

61 

62if TYPE_CHECKING: 

63 

64 from ...engine.interfaces import ConnectArgsType 

65 from ...engine.interfaces import DBAPIConnection 

66 from ...engine.interfaces import DBAPICursor 

67 from ...engine.interfaces import DBAPIModule 

68 from ...engine.interfaces import PoolProxiedConnection 

69 from ...engine.url import URL 

70 

71 

72class MySQLDialect_pymysql(MySQLDialect_mysqldb): 

73 driver = "pymysql" 

74 supports_statement_cache = True 

75 

76 description_encoding = None 

77 

78 @langhelpers.memoized_property 

79 def supports_server_side_cursors(self) -> bool: 

80 try: 

81 cursors = __import__("pymysql.cursors").cursors 

82 self._sscursor = cursors.SSCursor 

83 return True 

84 except (ImportError, AttributeError): 

85 return False 

86 

87 @classmethod 

88 def import_dbapi(cls) -> DBAPIModule: 

89 return __import__("pymysql") 

90 

91 @langhelpers.memoized_property 

92 def _send_false_to_ping(self) -> bool: 

93 """determine if pymysql has deprecated, changed the default of, 

94 or removed the 'reconnect' argument of connection.ping(). 

95 

96 See #10492 and 

97 https://github.com/PyMySQL/mysqlclient/discussions/651#discussioncomment-7308971 

98 for background. 

99 

100 """ # noqa: E501 

101 

102 try: 

103 Connection = __import__( 

104 "pymysql.connections" 

105 ).connections.Connection 

106 except (ImportError, AttributeError): 

107 return True 

108 else: 

109 insp = langhelpers.get_callable_argspec(Connection.ping) 

110 try: 

111 reconnect_arg = insp.args[1] 

112 except IndexError: 

113 return False 

114 else: 

115 return reconnect_arg == "reconnect" and ( 

116 not insp.defaults or insp.defaults[0] is not False 

117 ) 

118 

119 def do_ping(self, dbapi_connection: DBAPIConnection) -> Literal[True]: 

120 if self._send_false_to_ping: 

121 dbapi_connection.ping(False) 

122 else: 

123 dbapi_connection.ping() 

124 

125 return True 

126 

127 def create_connect_args( 

128 self, url: URL, _translate_args: Optional[Dict[str, Any]] = None 

129 ) -> ConnectArgsType: 

130 if _translate_args is None: 

131 _translate_args = dict(username="user") 

132 return super().create_connect_args( 

133 url, _translate_args=_translate_args 

134 ) 

135 

136 def is_disconnect( 

137 self, 

138 e: DBAPIModule.Error, 

139 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]], 

140 cursor: Optional[DBAPICursor], 

141 ) -> bool: 

142 if super().is_disconnect(e, connection, cursor): 

143 return True 

144 elif isinstance(e, self.loaded_dbapi.Error): 

145 str_e = str(e).lower() 

146 return ( 

147 "already closed" in str_e or "connection was killed" in str_e 

148 ) 

149 else: 

150 return False 

151 

152 def _extract_error_code(self, exception: BaseException) -> Any: 

153 if isinstance(exception.args[0], Exception): 

154 exception = exception.args[0] 

155 return exception.args[0] 

156 

157 

158dialect = MySQLDialect_pymysql