codekingpro/portable-devtools

1# dialects/mysql/pymysql.py
2# Copyright (C) 2005-2026 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
159