--- /dev/null
+.. change::
+ :tags: bug, mysql
+ :tickets: 13306
+
+ Fixed issue in aiomysql and asyncmy dialects that appears as of using
+ pymysql 1.2.0; the dialects were not properly taking into account logic
+ that detects the argument signature of pymysql's ``ping()`` method which
+ was added as part of :ticket:`10492`.
+
+
+
from typing import TYPE_CHECKING
from typing import Union
+from .pymysql import _connection_ping_reconnects_true
from .pymysql import MySQLDialect_pymysql
from ...connectors.asyncio import AsyncAdapt_dbapi_connection
from ...connectors.asyncio import AsyncAdapt_dbapi_cursor
from ...connectors.asyncio import AsyncAdapt_dbapi_module
from ...connectors.asyncio import AsyncAdapt_dbapi_ss_cursor
from ...connectors.asyncio import AsyncAdapt_terminate
+from ...util import langhelpers
from ...util.concurrency import await_
if TYPE_CHECKING:
_cursor_cls = AsyncAdapt_aiomysql_cursor
_ss_cursor_cls = AsyncAdapt_aiomysql_ss_cursor
- def ping(self, reconnect: bool) -> None:
+ def ping(self, reconnect: bool = False) -> None:
assert not reconnect
- await_(self._connection.ping(reconnect))
+
+ if self.dbapi._send_false_to_ping:
+ await_(self._connection.ping(reconnect=False))
+ else:
+ await_(self._connection.ping())
def character_set_name(self) -> Optional[str]:
return self._connection.character_set_name() # type: ignore[no-any-return] # noqa: E501
)
)
+ @langhelpers.memoized_property
+ def _send_false_to_ping(self) -> bool:
+ """determine if aiomysql has deprecated, changed the default of,
+ or removed the 'reconnect' argument of connection.ping().
+
+ See #13306 and #10492
+
+ """ # noqa: E501
+
+ try:
+ Connection = __import__(
+ "aiomysql.connection"
+ ).connection.Connection
+ except (ImportError, AttributeError):
+ return True
+ else:
+ return _connection_ping_reconnects_true(Connection)
+
def _init_cursors_subclasses(
self,
) -> tuple[AsyncIODBAPICursor, AsyncIODBAPICursor]:
from typing import TYPE_CHECKING
from typing import Union
+from .pymysql import _connection_ping_reconnects_true
from .pymysql import MySQLDialect_pymysql
from ... import util
from ...connectors.asyncio import AsyncAdapt_dbapi_connection
from ...connectors.asyncio import AsyncAdapt_dbapi_module
from ...connectors.asyncio import AsyncAdapt_dbapi_ss_cursor
from ...connectors.asyncio import AsyncAdapt_terminate
+from ...util import langhelpers
from ...util.concurrency import await_
if TYPE_CHECKING:
if isinstance(error, AttributeError):
raise dbapi.InternalError(
"network operation failed due to asyncmy attribute error"
- )
+ ) from error
raise error
- def ping(self, reconnect: bool) -> None:
+ def ping(self, reconnect: bool = False) -> None:
assert not reconnect
return await_(self._do_ping())
async def _do_ping(self) -> None:
try:
async with self._execute_mutex:
- await self._connection.ping(False)
+ if self.dbapi._send_false_to_ping:
+ await self._connection.ping(reconnect=False)
+ else:
+ await self._connection.ping()
except Exception as error:
self._handle_exception(error)
)
)
+ @langhelpers.memoized_property
+ def _send_false_to_ping(self) -> bool:
+ """determine if asyncmy has deprecated, changed the default of,
+ or removed the 'reconnect' argument of connection.ping().
+
+ See #13306 and #10492
+
+ """ # noqa: E501
+
+ try:
+ Connection = __import__("asyncmy.connection").connection.Connection
+ except (ImportError, AttributeError):
+ return True
+ else:
+ return _connection_ping_reconnects_true(Connection)
+
class MySQLDialect_asyncmy(MySQLDialect_pymysql):
driver = "asyncmy"
from typing import Any
from typing import Literal
from typing import Optional
+from typing import Type
from typing import TYPE_CHECKING
from typing import Union
from ...engine.url import URL
+def _connection_ping_reconnects_true(connection_cls: Type[Any]) -> bool:
+ """Given a Connection class like pymysql.Connection, aiomysql.Connection,
+ asyncmy.Connection, inspect the ping() method and determine if it
+ has a "reconnect" parameter that either defaults to True, or is positional.
+
+ a return value of True here means that when we call ``connection.ping()``,
+ we **must** pass `reconnect=False`. a return value of False means that
+ we should call ``connection.ping()`` with **no arguments**.
+
+ This routine originates from issue #10492 for pymysql, however arg
+ signature mismatches in aiomysql/asyncmy tracked by issue #13306
+ necessitated a more open ended function.
+
+ """
+ insp = langhelpers.get_callable_argspec(connection_cls.ping)
+ try:
+ reconnect_arg = insp.args[1]
+ except IndexError:
+ return False
+ else:
+ return reconnect_arg == "reconnect" and (
+ not insp.defaults or insp.defaults[0] is not False
+ )
+
+
class MySQLDialect_pymysql(MySQLDialect_mysqldb):
driver = "pymysql"
supports_statement_cache = True
https://github.com/PyMySQL/mysqlclient/discussions/651#discussioncomment-7308971
for background.
+ Revised as part of #13306
+
""" # noqa: E501
try:
except (ImportError, AttributeError):
return True
else:
- insp = langhelpers.get_callable_argspec(Connection.ping)
- try:
- reconnect_arg = insp.args[1]
- except IndexError:
- return False
- else:
- return reconnect_arg == "reconnect" and (
- not insp.defaults or insp.defaults[0] is not False
- )
+ return _connection_ping_reconnects_true(Connection)
def do_ping(self, dbapi_connection: DBAPIConnection) -> Literal[True]:
if self._send_false_to_ping:
if inspect.ismethod(func):
func = func.__func__
- if not inspect.isfunction(func):
+ if not inspect.isfunction(func) and not hasattr(func, "__code__"):
raise TypeError(f"{func!r} is not a Python function")
co = func.__code__
"""
if inspect.isbuiltin(fn):
raise TypeError("Can't inspect builtin: %s" % fn)
- elif inspect.isfunction(fn):
+ elif inspect.isfunction(fn) or (
+ hasattr(fn, "__code__")
+ and not inspect.isclass(fn)
+ and not inspect.ismethod(fn)
+ ):
if _is_init and no_self:
spec = compat.inspect_getfullargspec(fn)
return compat.FullArgSpec(
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy.dialects import mysql
+from sqlalchemy.dialects.mysql.pymysql import _connection_ping_reconnects_true
from sqlalchemy.engine.url import make_url
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import AssertsCompiledSQL
from .test_compiler import ReservedWordFixture
+class DBAPIPingReconnectTest(fixtures.TestBase):
+ __backend__ = True
+ __only_on__ = ("+pymysql", "+asyncmy", "+aiomysql")
+
+ @testing.combinations(
+ (
+ type("MyConnection", (object,), {"ping": lambda self: None}),
+ False,
+ ),
+ (
+ type(
+ "MyConnection",
+ (object,),
+ {"ping": lambda self, reconnect=True: None},
+ ),
+ True,
+ ),
+ (
+ type(
+ "MyConnection",
+ (object,),
+ {"ping": lambda self, reconnect: None},
+ ),
+ True,
+ ),
+ (
+ type(
+ "MyConnection",
+ (object,),
+ {"ping": lambda self, reconnect=False: None},
+ ),
+ False,
+ ),
+ )
+ def test_ping_send_false_inspect(self, connection, expected):
+ is_(_connection_ping_reconnects_true(connection), expected)
+
+ def test_do_ping(self):
+ """this is a copy of the same test in the generic suite"""
+ with testing.db.connect() as conn:
+ is_true(
+ testing.db.dialect.do_ping(conn.connection.dbapi_connection)
+ )
+
+ def test_do_ping_disconnected(self):
+ with testing.db.connect() as conn:
+ driver_connection = conn.connection.driver_connection
+ dbapi = testing.db.dialect.loaded_dbapi
+
+ # FIXME: detach seems to set driver_connection to None
+ conn.detach()
+
+ driver_connection.close()
+
+ with expect_raises(dbapi.Error):
+ testing.db.dialect.do_ping(conn.connection.dbapi_connection)
+
+
class BackendDialectTest(
ReservedWordFixture, fixtures.TestBase, AssertsCompiledSQL
):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
