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

1# dialects/mysql/pymysql.py 

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

42 

43MySQL-Python Compatibility 

44-------------------------- 

45 

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

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

48to the pymysql driver as well. 

49 

50""" # noqa 

51 

52from .mysqldb import MySQLDialect_mysqldb 

53from ...util import langhelpers 

54from ...util import py3k 

55 

56 

57class MySQLDialect_pymysql(MySQLDialect_mysqldb): 

58 driver = "pymysql" 

59 supports_statement_cache = True 

60 

61 description_encoding = None 

62 

63 # generally, these two values should be both True 

64 # or both False. PyMySQL unicode tests pass all the way back 

65 # to 0.4 either way. See [ticket:3337] 

66 supports_unicode_statements = True 

67 supports_unicode_binds = True 

68 

69 @langhelpers.memoized_property 

70 def supports_server_side_cursors(self): 

71 try: 

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

73 self._sscursor = cursors.SSCursor 

74 return True 

75 except (ImportError, AttributeError): 

76 return False 

77 

78 @classmethod 

79 def dbapi(cls): 

80 return __import__("pymysql") 

81 

82 @langhelpers.memoized_property 

83 def _send_false_to_ping(self): 

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

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

86 

87 See #10492 and 

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

89 for background. 

90 

91 """ # noqa: E501 

92 

93 try: 

94 Connection = __import__( 

95 "pymysql.connections" 

96 ).connections.Connection 

97 except (ImportError, AttributeError): 

98 return True 

99 else: 

100 insp = langhelpers.get_callable_argspec(Connection.ping) 

101 try: 

102 reconnect_arg = insp.args[1] 

103 except IndexError: 

104 return False 

105 else: 

106 return reconnect_arg == "reconnect" and ( 

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

108 ) 

109 

110 def _ping_impl(self, dbapi_connection): 

111 if self._send_false_to_ping: 

112 dbapi_connection.ping(False) 

113 else: 

114 dbapi_connection.ping() 

115 

116 return True 

117 

118 def create_connect_args(self, url, _translate_args=None): 

119 if _translate_args is None: 

120 _translate_args = dict(username="user") 

121 return super(MySQLDialect_pymysql, self).create_connect_args( 

122 url, _translate_args=_translate_args 

123 ) 

124 

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

126 if super(MySQLDialect_pymysql, self).is_disconnect( 

127 e, connection, cursor 

128 ): 

129 return True 

130 elif isinstance(e, self.dbapi.Error): 

131 str_e = str(e).lower() 

132 return ( 

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

134 ) 

135 else: 

136 return False 

137 

138 if py3k: 

139 

140 def _extract_error_code(self, exception): 

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

142 exception = exception.args[0] 

143 return exception.args[0] 

144 

145 

146dialect = MySQLDialect_pymysql