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

14 files changed, 14 insertions, 344 deletions

diff --git a/doc/build/changelog/unreleased_14/5747.rst b/doc/build/changelog/unreleased_14/5747.rst
deleted file mode 100644
index 47cf648cd..000000000
--- a/doc/build/changelog/unreleased_14/5747.rst
+++ /dev/null
@@ -1,10 +0,0 @@
-.. change::
-    :tags: feature, mysql
-    :tickets: 5747
-
-    Added support for the aiomysql driver when using the asyncio SQLAlchemy
-    extension.
-
-    .. seealso::
-
-      :ref:`aiomysql`
\ No newline at end of file
diff --git a/doc/build/dialects/mysql.rst b/doc/build/dialects/mysql.rst
index c0bfa7bc6..1f2236155 100644
--- a/doc/build/dialects/mysql.rst
+++ b/doc/build/dialects/mysql.rst
@@ -181,13 +181,6 @@ MySQL-Connector
 
 .. automodule:: sqlalchemy.dialects.mysql.mysqlconnector
 
-.. _aiomysql:
-
-aiomysql
---------
-
-.. automodule:: sqlalchemy.dialects.mysql.aiomysql
-
 cymysql
 -------
 
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(
diff --git a/setup.cfg b/setup.cfg
index 46fe78104..1912fd3cd 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -65,7 +65,6 @@ postgresql_asyncpg =
 postgresql_psycopg2binary = psycopg2-binary
 postgresql_psycopg2cffi = psycopg2cffi
 pymysql = pymysql
-aiomysql = aiomysql
 
 [egg_info]
 tag_build = dev
@@ -125,7 +124,6 @@ pg8000 = postgresql+pg8000://scott:tiger@127.0.0.1:5432/test
 postgresql_psycopg2cffi = postgresql+psycopg2cffi://scott:tiger@127.0.0.1:5432/test
 mysql = mysql://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4
 pymysql = mysql+pymysql://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4
-aiomysql = mysql+aiomysql://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4&async_fallback=true
 mariadb = mariadb://scott:tiger@127.0.0.1:3306/test
 mssql = mssql+pyodbc://scott:tiger^5HHH@mssql2017:1433/test?driver=ODBC+Driver+13+for+SQL+Server
 mssql_pymssql = mssql+pymssql://scott:tiger@ms_2008
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index f55d824d6..0dc35f99e 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -1369,7 +1369,6 @@ class InvalidateDuringResultTest(fixtures.TestBase):
             "+pymysql",
             "+pg8000",
             "+asyncpg",
-            "+aiomysql",
         ],
         "Buffers the result set and doesn't check for connection close",
     )
diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 6df8a0e00..a361ff835 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -111,9 +111,7 @@ class AsyncEngineTest(EngineFixture):
         dbapi_connection = connection_fairy.connection
 
         await conn.invalidate()
-
-        if testing.against("postgresql+asyncpg"):
-            assert dbapi_connection._connection.is_closed()
+        assert dbapi_connection._connection.is_closed()
 
         new_fairy = await conn.get_raw_connection()
         is_not(new_fairy.connection, dbapi_connection)
@@ -431,8 +429,6 @@ class AsyncResultTest(EngineFixture):
 
             eq_(result.keys(), ["user_id", "user_name"])
 
-            await result.close()
-
     @async_test
     async def test_unique_all(self, async_engine):
         users = self.tables.users
diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py
index 44e295542..a3b8add67 100644
--- a/test/ext/asyncio/test_session_py3k.py
+++ b/test/ext/asyncio/test_session_py3k.py
@@ -55,7 +55,6 @@ class AsyncSessionQueryTest(AsyncFixture):
         eq_(result.scalars().all(), self.static.user_address_result)
 
     @async_test
-    @testing.requires.independent_cursors
     async def test_stream_partitions(self, async_session):
         User = self.classes.User
 
@@ -100,7 +99,6 @@ class AsyncSessionTransactionTest(AsyncFixture):
                 result = await async_session.execute(select(User))
                 eq_(result.scalar(), u1)
 
-            await outer_conn.rollback()
             eq_(await outer_conn.scalar(select(func.count(User.id))), 1)
 
     @async_test
@@ -120,7 +118,6 @@ class AsyncSessionTransactionTest(AsyncFixture):
 
             await async_session.commit()
 
-            await outer_conn.rollback()
             eq_(await outer_conn.scalar(select(func.count(User.id))), 1)
 
     @async_test
@@ -142,7 +139,6 @@ class AsyncSessionTransactionTest(AsyncFixture):
             finally:
                 await trans.commit()
 
-            await outer_conn.rollback()
             eq_(await outer_conn.scalar(select(func.count(User.id))), 1)
 
     @async_test
diff --git a/test/requirements.py b/test/requirements.py
index d8be25238..5911d87af 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -1306,9 +1306,7 @@ class DefaultRequirements(SuiteRequirements):
     def async_dialect(self):
         """dialect makes use of await_() to invoke operations on the DBAPI."""
 
-        return only_on(
-            ["postgresql+asyncpg", "mysql+aiomysql", "mariadb+aiomysql"]
-        )
+        return only_on(["postgresql+asyncpg"])
 
     @property
     def oracle_test_dblink(self):
@@ -1355,10 +1353,7 @@ class DefaultRequirements(SuiteRequirements):
 
     @property
     def percent_schema_names(self):
-        return skip_if(
-            ["mysql+aiomysql", "mariadb+aiomysql"],
-            "see pr https://github.com/aio-libs/aiomysql/pull/545",
-        )
+        return exclusions.open()
 
     @property
     def order_by_label_with_expression(self):
diff --git a/tox.ini b/tox.ini
index 9d0c75f77..6cfcf62ef 100644
--- a/tox.ini
+++ b/tox.ini
@@ -25,7 +25,6 @@ deps=pytest>=4.6.11 # this can be 6.x once we are on python 3 only
      postgresql: .[postgresql_pg8000]; python_version >= '3'
      mysql: .[mysql]
      mysql: .[pymysql]
-     mysql: .[aiomysql]; python_version >= '3'
      mysql: .[mariadb_connector]; python_version >= '3'
 
      # we should probably try to get mysql_connector back in the mix
@@ -79,7 +78,7 @@ setenv=
 
     mysql: MYSQL={env:TOX_MYSQL:--db mysql}
     mysql: EXTRA_MYSQL_DRIVERS={env:EXTRA_MYSQL_DRIVERS:--dbdriver mysqldb --dbdriver pymysql}
-    py3{,5,6,7,8,9,10,11}-mysql: EXTRA_MYSQL_DRIVERS={env:EXTRA_MYSQL_DRIVERS:--dbdriver mysqldb --dbdriver pymysql --dbdriver mariadbconnector --dbdriver aiomysql?async_fallback=true}
+    py3{,5,6,7,8,9,10,11}-mysql: EXTRA_MYSQL_DRIVERS={env:EXTRA_MYSQL_DRIVERS:--dbdriver mysqldb --dbdriver pymysql --dbdriver mariadbconnector}
 
 
     mssql: MSSQL={env:TOX_MSSQL:--db mssql}