Revert "Merge "add aiomysql support""

This reverts commit 23343f87f3297ad31d7315ac0e5312db10ef7592, reversing
changes made to c5831b1abd98c46ef7eab7ee82ead18756aea112.

The crashes that occur in jenkins have not been solved and are
now impacting master.   I am not able to reproduce the failure,
including running on the CI machines directly, and a few runs
where I sat there for 20 minutes and watched, it didn't happen.
it is the ultimate heisenbug.

Additionally, there's a reference to "arraysize" that doesn't
exist in fetchmany() and there seem to be no tests that exercise
this for any DBAPI which is also a major bug to be fixed.

References: #5747

6 files changed, 10 insertions, 306 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/__init__.py b/lib/sqlalchemy/dialects/mysql/__init__.py
index c6781c168..9fdc96f6f 100644
--- a/lib/sqlalchemy/dialects/mysql/__init__.py
+++ b/lib/sqlalchemy/dialects/mysql/__init__.py
@@ -49,10 +49,6 @@ from .base import VARCHAR
 from .base import YEAR
 from .dml import Insert
 from .dml import insert
-from ...util import compat
-
-if compat.py3k:
-    from . import aiomysql  # noqa
 
 
 # default dialect
diff --git a/lib/sqlalchemy/dialects/mysql/aiomysql.py b/lib/sqlalchemy/dialects/mysql/aiomysql.py
deleted file mode 100644
index 2eabb91e4..000000000
--- a/lib/sqlalchemy/dialects/mysql/aiomysql.py
+++ /dev/null
@@ -1,270 +0,0 @@
-# mysql/aiomysql.py
-# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors <see AUTHORS
-# file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: http://www.opensource.org/licenses/mit-license.php
-r"""
-.. dialect:: mysql+aiomysql
-    :name: aiomysql
-    :dbapi: aiomysql
-    :connectstring: mysql+aiomysql://user:password@host:port/dbname[?key=value&key=value...]
-    :url: https://github.com/aio-libs/aiomysql
-
-The aiomysql dialect is SQLAlchemy's second Python asyncio dialect.
-
-Using a special asyncio mediation layer, the aiomysql dialect is usable
-as the backend for the :ref:`SQLAlchemy asyncio <asyncio_toplevel>`
-extension package.
-
-This dialect should normally be used only with the
-:func:`_asyncio.create_async_engine` engine creation function::
-
-    from sqlalchemy.ext.asyncio import create_async_engine
-    engine = create_async_engine("mysql+aiomysql://user:pass@hostname/dbname")
-
-Unicode
--------
-
-Please see :ref:`mysql_unicode` for current recommendations on unicode
-handling.
-
-
-"""  # noqa
-
-from .pymysql import MySQLDialect_pymysql
-from ... import pool
-from ...util.concurrency import await_fallback
-from ...util.concurrency import await_only
-
-
-class AsyncAdapt_aiomysql_cursor:
-    server_side = False
-
-    def __init__(self, adapt_connection):
-        self._adapt_connection = adapt_connection
-        self._connection = adapt_connection._connection
-        self.await_ = adapt_connection.await_
-
-        cursor = self._connection.cursor()
-
-        # see https://github.com/aio-libs/aiomysql/issues/543
-        self._cursor = self.await_(cursor.__aenter__())
-        self._rows = []
-
-    @property
-    def description(self):
-        return self._cursor.description
-
-    @property
-    def rowcount(self):
-        return self._cursor.rowcount
-
-    @property
-    def lastrowid(self):
-        return self._cursor.lastrowid
-
-    def close(self):
-        self._rows[:] = []
-
-    def execute(self, operation, parameters=None):
-        if parameters is None:
-            result = self.await_(self._cursor.execute(operation))
-        else:
-            result = self.await_(self._cursor.execute(operation, parameters))
-
-        if not self.server_side:
-            # aiomysql has a "fake" async result, so we have to pull it out
-            # of that here since our default result is not async.
-            # we could just as easily grab "_rows" here and be done with it
-            # but this is safer.
-            self._rows = list(self.await_(self._cursor.fetchall()))
-        return result
-
-    def executemany(self, operation, seq_of_parameters):
-        return self.await_(
-            self._cursor.executemany(operation, seq_of_parameters)
-        )
-
-    def setinputsizes(self, *inputsizes):
-        pass
-
-    def __iter__(self):
-        while self._rows:
-            yield self._rows.pop(0)
-
-    def fetchone(self):
-        if self._rows:
-            return self._rows.pop(0)
-        else:
-            return None
-
-    def fetchmany(self, size=None):
-        if size is None:
-            size = self.arraysize
-
-        retval = self._rows[0:size]
-        self._rows[:] = self._rows[size:]
-        return retval
-
-    def fetchall(self):
-        retval = self._rows[:]
-        self._rows[:] = []
-        return retval
-
-
-class AsyncAdapt_aiomysql_ss_cursor(AsyncAdapt_aiomysql_cursor):
-
-    server_side = True
-
-    def __init__(self, adapt_connection):
-        self._adapt_connection = adapt_connection
-        self._connection = adapt_connection._connection
-        self.await_ = adapt_connection.await_
-
-        cursor = self._connection.cursor(
-            adapt_connection.dbapi.aiomysql.SSCursor
-        )
-
-        self._cursor = self.await_(cursor.__aenter__())
-
-    def close(self):
-        if self._cursor is not None:
-            self.await_(self._cursor.close())
-            self._cursor = None
-
-    def fetchone(self):
-        return self.await_(self._cursor.fetchone())
-
-    def fetchmany(self, size=None):
-        return self.await_(self._cursor.fetchmany(size=size))
-
-    def fetchall(self):
-        return self.await_(self._cursor.fetchall())
-
-
-class AsyncAdapt_aiomysql_connection:
-    await_ = staticmethod(await_only)
-
-    def __init__(self, dbapi, connection):
-        self.dbapi = dbapi
-        self._connection = connection
-
-    def ping(self, reconnect):
-        return self.await_(self._connection.ping(reconnect))
-
-    def character_set_name(self):
-        return self._connection.character_set_name()
-
-    def autocommit(self, value):
-        self.await_(self._connection.autocommit(value))
-
-    def cursor(self, server_side=False):
-        if server_side:
-            return AsyncAdapt_aiomysql_ss_cursor(self)
-        else:
-            return AsyncAdapt_aiomysql_cursor(self)
-
-    def rollback(self):
-        self.await_(self._connection.rollback())
-
-    def commit(self):
-        self.await_(self._connection.commit())
-
-    def close(self):
-        # it's not awaitable.
-        self._connection.close()
-
-
-class AsyncAdaptFallback_aiomysql_connection(AsyncAdapt_aiomysql_connection):
-    __slots__ = ()
-
-    await_ = staticmethod(await_fallback)
-
-
-class AsyncAdapt_aiomysql_dbapi:
-    def __init__(self, aiomysql, pymysql):
-        self.aiomysql = aiomysql
-        self.pymysql = pymysql
-        self.paramstyle = "format"
-        self._init_dbapi_attributes()
-
-    def _init_dbapi_attributes(self):
-        for name in (
-            "Warning",
-            "Error",
-            "InterfaceError",
-            "DataError",
-            "DatabaseError",
-            "OperationalError",
-            "InterfaceError",
-            "IntegrityError",
-            "ProgrammingError",
-            "InternalError",
-            "NotSupportedError",
-        ):
-            setattr(self, name, getattr(self.aiomysql, name))
-
-        for name in (
-            "NUMBER",
-            "STRING",
-            "DATETIME",
-            "BINARY",
-            "TIMESTAMP",
-            "Binary",
-        ):
-            setattr(self, name, getattr(self.pymysql, name))
-
-    def connect(self, *arg, **kw):
-        async_fallback = kw.pop("async_fallback", False)
-
-        if async_fallback:
-            return AsyncAdaptFallback_aiomysql_connection(
-                self,
-                await_fallback(self.aiomysql.connect(*arg, **kw)),
-            )
-        else:
-            return AsyncAdapt_aiomysql_connection(
-                self,
-                await_only(self.aiomysql.connect(*arg, **kw)),
-            )
-
-
-class MySQLDialect_aiomysql(MySQLDialect_pymysql):
-    driver = "aiomysql"
-
-    supports_server_side_cursors = True
-    _sscursor = AsyncAdapt_aiomysql_ss_cursor
-
-    @classmethod
-    def dbapi(cls):
-        return AsyncAdapt_aiomysql_dbapi(
-            __import__("aiomysql"), __import__("pymysql")
-        )
-
-    @classmethod
-    def get_pool_class(self, url):
-        return pool.AsyncAdaptedQueuePool
-
-    def create_connect_args(self, url):
-        args, kw = super(MySQLDialect_aiomysql, self).create_connect_args(url)
-        if "passwd" in kw:
-            kw["password"] = kw.pop("passwd")
-        return args, kw
-
-    def is_disconnect(self, e, connection, cursor):
-        if super(MySQLDialect_aiomysql, self).is_disconnect(
-            e, connection, cursor
-        ):
-            return True
-        else:
-            str_e = str(e).lower()
-            return "not connected" in str_e
-
-    def _found_rows_client_flag(self):
-        from pymysql.constants import CLIENT
-
-        return CLIENT.FOUND_ROWS
-
-
-dialect = MySQLDialect_aiomysql
diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index 605407f46..b20e061fb 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -211,25 +211,16 @@ class MySQLDialect_mysqldb(MySQLDialect):
         # FOUND_ROWS must be set in CLIENT_FLAGS to enable
         # supports_sane_rowcount.
         client_flag = opts.get("client_flag", 0)
-
-        client_flag_found_rows = self._found_rows_client_flag()
-        if client_flag_found_rows is not None:
-            client_flag |= client_flag_found_rows
-            opts["client_flag"] = client_flag
-        return [[], opts]
-
-    def _found_rows_client_flag(self):
         if self.dbapi is not None:
             try:
                 CLIENT_FLAGS = __import__(
                     self.dbapi.__name__ + ".constants.CLIENT"
                 ).constants.CLIENT
+                client_flag |= CLIENT_FLAGS.FOUND_ROWS
             except (AttributeError, ImportError):
-                return None
-            else:
-                return CLIENT_FLAGS.FOUND_ROWS
-        else:
-            return None
+                self.supports_sane_rowcount = False
+            opts["client_flag"] = client_flag
+        return [[], opts]
 
     def _extract_error_code(self, exception):
         return exception.args[0]
diff --git a/lib/sqlalchemy/ext/asyncio/result.py b/lib/sqlalchemy/ext/asyncio/result.py
index 9c7e0420f..7f8a707d5 100644
--- a/lib/sqlalchemy/ext/asyncio/result.py
+++ b/lib/sqlalchemy/ext/asyncio/result.py
@@ -17,14 +17,7 @@ if util.TYPE_CHECKING:
     from ...engine.result import Row
 
 
-class AsyncCommon(FilterResult):
-    async def close(self):
-        """Close this result."""
-
-        await greenlet_spawn(self._real_result.close)
-
-
-class AsyncResult(AsyncCommon):
+class AsyncResult(FilterResult):
     """An asyncio wrapper around a :class:`_result.Result` object.
 
     The :class:`_asyncio.AsyncResult` only applies to statement executions that
@@ -377,7 +370,7 @@ class AsyncResult(AsyncCommon):
         return AsyncMappingResult(self._real_result)
 
 
-class AsyncScalarResult(AsyncCommon):
+class AsyncScalarResult(FilterResult):
     """A wrapper for a :class:`_asyncio.AsyncResult` that returns scalar values
     rather than :class:`_row.Row` values.
 
@@ -507,7 +500,7 @@ class AsyncScalarResult(AsyncCommon):
         return await greenlet_spawn(self._only_one_row, True, True, False)
 
 
-class AsyncMappingResult(AsyncCommon):
+class AsyncMappingResult(FilterResult):
     """A wrapper for a :class:`_asyncio.AsyncResult` that returns dictionary values
     rather than :class:`_engine.Row` values.
 
diff --git a/lib/sqlalchemy/testing/suite/test_results.py b/lib/sqlalchemy/testing/suite/test_results.py
index f31c7c137..9484d41d0 100644
--- a/lib/sqlalchemy/testing/suite/test_results.py
+++ b/lib/sqlalchemy/testing/suite/test_results.py
@@ -114,8 +114,9 @@ class RowFetchTest(fixtures.TablesTest):
 class PercentSchemaNamesTest(fixtures.TablesTest):
     """tests using percent signs, spaces in table and column names.
 
-    This didn't work for PostgreSQL / MySQL drivers for a long time
-    but is now supported.
+    This is a very fringe use case, doesn't work for MySQL
+    or PostgreSQL.  the requirement, "percent_schema_names",
+    is marked "skip" by default.
 
     """
 
@@ -232,8 +233,6 @@ class ServerSideCursorsTest(
         elif self.engine.dialect.driver == "pymysql":
             sscursor = __import__("pymysql.cursors").cursors.SSCursor
             return isinstance(cursor, sscursor)
-        elif self.engine.dialect.driver == "aiomysql":
-            return cursor.server_side
         elif self.engine.dialect.driver == "mysqldb":
             sscursor = __import__("MySQLdb.cursors").cursors.SSCursor
             return isinstance(cursor, sscursor)
diff --git a/lib/sqlalchemy/testing/warnings.py b/lib/sqlalchemy/testing/warnings.py
index 34a968aff..b230bad6f 100644
--- a/lib/sqlalchemy/testing/warnings.py
+++ b/lib/sqlalchemy/testing/warnings.py
@@ -30,11 +30,6 @@ def setup_filters():
     warnings.filterwarnings(
         "ignore", category=DeprecationWarning, message=".*inspect.get.*argspec"
     )
-    warnings.filterwarnings(
-        "ignore",
-        category=DeprecationWarning,
-        message="The loop argument is deprecated",
-    )
 
     # ignore things that are deprecated *as of* 2.0 :)
     warnings.filterwarnings(