SQLAlchemy ping no longer detects disconnects before SQLAlchemy 2.0.5 #1199

Problem

Since #1185 (unreleased), AthenaDialect.do_ping() runs the ping itself instead of through DefaultDialect.do_ping().
Before SQLAlchemy 2.0.5, DefaultDialect.do_ping() is also where a failed ping is checked for a disconnect: it catches the DBAPI error, asks is_disconnect() and the handle_error event listeners, and returns False for a disconnect, so the pool discards the connection and opens a new one.
The override drops that, so with SQLAlchemy 2.0.0 to 2.0.4 a ping error that a handle_error listener marks as a disconnect makes the checkout raise instead of reconnecting.
SQLAlchemy 2.0.5 and later handle the error around do_ping() (DefaultDialect._do_ping_w_event()), so they are not affected.

The 3.x backport of #1185 (#1198) has the same problem with SQLAlchemy 1.x, whose DefaultDialect.do_ping() also handles disconnects itself.

Reproduction

No AWS access is needed. A handle_error listener marks every error as a disconnect, and only the ping of the pooled connection fails:

from unittest import mock

import sqlalchemy
from sqlalchemy import event

from pyathena.cursor import Cursor
from pyathena.error import OperationalError

engine = sqlalchemy.create_engine(
    "awsathena+rest://AKID:SECRET@athena.us-west-2.amazonaws.com:443/default"
    "?s3_staging_dir=s3%3A%2F%2Fbucket%2Fpath%2F",
    pool_pre_ping=True,
)
connects = []
event.listen(engine, "connect", lambda dbapi_connection, record: connects.append(dbapi_connection))


@event.listens_for(engine, "handle_error")
def mark_disconnect(context):
    context.is_disconnect = True


failing = [False]


def execute(self, operation, *args, **kwargs):
    if failing[0]:
        failing[0] = False  # only the ping of the stale connection fails
        raise OperationalError("ExpiredTokenException")
    return self


with mock.patch.object(Cursor, "execute", execute):
    engine.raw_connection().close()
    failing[0] = True
    try:
        engine.raw_connection().close()
        print(f"SQLAlchemy {sqlalchemy.__version__}: reconnected, connections opened = {len(connects)}")
    except Exception as e:
        print(f"SQLAlchemy {sqlalchemy.__version__}: checkout raised {type(e).__name__}: {e}")

Results on Python 3.11:

Environment

PyAthena master at 52ff4e00, Python 3.11, SQLAlchemy 2.0.4, 2.0.5, and 2.0.46 (uv run --with sqlalchemy==<version>). The sqlalchemy extra allows sqlalchemy>=2.0.0. Found by an independent review of #1198.

Proposed fix (optional)

Keep #1185's API cursor, but let DefaultDialect.do_ping() run the ping: pass it a connection whose cursor() opens the API cursor (_internal_cursor(Cursor)), so each SQLAlchemy version keeps its own disconnect handling.
Add a regression test that fails when the override handles the ping without DefaultDialect.do_ping(), and check the reproduction above with SQLAlchemy 2.0.4.
Then port the fix to #1198 before the next 3.x release.