Clean up most py3k compat

Change-Id: I8172fdcc3103ff92aa049827728484c8779af6b7

105 files changed, 590 insertions, 1486 deletions

diff --git a/lib/sqlalchemy/connectors/pyodbc.py b/lib/sqlalchemy/connectors/pyodbc.py
index 411985b5d..3aff7b479 100644
--- a/lib/sqlalchemy/connectors/pyodbc.py
+++ b/lib/sqlalchemy/connectors/pyodbc.py
@@ -6,6 +6,7 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 import re
+from urllib.parse import unquote_plus
 
 from . import Connector
 from .. import util
@@ -49,7 +50,7 @@ class PyODBCConnector(Connector):
                 connect_args[param] = util.asbool(keys.pop(param))
 
         if "odbc_connect" in keys:
-            connectors = [util.unquote_plus(keys.pop("odbc_connect"))]
+            connectors = [unquote_plus(keys.pop("odbc_connect"))]
         else:
 
             def check_quote(token):
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index e5745bf69..353c78c76 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -847,7 +847,6 @@ from ...types import NVARCHAR
 from ...types import SMALLINT
 from ...types import TEXT
 from ...types import VARCHAR
-from ...util import compat
 from ...util import update_wrapper
 from ...util.langhelpers import public_factory
 
@@ -1084,7 +1083,7 @@ class _MSDate(sqltypes.Date):
         def process(value):
             if isinstance(value, datetime.datetime):
                 return value.date()
-            elif isinstance(value, util.string_types):
+            elif isinstance(value, str):
                 m = self._reg.match(value)
                 if not m:
                     raise ValueError(
@@ -1126,7 +1125,7 @@ class TIME(sqltypes.TIME):
         def process(value):
             if isinstance(value, datetime.datetime):
                 return value.time()
-            elif isinstance(value, util.string_types):
+            elif isinstance(value, str):
                 m = self._reg.match(value)
                 if not m:
                     raise ValueError(
@@ -2383,9 +2382,7 @@ class MSDDLCompiler(compiler.DDLCompiler):
         # handle other included columns
         if index.dialect_options["mssql"]["include"]:
             inclusions = [
-                index.table.c[col]
-                if isinstance(col, util.string_types)
-                else col
+                index.table.c[col] if isinstance(col, str) else col
                 for col in index.dialect_options["mssql"]["include"]
             ]
 
@@ -3256,8 +3253,8 @@ class MSDialect(default.DefaultDialect):
                     cdict["identity"] = {}
                 else:
                     if isinstance(coltype, sqltypes.BigInteger):
-                        start = compat.long_type(identity_start)
-                        increment = compat.long_type(identity_increment)
+                        start = int(identity_start)
+                        increment = int(identity_increment)
                     elif isinstance(coltype, sqltypes.Integer):
                         start = int(identity_start)
                         increment = int(identity_increment)
diff --git a/lib/sqlalchemy/dialects/mssql/pyodbc.py b/lib/sqlalchemy/dialects/mssql/pyodbc.py
index 6ce55d392..7bcc2a467 100644
--- a/lib/sqlalchemy/dialects/mssql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mssql/pyodbc.py
@@ -398,7 +398,7 @@ class _ODBCDateTimeBindProcessor:
         def process(value):
             if value is None:
                 return None
-            elif isinstance(value, util.string_types):
+            elif isinstance(value, str):
                 # if a string was passed directly, allow it through
                 return value
             elif not value.tzinfo or (not self.timezone and not self.has_tz):
@@ -577,7 +577,7 @@ class MSDialect_pyodbc(PyODBCConnector, MSDialect):
                 tup[4],
                 tup[5],
                 tup[6] // 1000,
-                util.timezone(
+                datetime.timezone(
                     datetime.timedelta(hours=tup[7], minutes=tup[8])
                 ),
             )
diff --git a/lib/sqlalchemy/dialects/mysql/__init__.py b/lib/sqlalchemy/dialects/mysql/__init__.py
index 9fe6f6d84..389720213 100644
--- a/lib/sqlalchemy/dialects/mysql/__init__.py
+++ b/lib/sqlalchemy/dialects/mysql/__init__.py
@@ -5,6 +5,8 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+from . import aiomysql  # noqa
+from . import asyncmy  # noqa
 from . import base  # noqa
 from . import cymysql  # noqa
 from . import mariadbconnector  # noqa
@@ -51,10 +53,6 @@ from .dml import insert
 from .expression import match
 from ...util import compat
 
-if compat.py3k:
-    from . import aiomysql  # noqa
-    from . import asyncmy  # noqa
-
 # default dialect
 base.dialect = dialect = mysqldb.dialect
 
diff --git a/lib/sqlalchemy/dialects/mysql/asyncmy.py b/lib/sqlalchemy/dialects/mysql/asyncmy.py
index 0fca338f5..b59571460 100644
--- a/lib/sqlalchemy/dialects/mysql/asyncmy.py
+++ b/lib/sqlalchemy/dialects/mysql/asyncmy.py
@@ -28,11 +28,12 @@ This dialect should normally be used only with the
 
 """  # noqa
 
+from contextlib import asynccontextmanager
+
 from .pymysql import MySQLDialect_pymysql
 from ... import pool
 from ... import util
 from ...engine import AdaptedConnection
-from ...util.concurrency import asynccontextmanager
 from ...util.concurrency import asyncio
 from ...util.concurrency import await_fallback
 from ...util.concurrency import await_only
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 54fe1f57f..f77d839f3 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1469,7 +1469,7 @@ class MySQLCompiler(compiler.SQLCompiler):
            keywords at the start of a SELECT.
 
         """
-        if isinstance(select._distinct, util.string_types):
+        if isinstance(select._distinct, str):
             util.warn_deprecated(
                 "Sending string values for 'distinct' is deprecated in the "
                 "MySQL dialect and will be removed in a future release.  "
@@ -2425,7 +2425,7 @@ class MySQLDialect(default.DefaultDialect):
             raise NotImplementedError()
         val = row[0]
         cursor.close()
-        if util.py3k and isinstance(val, bytes):
+        if isinstance(val, bytes):
             val = val.decode()
         return val.upper().replace("-", " ")
 
@@ -2456,7 +2456,7 @@ class MySQLDialect(default.DefaultDialect):
         cursor.execute("SELECT VERSION()")
         val = cursor.fetchone()[0]
         cursor.close()
-        if util.py3k and isinstance(val, bytes):
+        if isinstance(val, bytes):
             val = val.decode()
 
         return self._parse_server_version(val)
@@ -2607,8 +2607,8 @@ class MySQLDialect(default.DefaultDialect):
                 sql.bindparam("table_name", type_=Unicode),
             ),
             {
-                "table_schema": util.text_type(schema),
-                "table_name": util.text_type(table_name),
+                "table_schema": str(schema),
+                "table_name": str(table_name),
             },
         )
         return bool(rs.scalar())
@@ -2627,8 +2627,8 @@ class MySQLDialect(default.DefaultDialect):
                 "TABLE_SCHEMA=:schema_name"
             ),
             dict(
-                name=util.text_type(sequence_name),
-                schema_name=util.text_type(schema),
+                name=str(sequence_name),
+                schema_name=str(schema),
             ),
         )
         return cursor.first() is not None
@@ -3228,7 +3228,7 @@ class _DecodingRow:
         if isinstance(item, _array):
             item = item.tostring()
 
-        if self.charset and isinstance(item, util.binary_type):
+        if self.charset and isinstance(item, bytes):
             return item.decode(self.charset)
         else:
             return item
@@ -3237,7 +3237,7 @@ class _DecodingRow:
         item = getattr(self.rowproxy, attr)
         if isinstance(item, _array):
             item = item.tostring()
-        if self.charset and isinstance(item, util.binary_type):
+        if self.charset and isinstance(item, bytes):
             return item.decode(self.charset)
         else:
             return item
diff --git a/lib/sqlalchemy/dialects/mysql/cymysql.py b/lib/sqlalchemy/dialects/mysql/cymysql.py
index f729e4a18..4fe441031 100644
--- a/lib/sqlalchemy/dialects/mysql/cymysql.py
+++ b/lib/sqlalchemy/dialects/mysql/cymysql.py
@@ -33,7 +33,7 @@ class _cymysqlBIT(BIT):
         def process(value):
             if value is not None:
                 v = 0
-                for i in util.iterbytes(value):
+                for i in iter(value):
                     v = v << 8 | i
                 return v
             return value
diff --git a/lib/sqlalchemy/dialects/mysql/enumerated.py b/lib/sqlalchemy/dialects/mysql/enumerated.py
index 9f9a838c5..b84608f58 100644
--- a/lib/sqlalchemy/dialects/mysql/enumerated.py
+++ b/lib/sqlalchemy/dialects/mysql/enumerated.py
@@ -201,7 +201,7 @@ class SET(_StringType):
             super_convert = super(SET, self).result_processor(dialect, coltype)
 
             def process(value):
-                if isinstance(value, util.string_types):
+                if isinstance(value, str):
                     # MySQLdb returns a string, let's parse
                     if super_convert:
                         value = super_convert(value)
@@ -222,7 +222,7 @@ class SET(_StringType):
             def process(value):
                 if value is None:
                     return None
-                elif isinstance(value, util.int_types + util.string_types):
+                elif isinstance(value, (int, str)):
                     if super_convert:
                         return super_convert(value)
                     else:
@@ -237,9 +237,7 @@ class SET(_StringType):
 
             def process(value):
                 # accept strings and int (actually bitflag) values directly
-                if value is not None and not isinstance(
-                    value, util.int_types + util.string_types
-                ):
+                if value is not None and not isinstance(value, (int, str)):
                     value = ",".join(value)
 
                 if super_convert:
diff --git a/lib/sqlalchemy/dialects/mysql/json.py b/lib/sqlalchemy/dialects/mysql/json.py
index 384d3b9b6..c46878fc3 100644
--- a/lib/sqlalchemy/dialects/mysql/json.py
+++ b/lib/sqlalchemy/dialects/mysql/json.py
@@ -4,9 +4,6 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
-
-from __future__ import absolute_import
-
 from ... import types as sqltypes
 
 
diff --git a/lib/sqlalchemy/dialects/mysql/mysqlconnector.py b/lib/sqlalchemy/dialects/mysql/mysqlconnector.py
index fef4f14ca..7b62e9ed1 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqlconnector.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqlconnector.py
@@ -32,36 +32,17 @@ from ... import util
 
 class MySQLCompiler_mysqlconnector(MySQLCompiler):
     def visit_mod_binary(self, binary, operator, **kw):
-        if self.dialect._mysqlconnector_double_percents:
-            return (
-                self.process(binary.left, **kw)
-                + " %% "
-                + self.process(binary.right, **kw)
-            )
-        else:
-            return (
-                self.process(binary.left, **kw)
-                + " % "
-                + self.process(binary.right, **kw)
-            )
-
-    def post_process_text(self, text):
-        if self.dialect._mysqlconnector_double_percents:
-            return text.replace("%", "%%")
-        else:
-            return text
-
-    def escape_literal_column(self, text):
-        if self.dialect._mysqlconnector_double_percents:
-            return text.replace("%", "%%")
-        else:
-            return text
+        return (
+            self.process(binary.left, **kw)
+            + " % "
+            + self.process(binary.right, **kw)
+        )
 
 
 class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
     @property
     def _double_percents(self):
-        return self.dialect._mysqlconnector_double_percents
+        return False
 
     @_double_percents.setter
     def _double_percents(self, value):
@@ -69,10 +50,7 @@ class MySQLIdentifierPreparer_mysqlconnector(MySQLIdentifierPreparer):
 
     def _escape_identifier(self, value):
         value = value.replace(self.escape_quote, self.escape_to_quote)
-        if self.dialect._mysqlconnector_double_percents:
-            return value.replace("%", "%%")
-        else:
-            return value
+        return value
 
 
 class _myconnpyBIT(BIT):
@@ -163,10 +141,6 @@ class MySQLDialect_mysqlconnector(MySQLDialect):
             if m:
                 return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
 
-    @util.memoized_property
-    def _mysqlconnector_double_percents(self):
-        return not util.py3k and self._mysqlconnector_version_info < (2, 0)
-
     def _detect_charset(self, connection):
         return connection.connection.charset
 
diff --git a/lib/sqlalchemy/dialects/mysql/pymysql.py b/lib/sqlalchemy/dialects/mysql/pymysql.py
index 3c30fb9ea..dd7bd8bda 100644
--- a/lib/sqlalchemy/dialects/mysql/pymysql.py
+++ b/lib/sqlalchemy/dialects/mysql/pymysql.py
@@ -39,7 +39,6 @@ to the pymysql driver as well.
 
 from .mysqldb import MySQLDialect_mysqldb
 from ...util import langhelpers
-from ...util import py3k
 
 
 class MySQLDialect_pymysql(MySQLDialect_mysqldb):
@@ -81,12 +80,10 @@ class MySQLDialect_pymysql(MySQLDialect_mysqldb):
         else:
             return False
 
-    if py3k:
-
-        def _extract_error_code(self, exception):
-            if isinstance(exception.args[0], Exception):
-                exception = exception.args[0]
-            return exception.args[0]
+    def _extract_error_code(self, exception):
+        if isinstance(exception.args[0], Exception):
+            exception = exception.args[0]
+        return exception.args[0]
 
 
 dialect = MySQLDialect_pymysql
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index a1d0f80d6..7df2422b3 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -564,7 +564,6 @@ from ...types import NCHAR
 from ...types import NVARCHAR
 from ...types import TIMESTAMP
 from ...types import VARCHAR
-from ...util import compat
 
 RESERVED_WORDS = set(
     "SHARE RAW DROP BETWEEN FROM DESC OPTION PRIOR LONG THEN "
@@ -1414,7 +1413,7 @@ class OracleIdentifierPreparer(compiler.IdentifierPreparer):
         return (
             lc_value in self.reserved_words
             or value[0] in self.illegal_initial_characters
-            or not self.legal_characters.match(util.text_type(value))
+            or not self.legal_characters.match(str(value))
         )
 
     def format_savepoint(self, savepoint):
@@ -2029,17 +2028,17 @@ class OracleDialect(default.DefaultDialect):
             value = value.strip()
 
             if "START WITH" in option:
-                identity["start"] = compat.long_type(value)
+                identity["start"] = int(value)
             elif "INCREMENT BY" in option:
-                identity["increment"] = compat.long_type(value)
+                identity["increment"] = int(value)
             elif "MAX_VALUE" in option:
-                identity["maxvalue"] = compat.long_type(value)
+                identity["maxvalue"] = int(value)
             elif "MIN_VALUE" in option:
-                identity["minvalue"] = compat.long_type(value)
+                identity["minvalue"] = int(value)
             elif "CYCLE_FLAG" in option:
                 identity["cycle"] = value == "Y"
             elif "CACHE_SIZE" in option:
-                identity["cache"] = compat.long_type(value)
+                identity["cache"] = int(value)
             elif "ORDER_FLAG" in option:
                 identity["order"] = value == "Y"
         return identity
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index 2cfcb0e5c..07317126d 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -430,9 +430,6 @@ SQLAlchemy type (or a subclass of such).
    as better integration of outputtypehandlers.
 
 """  # noqa
-
-from __future__ import absolute_import
-
 import decimal
 import random
 import re
@@ -1127,7 +1124,7 @@ class OracleDialect_cx_oracle(OracleDialect):
                 and default_type is not cx_Oracle.NCLOB
             ):
                 return cursor.var(
-                    util.text_type,
+                    str,
                     size,
                     cursor.arraysize,
                     **dialect._cursor_var_unicode_kwargs
@@ -1213,7 +1210,7 @@ class OracleDialect_cx_oracle(OracleDialect):
             opts.setdefault("threaded", self._cx_oracle_threaded)
 
         def convert_cx_oracle_constant(value):
-            if isinstance(value, util.string_types):
+            if isinstance(value, str):
                 try:
                     int_val = int(value)
                 except ValueError:
diff --git a/lib/sqlalchemy/dialects/postgresql/__init__.py b/lib/sqlalchemy/dialects/postgresql/__init__.py
index 056de66be..08b05dc74 100644
--- a/lib/sqlalchemy/dialects/postgresql/__init__.py
+++ b/lib/sqlalchemy/dialects/postgresql/__init__.py
@@ -4,6 +4,7 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
+from . import asyncpg  # noqa
 from . import base
 from . import pg8000  # noqa
 from . import psycopg2  # noqa
@@ -57,9 +58,6 @@ from .ranges import TSRANGE
 from .ranges import TSTZRANGE
 from ...util import compat
 
-if compat.py3k:
-    from . import asyncpg  # noqa
-
 base.dialect = dialect = psycopg2.dialect
 
 
diff --git a/lib/sqlalchemy/dialects/postgresql/array.py b/lib/sqlalchemy/dialects/postgresql/array.py
index 0cb574dac..614559035 100644
--- a/lib/sqlalchemy/dialects/postgresql/array.py
+++ b/lib/sqlalchemy/dialects/postgresql/array.py
@@ -374,12 +374,12 @@ class ARRAY(sqltypes.ARRAY):
             def process(value):
                 if value is None:
                     return value
-                # isinstance(value, util.string_types) is required to handle
+                # isinstance(value, str) is required to handle
                 # the case where a TypeDecorator for and Array of Enum is
                 # used like was required in sa < 1.3.17
                 return super_rp(
                     handle_raw_string(value)
-                    if isinstance(value, util.string_types)
+                    if isinstance(value, str)
                     else value
                 )
 
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index fe1f9fd5a..d6cde0087 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -99,6 +99,7 @@ To disable the prepared statement cache, use a value of zero::
 """  # noqa
 
 import collections
+import collections.abc as collections_abc
 import decimal
 import json as _py_json
 import re
@@ -216,8 +217,8 @@ class AsyncpgJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
 class AsyncpgJSONPathType(json.JSONPathType):
     def bind_processor(self, dialect):
         def process(value):
-            assert isinstance(value, util.collections_abc.Sequence)
-            tokens = [util.text_type(elem) for elem in value]
+            assert isinstance(value, collections_abc.Sequence)
+            tokens = [str(elem) for elem in value]
             return tokens
 
         return process
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 583d9c263..d00318fc8 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1721,7 +1721,7 @@ class UUID(sqltypes.TypeEngine):
     def coerce_compared_value(self, op, value):
         """See :meth:`.TypeEngine.coerce_compared_value` for a description."""
 
-        if isinstance(value, util.string_types):
+        if isinstance(value, str):
             return self
         else:
             return super(UUID, self).coerce_compared_value(op, value)
@@ -1731,7 +1731,7 @@ class UUID(sqltypes.TypeEngine):
 
             def process(value):
                 if value is not None:
-                    value = util.text_type(value)
+                    value = str(value)
                 return value
 
             return process
@@ -2375,7 +2375,7 @@ class PGCompiler(compiler.SQLCompiler):
             target_text = "(%s)" % ", ".join(
                 (
                     self.preparer.quote(c)
-                    if isinstance(c, util.string_types)
+                    if isinstance(c, str)
                     else self.process(c, include_table=False, use_schema=False)
                 )
                 for c in clause.inferred_target_elements
@@ -2451,7 +2451,7 @@ class PGCompiler(compiler.SQLCompiler):
             for k, v in set_parameters.items():
                 key_text = (
                     self.preparer.quote(k)
-                    if isinstance(k, util.string_types)
+                    if isinstance(k, str)
                     else self.process(k, use_schema=False)
                 )
                 value_text = self.process(
@@ -2653,9 +2653,7 @@ class PGDDLCompiler(compiler.DDLCompiler):
         includeclause = index.dialect_options["postgresql"]["include"]
         if includeclause:
             inclusions = [
-                index.table.c[col]
-                if isinstance(col, util.string_types)
-                else col
+                index.table.c[col] if isinstance(col, str) else col
                 for col in includeclause
             ]
             text += " INCLUDE (%s)" % ", ".join(
@@ -3326,7 +3324,7 @@ class PGDialect(default.DefaultDialect):
             sql.text(query).bindparams(
                 sql.bindparam(
                     "schema",
-                    util.text_type(schema.lower()),
+                    str(schema.lower()),
                     type_=sqltypes.Unicode,
                 )
             )
@@ -3347,7 +3345,7 @@ class PGDialect(default.DefaultDialect):
                 ).bindparams(
                     sql.bindparam(
                         "name",
-                        util.text_type(table_name),
+                        str(table_name),
                         type_=sqltypes.Unicode,
                     )
                 )
@@ -3361,12 +3359,12 @@ class PGDialect(default.DefaultDialect):
                 ).bindparams(
                     sql.bindparam(
                         "name",
-                        util.text_type(table_name),
+                        str(table_name),
                         type_=sqltypes.Unicode,
                     ),
                     sql.bindparam(
                         "schema",
-                        util.text_type(schema),
+                        str(schema),
                         type_=sqltypes.Unicode,
                     ),
                 )
@@ -3384,12 +3382,12 @@ class PGDialect(default.DefaultDialect):
             ).bindparams(
                 sql.bindparam(
                     "name",
-                    util.text_type(sequence_name),
+                    str(sequence_name),
                     type_=sqltypes.Unicode,
                 ),
                 sql.bindparam(
                     "schema",
-                    util.text_type(schema),
+                    str(schema),
                     type_=sqltypes.Unicode,
                 ),
             )
@@ -3418,15 +3416,11 @@ class PGDialect(default.DefaultDialect):
                 """
             query = sql.text(query)
         query = query.bindparams(
-            sql.bindparam(
-                "typname", util.text_type(type_name), type_=sqltypes.Unicode
-            )
+            sql.bindparam("typname", str(type_name), type_=sqltypes.Unicode)
         )
         if schema is not None:
             query = query.bindparams(
-                sql.bindparam(
-                    "nspname", util.text_type(schema), type_=sqltypes.Unicode
-                )
+                sql.bindparam("nspname", str(schema), type_=sqltypes.Unicode)
             )
         cursor = connection.execute(query)
         return bool(cursor.scalar())
@@ -3471,9 +3465,9 @@ class PGDialect(default.DefaultDialect):
         )
         # Since we're binding to unicode, table_name and schema_name must be
         # unicode.
-        table_name = util.text_type(table_name)
+        table_name = str(table_name)
         if schema is not None:
-            schema = util.text_type(schema)
+            schema = str(schema)
         s = sql.text(query).bindparams(table_name=sqltypes.Unicode)
         s = s.columns(oid=sqltypes.Integer)
         if schema:
@@ -3573,7 +3567,7 @@ class PGDialect(default.DefaultDialect):
             ).bindparams(
                 sql.bindparam(
                     "schema",
-                    util.text_type(schema),
+                    str(schema),
                     type_=sqltypes.Unicode,
                 ),
             )
diff --git a/lib/sqlalchemy/dialects/postgresql/dml.py b/lib/sqlalchemy/dialects/postgresql/dml.py
index bb6345cf4..c561b73a1 100644
--- a/lib/sqlalchemy/dialects/postgresql/dml.py
+++ b/lib/sqlalchemy/dialects/postgresql/dml.py
@@ -185,7 +185,7 @@ class OnConflictClause(ClauseElement):
     def __init__(self, constraint=None, index_elements=None, index_where=None):
 
         if constraint is not None:
-            if not isinstance(constraint, util.string_types) and isinstance(
+            if not isinstance(constraint, str) and isinstance(
                 constraint,
                 (schema.Index, schema.Constraint, ext.ExcludeConstraint),
             ):
@@ -197,7 +197,7 @@ class OnConflictClause(ClauseElement):
                     "'constraint' and 'index_elements' are mutually exclusive"
                 )
 
-            if isinstance(constraint, util.string_types):
+            if isinstance(constraint, str):
                 self.constraint_target = constraint
                 self.inferred_target_elements = None
                 self.inferred_target_whereclause = None
diff --git a/lib/sqlalchemy/dialects/postgresql/ext.py b/lib/sqlalchemy/dialects/postgresql/ext.py
index f9e4c1d6c..f779a8010 100644
--- a/lib/sqlalchemy/dialects/postgresql/ext.py
+++ b/lib/sqlalchemy/dialects/postgresql/ext.py
@@ -4,9 +4,9 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
+from itertools import zip_longest
 
 from .array import ARRAY
-from ... import util
 from ...sql import coercions
 from ...sql import elements
 from ...sql import expression
@@ -237,7 +237,7 @@ class ExcludeConstraint(ColumnCollectionConstraint):
                 name,
                 operator,
             )
-            for (expr, name, operator), colexpr in util.zip_longest(
+            for (expr, name, operator), colexpr in zip_longest(
                 self._render_exprs, self.columns
             )
         ]
diff --git a/lib/sqlalchemy/dialects/postgresql/hstore.py b/lib/sqlalchemy/dialects/postgresql/hstore.py
index 85d678ef5..2ade4b7c1 100644
--- a/lib/sqlalchemy/dialects/postgresql/hstore.py
+++ b/lib/sqlalchemy/dialects/postgresql/hstore.py
@@ -9,7 +9,6 @@ import re
 
 from .array import ARRAY
 from ... import types as sqltypes
-from ... import util
 from ...sql import functions as sqlfunc
 from ...sql import operators
 
@@ -413,7 +412,7 @@ def _serialize_hstore(val):
     def esc(s, position):
         if position == "value" and s is None:
             return "NULL"
-        elif isinstance(s, util.string_types):
+        elif isinstance(s, str):
             return '"%s"' % s.replace("\\", "\\\\").replace('"', r"\"")
         else:
             raise ValueError(
diff --git a/lib/sqlalchemy/dialects/postgresql/json.py b/lib/sqlalchemy/dialects/postgresql/json.py
index ef046e3ae..fb7621365 100644
--- a/lib/sqlalchemy/dialects/postgresql/json.py
+++ b/lib/sqlalchemy/dialects/postgresql/json.py
@@ -4,10 +4,9 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
-from __future__ import absolute_import
+import collections.abc as collections_abc
 
 from ... import types as sqltypes
-from ... import util
 from ...sql import operators
 
 
@@ -71,8 +70,8 @@ class JSONPathType(sqltypes.JSON.JSONPathType):
         super_proc = self.string_bind_processor(dialect)
 
         def process(value):
-            assert isinstance(value, util.collections_abc.Sequence)
-            tokens = [util.text_type(elem) for elem in value]
+            assert isinstance(value, collections_abc.Sequence)
+            tokens = [str(elem) for elem in value]
             value = "{%s}" % (", ".join(tokens))
             if super_proc:
                 value = super_proc(value)
@@ -84,8 +83,8 @@ class JSONPathType(sqltypes.JSON.JSONPathType):
         super_proc = self.string_literal_processor(dialect)
 
         def process(value):
-            assert isinstance(value, util.collections_abc.Sequence)
-            tokens = [util.text_type(elem) for elem in value]
+            assert isinstance(value, collections_abc.Sequence)
+            tokens = [str(elem) for elem in value]
             value = "{%s}" % (", ".join(tokens))
             if super_proc:
                 value = super_proc(value)
diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index 324007e7e..ac29b28e9 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -554,7 +554,7 @@ class PGDialect_pg8000(PGDialect):
         fns = []
 
         def on_connect(conn):
-            conn.py_types[quoted_name] = conn.py_types[util.text_type]
+            conn.py_types[quoted_name] = conn.py_types[str]
 
         fns.append(on_connect)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index f62830a0d..11a5f31a3 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -441,8 +441,7 @@ place within SQLAlchemy's own marshalling logic, and not that of ``psycopg2``
 which may be more performant.
 
 """  # noqa
-from __future__ import absolute_import
-
+import collections.abc as collections_abc
 import decimal
 import logging
 import re
@@ -466,7 +465,6 @@ from ... import processors
 from ... import types as sqltypes
 from ... import util
 from ...engine import cursor as _cursor
-from ...util import collections_abc
 
 
 logger = logging.getLogger("sqlalchemy.dialects.postgresql")
diff --git a/lib/sqlalchemy/dialects/sqlite/__init__.py b/lib/sqlalchemy/dialects/sqlite/__init__.py
index 6e3ad0e66..e2f59907e 100644
--- a/lib/sqlalchemy/dialects/sqlite/__init__.py
+++ b/lib/sqlalchemy/dialects/sqlite/__init__.py
@@ -5,6 +5,7 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+from . import aiosqlite  # noqa
 from . import base  # noqa
 from . import pysqlcipher  # noqa
 from . import pysqlite  # noqa
@@ -26,10 +27,6 @@ from .base import TIMESTAMP
 from .base import VARCHAR
 from .dml import Insert
 from .dml import insert
-from ...util import compat
-
-if compat.py3k:
-    from . import aiosqlite  # noqa
 
 # default dialect
 base.dialect = dialect = pysqlite.dialect
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 7ded73973..3dea23e18 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1324,7 +1324,7 @@ class SQLiteCompiler(compiler.SQLCompiler):
             target_text = "(%s)" % ", ".join(
                 (
                     self.preparer.quote(c)
-                    if isinstance(c, util.string_types)
+                    if isinstance(c, str)
                     else self.process(c, include_table=False, use_schema=False)
                 )
                 for c in clause.inferred_target_elements
@@ -1401,7 +1401,7 @@ class SQLiteCompiler(compiler.SQLCompiler):
             for k, v in set_parameters.items():
                 key_text = (
                     self.preparer.quote(k)
-                    if isinstance(k, util.string_types)
+                    if isinstance(k, str)
                     else self.process(k, use_schema=False)
                 )
                 value_text = self.process(
@@ -2110,7 +2110,7 @@ class SQLiteDialect(default.DefaultDialect):
         coltype = self._resolve_type_affinity(type_)
 
         if default is not None:
-            default = util.text_type(default)
+            default = str(default)
 
         colspec = {
             "name": name,
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
index 3765191c1..d3c504fd4 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlcipher.py
@@ -94,11 +94,8 @@ time, at the expense of slower startup time for new connections.
 
 """  # noqa
 
-from __future__ import absolute_import
-
 from .pysqlite import SQLiteDialect_pysqlite
 from ... import pool
-from ... import util
 
 
 class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
@@ -109,18 +106,14 @@ class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
 
     @classmethod
     def dbapi(cls):
-        if util.py3k:
-            try:
-                import sqlcipher3 as sqlcipher
-            except ImportError:
-                pass
-            else:
-                return sqlcipher
-
-            from pysqlcipher3 import dbapi2 as sqlcipher
-
+        try:
+            import sqlcipher3 as sqlcipher
+        except ImportError:
+            pass
         else:
-            from pysqlcipher import dbapi2 as sqlcipher
+            return sqlcipher
+
+        from pysqlcipher3 import dbapi2 as sqlcipher
 
         return sqlcipher
 
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 389270e45..d2939c035 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -4,8 +4,6 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
-from __future__ import with_statement
-
 import contextlib
 import sys
 
@@ -1464,7 +1462,7 @@ class Connection(Connectable):
             raise
         except BaseException as e:
             self._handle_dbapi_exception(
-                e, util.text_type(statement), parameters, None, None
+                e, str(statement), parameters, None, None
             )
             return  # not reached
 
@@ -2518,7 +2516,7 @@ class Engine(ConnectionEventsTarget, log.Identified):
         else:
             yield connection
 
-    @util.contextmanager
+    @contextlib.contextmanager
     def begin(self):
         """Return a context manager delivering a :class:`_engine.Connection`
         with a :class:`.Transaction` established.
diff --git a/lib/sqlalchemy/engine/characteristics.py b/lib/sqlalchemy/engine/characteristics.py
index 2543f591b..10455451f 100644
--- a/lib/sqlalchemy/engine/characteristics.py
+++ b/lib/sqlalchemy/engine/characteristics.py
@@ -1,9 +1,7 @@
 import abc
 
-from ..util import ABC
 
-
-class ConnectionCharacteristic(ABC):
+class ConnectionCharacteristic(abc.ABC):
     """An abstract base for an object that can set, get and reset a
     per-connection characteristic, typically one that gets reset when the
     connection is returned to the connection pool.
diff --git a/lib/sqlalchemy/engine/cursor.py b/lib/sqlalchemy/engine/cursor.py
index 54db9f6c2..1f1a2fcf1 100644
--- a/lib/sqlalchemy/engine/cursor.py
+++ b/lib/sqlalchemy/engine/cursor.py
@@ -669,7 +669,7 @@ class CursorResultMetaData(ResultMetaData):
             "_keymap": {
                 key: (rec[MD_INDEX], rec[MD_RESULT_MAP_INDEX], _UNPICKLED, key)
                 for key, rec in self._keymap.items()
-                if isinstance(key, util.string_types + util.int_types)
+                if isinstance(key, (str, int))
             },
             "_keys": self._keys,
             "_translated_indexes": self._translated_indexes,
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 3af24d913..a47ed963d 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -16,6 +16,7 @@ as the base class for their own corresponding classes.
 import functools
 import random
 import re
+from time import perf_counter
 import weakref
 
 from . import characteristics
@@ -780,7 +781,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
 
         self.execution_options = execution_options
 
-        self.unicode_statement = util.text_type(compiled)
+        self.unicode_statement = str(compiled)
         if compiled.schema_translate_map:
             schema_translate_map = self.execution_options.get(
                 "schema_translate_map", {}
@@ -1003,7 +1004,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         if self.compiled is None:
             return "raw sql"
 
-        now = util.perf_counter()
+        now = perf_counter()
 
         ch = self.cache_hit
 
@@ -1515,7 +1516,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
                 for key in compiled_params
             )
         return self._execute_scalar(
-            util.text_type(compiled), type_, parameters=parameters
+            str(compiled), type_, parameters=parameters
         )
 
     current_parameters = None
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index c918f15b3..7abc404f0 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -47,7 +47,7 @@ def cache(fn, self, con, *args, **kw):
         return fn(self, con, *args, **kw)
     key = (
         fn.__name__,
-        tuple(a for a in args if isinstance(a, util.string_types)),
+        tuple(a for a in args if isinstance(a, str)),
         tuple((k, v) for k, v in kw.items() if k != "info_cache"),
     )
     ret = info_cache.get(key)
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 4187c6c13..e2f4033e0 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -8,6 +8,7 @@
 """Define generic result set constructs."""
 
 
+import collections.abc as collections_abc
 import functools
 import itertools
 import operator
@@ -19,7 +20,6 @@ from .. import util
 from ..sql.base import _generative
 from ..sql.base import HasMemoized
 from ..sql.base import InPlaceGenerative
-from ..util import collections_abc
 
 
 if _baserow_usecext:
diff --git a/lib/sqlalchemy/engine/row.py b/lib/sqlalchemy/engine/row.py
index 782fc21b8..1a8c4e555 100644
--- a/lib/sqlalchemy/engine/row.py
+++ b/lib/sqlalchemy/engine/row.py
@@ -8,11 +8,11 @@
 """Define row constructs including :class:`.Row`."""
 
 
+import collections.abc as collections_abc
 import operator
 
 from .. import util
 from ..sql import util as sql_util
-from ..util.compat import collections_abc
 
 MD_INDEX = 0  # integer index in cursor.description
 
diff --git a/lib/sqlalchemy/engine/url.py b/lib/sqlalchemy/engine/url.py
index 7f09b1eac..c83753bdc 100644
--- a/lib/sqlalchemy/engine/url.py
+++ b/lib/sqlalchemy/engine/url.py
@@ -14,15 +14,17 @@ argument; alternatively, the URL is a public-facing construct which can
 be used directly and is also accepted directly by ``create_engine()``.
 """
 
+import collections.abc as collections_abc
 import re
+from urllib.parse import parse_qsl
+from urllib.parse import quote_plus
+from urllib.parse import unquote
 
 from .interfaces import Dialect
 from .. import exc
 from .. import util
 from ..dialects import plugins
 from ..dialects import registry
-from ..util import collections_abc
-from ..util import compat
 
 
 class URL(
@@ -165,7 +167,7 @@ class URL(
 
     @classmethod
     def _assert_str(cls, v, paramname):
-        if not isinstance(v, compat.string_types):
+        if not isinstance(v, str):
             raise TypeError("%s must be a string" % paramname)
         return v
 
@@ -193,7 +195,7 @@ class URL(
                 )
 
         def _assert_str(v):
-            if not isinstance(v, compat.string_types):
+            if not isinstance(v, str):
                 raise TypeError("Query dictionary keys must be strings")
             return v
 
@@ -308,9 +310,7 @@ class URL(
             :meth:`_engine.URL.update_query_dict`
 
         """  # noqa: E501
-        return self.update_query_pairs(
-            util.parse_qsl(query_string), append=append
-        )
+        return self.update_query_pairs(parse_qsl(query_string), append=append)
 
     def update_query_pairs(self, key_value_pairs, append=False):
         """Return a new :class:`_engine.URL` object with the
@@ -548,7 +548,7 @@ class URL(
             keys = list(self.query)
             keys.sort()
             s += "?" + "&".join(
-                "%s=%s" % (util.quote_plus(k), util.quote_plus(element))
+                "%s=%s" % (quote_plus(k), quote_plus(element))
                 for k in keys
                 for element in util.to_list(self.query[k])
             )
@@ -711,7 +711,7 @@ def make_url(name_or_url):
     existing URL object is passed, just returns the object.
     """
 
-    if isinstance(name_or_url, util.string_types):
+    if isinstance(name_or_url, str):
         return _parse_rfc1738_args(name_or_url)
     else:
         return name_or_url
@@ -744,7 +744,7 @@ def _parse_rfc1738_args(name):
         if components["query"] is not None:
             query = {}
 
-            for key, value in util.parse_qsl(components["query"]):
+            for key, value in parse_qsl(components["query"]):
                 if key in query:
                     query[key] = util.to_list(query[key])
                     query[key].append(value)
@@ -780,15 +780,14 @@ def _rfc_1738_quote(text):
     return re.sub(r"[:@/]", lambda m: "%%%X" % ord(m.group(0)), text)
 
 
-def _rfc_1738_unquote(text):
-    return util.unquote(text)
+_rfc_1738_unquote = unquote
 
 
 def _parse_keyvalue_args(name):
     m = re.match(r"(\w+)://(.*)", name)
     if m is not None:
         (name, args) = m.group(1, 2)
-        opts = dict(util.parse_qsl(args))
+        opts = dict(parse_qsl(args))
         return URL(name, *opts)
     else:
         return None
diff --git a/lib/sqlalchemy/engine/util.py b/lib/sqlalchemy/engine/util.py
index 4467bafd3..732ae7fa1 100644
--- a/lib/sqlalchemy/engine/util.py
+++ b/lib/sqlalchemy/engine/util.py
@@ -5,9 +5,10 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+import collections.abc as collections_abc
+
 from .. import exc
 from .. import util
-from ..util import collections_abc
 from ..util import immutabledict
 
 
diff --git a/lib/sqlalchemy/event/api.py b/lib/sqlalchemy/event/api.py
index 5487c9f1a..8c928e3fe 100644
--- a/lib/sqlalchemy/event/api.py
+++ b/lib/sqlalchemy/event/api.py
@@ -8,8 +8,6 @@
 """Public API functions for the event system.
 
 """
-from __future__ import absolute_import
-
 from .base import _registrars
 from .registry import _EventKey
 from .. import exc
diff --git a/lib/sqlalchemy/event/attr.py b/lib/sqlalchemy/event/attr.py
index 77eb0472c..be36be661 100644
--- a/lib/sqlalchemy/event/attr.py
+++ b/lib/sqlalchemy/event/attr.py
@@ -28,10 +28,6 @@ as well as support for subclass propagation (e.g. events assigned to
 ``Pool`` vs. ``QueuePool``) are all implemented here.
 
 """
-
-from __future__ import absolute_import
-from __future__ import with_statement
-
 import collections
 from itertools import chain
 import weakref
diff --git a/lib/sqlalchemy/event/base.py b/lib/sqlalchemy/event/base.py
index b084207b4..1041647c8 100644
--- a/lib/sqlalchemy/event/base.py
+++ b/lib/sqlalchemy/event/base.py
@@ -15,8 +15,6 @@ at the class level of a particular ``_Dispatch`` class as well as within
 instances of ``_Dispatch``.
 
 """
-from __future__ import absolute_import
-
 import weakref
 
 from .attr import _ClsLevelDispatch
@@ -218,7 +216,7 @@ def _remove_dispatcher(cls):
             del _registrars[k]
 
 
-class Events(util.with_metaclass(_EventMeta, object)):
+class Events(metaclass=_EventMeta):
     """Define event listening functions for a particular target type."""
 
     @staticmethod
diff --git a/lib/sqlalchemy/event/registry.py b/lib/sqlalchemy/event/registry.py
index d81f27cd4..94cb5cb95 100644
--- a/lib/sqlalchemy/event/registry.py
+++ b/lib/sqlalchemy/event/registry.py
@@ -14,9 +14,6 @@ membership in all those collections can be revoked at once, based on
 an equivalent :class:`._EventKey`.
 
 """
-
-from __future__ import absolute_import
-
 import collections
 import types
 import weakref
diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index ef246d04f..e35c41836 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -53,34 +53,27 @@ class HasDescriptionCode:
 class SQLAlchemyError(HasDescriptionCode, Exception):
     """Generic error class."""
 
-    def _message(self, as_unicode=compat.py3k):
+    def _message(self):
         # rules:
         #
-        # 1. under py2k, for __str__ return single string arg as it was
-        # given without converting to unicode.  for __unicode__
-        # do a conversion but check that it's not unicode already just in
-        # case
-        #
-        # 2. under py3k, single arg string will usually be a unicode
+        # 1. single arg string will usually be a unicode
         # object, but since __str__() must return unicode, check for
         # bytestring just in case
         #
-        # 3. for multiple self.args, this is not a case in current
+        # 2. for multiple self.args, this is not a case in current
         # SQLAlchemy though this is happening in at least one known external
         # library, call str() which does a repr().
         #
         if len(self.args) == 1:
             text = self.args[0]
 
-            if as_unicode and isinstance(text, compat.binary_types):
+            if isinstance(text, bytes):
                 text = compat.decode_backslashreplace(text, "utf-8")
             # This is for when the argument is not a string of any sort.
             # Otherwise, converting this exception to string would fail for
             # non-string arguments.
-            elif compat.py3k or not as_unicode:
-                text = str(text)
             else:
-                text = compat.text_type(text)
+                text = str(text)
 
             return text
         else:
@@ -89,8 +82,8 @@ class SQLAlchemyError(HasDescriptionCode, Exception):
             # a repr() of the tuple
             return str(self.args)
 
-    def _sql_message(self, as_unicode):
-        message = self._message(as_unicode)
+    def _sql_message(self):
+        message = self._message()
 
         if self.code:
             message = "%s %s" % (message, self._code_str())
@@ -98,10 +91,7 @@ class SQLAlchemyError(HasDescriptionCode, Exception):
         return message
 
     def __str__(self):
-        return self._sql_message(compat.py3k)
-
-    def __unicode__(self):
-        return self._sql_message(as_unicode=True)
+        return self._sql_message()
 
 
 class ArgumentError(SQLAlchemyError):
@@ -458,17 +448,12 @@ class StatementError(SQLAlchemyError):
         )
 
     @_preloaded.preload_module("sqlalchemy.sql.util")
-    def _sql_message(self, as_unicode):
+    def _sql_message(self):
         util = _preloaded.preloaded.sql_util
 
-        details = [self._message(as_unicode=as_unicode)]
+        details = [self._message()]
         if self.statement:
-            if not as_unicode and not compat.py3k:
-                stmt_detail = "[SQL: %s]" % compat.safe_bytestring(
-                    self.statement
-                )
-            else:
-                stmt_detail = "[SQL: %s]" % self.statement
+            stmt_detail = "[SQL: %s]" % self.statement
             details.append(stmt_detail)
             if self.params:
                 if self.hide_parameters:
diff --git a/lib/sqlalchemy/ext/associationproxy.py b/lib/sqlalchemy/ext/associationproxy.py
index 0aa836a3c..9e8f84f46 100644
--- a/lib/sqlalchemy/ext/associationproxy.py
+++ b/lib/sqlalchemy/ext/associationproxy.py
@@ -1093,14 +1093,11 @@ class _AssociationList(_AssociationCollection):
         col.append(item)
 
     def count(self, value):
-        return sum(
-            [
-                1
-                for _ in util.itertools_filter(
-                    lambda v: v == value, iter(self)
-                )
-            ]
-        )
+        count = 0
+        for v in self:
+            if v == value:
+                count += 1
+        return count
 
     def extend(self, values):
         for v in values:
diff --git a/lib/sqlalchemy/ext/asyncio/session.py b/lib/sqlalchemy/ext/asyncio/session.py
index d2c969056..89da4b497 100644
--- a/lib/sqlalchemy/ext/asyncio/session.py
+++ b/lib/sqlalchemy/ext/asyncio/session.py
@@ -540,7 +540,7 @@ class AsyncSession(ReversibleProxy):
         await self.close()
 
     def _maker_context_manager(self):
-        # no @contextlib.asynccontextmanager until python3.7, gr
+        # TODO: can this use asynccontextmanager ??
         return _AsyncSessionContextManager(self)
 
 
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index efdafff96..e91277311 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -13,6 +13,7 @@ compiled result to be fully cached.
 
 """
 
+import collections.abc as collections_abc
 import logging
 
 from .. import exc as sa_exc
@@ -24,7 +25,6 @@ from ..orm.session import Session
 from ..sql import func
 from ..sql import literal_column
 from ..sql import util as sql_util
-from ..util import collections_abc
 
 
 log = logging.getLogger(__name__)
diff --git a/lib/sqlalchemy/ext/indexable.py b/lib/sqlalchemy/ext/indexable.py
index 313ad11af..70673ac26 100644
--- a/lib/sqlalchemy/ext/indexable.py
+++ b/lib/sqlalchemy/ext/indexable.py
@@ -221,8 +221,6 @@ The above query will render::
     WHERE CAST(person.data ->> %(data_1)s AS INTEGER) < %(param_1)s
 
 """  # noqa
-from __future__ import absolute_import
-
 from .. import inspect
 from .. import util
 from ..ext.hybrid import hybrid_property
diff --git a/lib/sqlalchemy/ext/serializer.py b/lib/sqlalchemy/ext/serializer.py
index 18a54e079..c9bff9d28 100644
--- a/lib/sqlalchemy/ext/serializer.py
+++ b/lib/sqlalchemy/ext/serializer.py
@@ -53,6 +53,8 @@ needed for:
 
 """
 
+from io import BytesIO
+import pickle
 import re
 
 from .. import Column
@@ -64,9 +66,6 @@ from ..orm.mapper import Mapper
 from ..orm.session import Session
 from ..util import b64decode
 from ..util import b64encode
-from ..util import byte_buffer
-from ..util import pickle
-from ..util import text_type
 
 
 __all__ = ["Serializer", "Deserializer", "dumps", "loads"]
@@ -92,11 +91,9 @@ def Serializer(*args, **kw):
                     pickle.dumps(obj._annotations["parententity"].class_)
                 )
             else:
-                id_ = "table:" + text_type(obj.key)
+                id_ = f"table:{obj.key}"
         elif isinstance(obj, Column) and isinstance(obj.table, Table):
-            id_ = (
-                "column:" + text_type(obj.table.key) + ":" + text_type(obj.key)
-            )
+            id_ = f"column:{obj.table.key}:{obj.key}"
         elif isinstance(obj, Session):
             id_ = "session:"
         elif isinstance(obj, Engine):
@@ -129,7 +126,7 @@ def Deserializer(file, metadata=None, scoped_session=None, engine=None):
             return None
 
     def persistent_load(id_):
-        m = our_ids.match(text_type(id_))
+        m = our_ids.match(str(id_))
         if not m:
             return None
         else:
@@ -165,13 +162,13 @@ def Deserializer(file, metadata=None, scoped_session=None, engine=None):
 
 
 def dumps(obj, protocol=pickle.HIGHEST_PROTOCOL):
-    buf = byte_buffer()
+    buf = BytesIO()
     pickler = Serializer(buf, protocol)
     pickler.dump(obj)
     return buf.getvalue()
 
 
 def loads(data, metadata=None, scoped_session=None, engine=None):
-    buf = byte_buffer(data)
+    buf = BytesIO(data)
     unpickler = Deserializer(buf, metadata, scoped_session, engine)
     return unpickler.load()
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index dc5813866..aa48bf496 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -15,6 +15,8 @@ defines a large part of the ORM's interactivity.
 """
 
 import operator
+from typing import Generic
+from typing import TypeVar
 
 from . import collections
 from . import exc as orm_exc
@@ -366,13 +368,8 @@ def _queryable_attribute_unreduce(key, mapped_class, parententity, entity):
         return getattr(entity, key)
 
 
-if util.py3k:
-    from typing import TypeVar, Generic
-
-    _T = TypeVar("_T")
-    _Generic_T = Generic[_T]
-else:
-    _Generic_T = type("_Generic_T", (), {})
+_T = TypeVar("_T")
+_Generic_T = Generic[_T]
 
 
 class Mapped(QueryableAttribute, _Generic_T):
@@ -1555,12 +1552,7 @@ class CollectionAttributeImpl(AttributeImpl):
                 if hasattr(iterable, "_sa_iterator"):
                     iterable = iterable._sa_iterator()
                 elif setting_type is dict:
-                    if util.py3k:
-                        iterable = iterable.values()
-                    else:
-                        iterable = getattr(
-                            iterable, "itervalues", iterable.values
-                        )()
+                    iterable = iterable.values()
                 else:
                     iterable = iter(iterable)
         new_values = list(iterable)
diff --git a/lib/sqlalchemy/orm/collections.py b/lib/sqlalchemy/orm/collections.py
index 01d77e92b..ccb88866e 100644
--- a/lib/sqlalchemy/orm/collections.py
+++ b/lib/sqlalchemy/orm/collections.py
@@ -1615,9 +1615,7 @@ __interfaces = {
         _set_decorators(),
     ),
     # decorators are required for dicts and object collections.
-    dict: ({"iterator": "values"}, _dict_decorators())
-    if util.py3k
-    else ({"iterator": "itervalues"}, _dict_decorators()),
+    dict: ({"iterator": "values"}, _dict_decorators()),
 }
 
 
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 986daf7eb..4a3d5286b 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -1434,11 +1434,9 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             aliased_generation = flags["aliased_generation"]
 
             # do a quick inspect to accommodate for a lambda
-            if right is not None and not isinstance(right, util.string_types):
+            if right is not None and not isinstance(right, str):
                 right = inspect(right)
-            if onclause is not None and not isinstance(
-                onclause, util.string_types
-            ):
+            if onclause is not None and not isinstance(onclause, str):
                 onclause = inspect(onclause)
 
             # legacy vvvvvvvvvvvvvvvvvvvvvvvvvv
@@ -1459,9 +1457,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             # legacy ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
             if (
-                isinstance(
-                    right, (interfaces.PropComparator, util.string_types)
-                )
+                isinstance(right, (interfaces.PropComparator, str))
                 and onclause is None
             ):
                 onclause = right
@@ -1481,7 +1477,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             else:
                 of_type = None
 
-            if isinstance(onclause, util.string_types):
+            if isinstance(onclause, str):
                 # string given, e.g. query(Foo).join("bar").
                 # we look to the left entity or what we last joined
                 # towards
diff --git a/lib/sqlalchemy/orm/decl_api.py b/lib/sqlalchemy/orm/decl_api.py
index 21bb9d81b..6bc857094 100644
--- a/lib/sqlalchemy/orm/decl_api.py
+++ b/lib/sqlalchemy/orm/decl_api.py
@@ -5,8 +5,6 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 """Public API functions and helpers for declarative."""
-from __future__ import absolute_import
-
 import itertools
 import re
 import weakref
diff --git a/lib/sqlalchemy/orm/decl_base.py b/lib/sqlalchemy/orm/decl_base.py
index bab890b82..b92ae5aa4 100644
--- a/lib/sqlalchemy/orm/decl_base.py
+++ b/lib/sqlalchemy/orm/decl_base.py
@@ -5,8 +5,6 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 """Internal implementation for declarative."""
-from __future__ import absolute_import
-
 import collections
 import weakref
 
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index c425f012b..264161085 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -16,8 +16,6 @@ are exposed when inspecting mappings.
 
 """
 
-from __future__ import absolute_import
-
 import collections
 
 from . import exc as orm_exc
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 94ad7b80d..d6ee9b7a7 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -12,7 +12,6 @@ the functions here are called primarily by Query, Mapper,
 as well as some of the attribute loading strategies.
 
 """
-from __future__ import absolute_import
 
 from . import attributes
 from . import exc as orm_exc
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 4de12b88c..60083bf4d 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -14,9 +14,9 @@ This is a semi-private module; the main configurational API of the ORM is
 available in :class:`~sqlalchemy.orm.`.
 
 """
-from __future__ import absolute_import
 
 from collections import deque
+from functools import reduce
 from itertools import chain
 import sys
 import weakref
@@ -1153,9 +1153,7 @@ class Mapper(
         if with_polymorphic == "*":
             self.with_polymorphic = ("*", None)
         elif isinstance(with_polymorphic, (tuple, list)):
-            if isinstance(
-                with_polymorphic[0], util.string_types + (tuple, list)
-            ):
+            if isinstance(with_polymorphic[0], (str, tuple, list)):
                 self.with_polymorphic = with_polymorphic
             else:
                 self.with_polymorphic = (with_polymorphic, None)
@@ -1500,7 +1498,7 @@ class Mapper(
         if self.polymorphic_on is not None:
             setter = True
 
-            if isinstance(self.polymorphic_on, util.string_types):
+            if isinstance(self.polymorphic_on, str):
                 # polymorphic_on specified as a string - link
                 # it to mapped ColumnProperty
                 try:
@@ -3314,7 +3312,7 @@ class Mapper(
             cols = set(table.c)
             for m in self.iterate_to_root():
                 if m._inherits_equated_pairs and cols.intersection(
-                    util.reduce(
+                    reduce(
                         set.union,
                         [l.proxy_set for l, r in m._inherits_equated_pairs],
                     )
diff --git a/lib/sqlalchemy/orm/path_registry.py b/lib/sqlalchemy/orm/path_registry.py
index 6bebbd006..0aa9de817 100644
--- a/lib/sqlalchemy/orm/path_registry.py
+++ b/lib/sqlalchemy/orm/path_registry.py
@@ -8,6 +8,7 @@
 
 """
 
+from functools import reduce
 from itertools import chain
 import logging
 
@@ -221,7 +222,7 @@ class PathRegistry(HasCacheKey):
 
     @classmethod
     def coerce(cls, raw):
-        return util.reduce(lambda prev, next: prev[next], raw, cls.root)
+        return reduce(lambda prev, next: prev[next], raw, cls.root)
 
     def token(self, token):
         if token.endswith(":" + _WILDCARD_TOKEN):
@@ -232,7 +233,7 @@ class PathRegistry(HasCacheKey):
             raise exc.ArgumentError("invalid token: %s" % token)
 
     def __add__(self, other):
-        return util.reduce(lambda prev, next: prev[next], other.path, self)
+        return reduce(lambda prev, next: prev[next], other.path, self)
 
     def __repr__(self):
         return "%s(%r)" % (self.__class__.__name__, self.path)
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 3c0c637a2..6dd827b43 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -13,9 +13,9 @@ The functions here are called only by the unit of work functions
 in unitofwork.py.
 
 """
-
 from itertools import chain
 from itertools import groupby
+from itertools import zip_longest
 import operator
 
 from . import attributes
@@ -1171,7 +1171,7 @@ def _emit_insert_statements(
                         last_inserted_params,
                         inserted_primary_key,
                         returned_defaults,
-                    ) in util.zip_longest(
+                    ) in zip_longest(
                         records,
                         c.context.compiled_parameters,
                         c.inserted_primary_key_rows,
@@ -2189,7 +2189,7 @@ class BulkORMUpdate(UpdateDMLState, BulkUDCompileState):
         for k, v in kv_iterator:
             k = coercions.expect(roles.DMLColumnRole, k)
 
-            if isinstance(k, util.string_types):
+            if isinstance(k, str):
                 desc = _entity_namespace_key(mapper, k, default=NO_VALUE)
                 if desc is NO_VALUE:
                     values.append(
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index fa230d109..de46f84a7 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -11,7 +11,6 @@ This is a private module which defines the behavior of individual ORM-
 mapped attributes.
 
 """
-from __future__ import absolute_import
 
 from . import attributes
 from .descriptor_props import CompositeProperty
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 8cf253efb..38eb33bc4 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -18,6 +18,7 @@ ORM session, whereas the ``Select`` construct interacts directly with the
 database to return iterable result sets.
 
 """
+import collections.abc as collections_abc
 import itertools
 import operator
 import types
@@ -68,7 +69,6 @@ from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
 from ..sql.selectable import SelectBase
 from ..sql.selectable import SelectStatementGrouping
 from ..sql.visitors import InternalTraversal
-from ..util import collections_abc
 
 __all__ = ["Query", "QueryContext", "aliased"]
 
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index b692daf21..b4aef874f 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -13,8 +13,6 @@ SQL annotation and aliasing behavior focused on the `primaryjoin`
 and `secondaryjoin` aspects of :func:`_orm.relationship`.
 
 """
-from __future__ import absolute_import
-
 import collections
 import re
 import weakref
@@ -2108,7 +2106,7 @@ class RelationshipProperty(StrategizedProperty):
 
         mapperlib = util.preloaded.orm_mapper
 
-        if isinstance(self.argument, util.string_types):
+        if isinstance(self.argument, str):
             argument = self._clsregistry_resolve_name(self.argument)()
 
         elif callable(self.argument) and not isinstance(
@@ -2183,7 +2181,7 @@ class RelationshipProperty(StrategizedProperty):
         ):
             attr_value = getattr(self, attr)
 
-            if isinstance(attr_value, util.string_types):
+            if isinstance(attr_value, str):
                 setattr(
                     self,
                     attr,
@@ -2420,7 +2418,7 @@ class RelationshipProperty(StrategizedProperty):
         if self.parent.non_primary:
             return
         if self.backref is not None and not self.back_populates:
-            if isinstance(self.backref, util.string_types):
+            if isinstance(self.backref, str):
                 backref_key, kwargs = self.backref, {}
             else:
                 backref_key, kwargs = self.backref
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index e18e35847..8212a111d 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -6,7 +6,7 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 """Provides the Session class and related utilities."""
 
-
+import contextlib
 import itertools
 import sys
 import weakref
@@ -1134,7 +1134,7 @@ class Session(_SessionClassMethods):
     def __exit__(self, type_, value, traceback):
         self.close()
 
-    @util.contextmanager
+    @contextlib.contextmanager
     def _maker_context_manager(self):
         with self:
             with self.begin():
@@ -2113,7 +2113,7 @@ class Session(_SessionClassMethods):
         return loading.get_from_identity(self, mapper, key, passive)
 
     @property
-    @util.contextmanager
+    @contextlib.contextmanager
     def no_autoflush(self):
         """Return a context manager that disables autoflush.
 
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 130ff2d1e..ff4ac9d33 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -7,7 +7,6 @@
 
 """sqlalchemy.orm.interfaces.LoaderStrategy
    implementations, and related MapperOptions."""
-from __future__ import absolute_import
 
 import collections
 import itertools
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index 997fca1ac..eccb42164 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -301,7 +301,7 @@ class Load(Generative, LoaderOption):
                     "refer to a mapped entity" % (path.prop,)
                 )
 
-        if isinstance(attr, util.string_types):
+        if isinstance(attr, str):
 
             default_token = attr.endswith(_DEFAULT_TOKEN)
             attr_str_name = attr
@@ -643,7 +643,7 @@ class Load(Generative, LoaderOption):
         i = -1
 
         for i, (c_token, p_token) in enumerate(zip(to_chop, path.path)):
-            if isinstance(c_token, util.string_types):
+            if isinstance(c_token, str):
                 # TODO: this is approximated from the _UnboundLoad
                 # version and probably has issues, not fully covered.
 
@@ -816,9 +816,7 @@ class _UnboundLoad(Load):
         cloned.strategy = self.strategy
         if self.path:
             attr = self.path[-1]
-            if isinstance(attr, util.string_types) and attr.endswith(
-                _DEFAULT_TOKEN
-            ):
+            if isinstance(attr, str) and attr.endswith(_DEFAULT_TOKEN):
                 attr = attr.split(":")[0] + ":" + _WILDCARD_TOKEN
             cloned._generate_path(
                 parent.path + self.path[0:-1], attr, self.strategy, None
@@ -851,7 +849,7 @@ class _UnboundLoad(Load):
     def _generate_path(self, path, attr, for_strategy, wildcard_key):
         if (
             wildcard_key
-            and isinstance(attr, util.string_types)
+            and isinstance(attr, str)
             and attr in (_WILDCARD_TOKEN, _DEFAULT_TOKEN)
         ):
             if attr == _DEFAULT_TOKEN:
@@ -914,7 +912,7 @@ class _UnboundLoad(Load):
         opt = _UnboundLoad()
 
         def _split_key(key):
-            if isinstance(key, util.string_types):
+            if isinstance(key, str):
                 # coerce fooload('*') into "default loader strategy"
                 if key == _WILDCARD_TOKEN:
                     return (_DEFAULT_TOKEN,)
@@ -945,7 +943,7 @@ class _UnboundLoad(Load):
         for i, (c_token, (p_entity, p_prop)) in enumerate(
             zip(to_chop, path.pairs())
         ):
-            if isinstance(c_token, util.string_types):
+            if isinstance(c_token, str):
                 if i == 0 and c_token.endswith(":" + _DEFAULT_TOKEN):
                     return to_chop
                 elif (
@@ -1045,7 +1043,7 @@ class _UnboundLoad(Load):
         # what entity we are referring towards.
         token = start_path[0]
 
-        if isinstance(token, util.string_types):
+        if isinstance(token, str):
             entity = self._find_entity_basestring(entities, token, raiseerr)
         elif isinstance(token, PropComparator):
             prop = token.property
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 3295bd39e..8403dd5ac 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -82,7 +82,7 @@ class CascadeOptions(frozenset):
     )
 
     def __new__(cls, value_list):
-        if isinstance(value_list, util.string_types) or value_list is None:
+        if isinstance(value_list, str) or value_list is None:
             return cls.from_string(value_list)
         values = set(value_list)
         if values.difference(cls._allowed_cascades):
@@ -1645,7 +1645,7 @@ class _ORMJoin(expression.Join):
         # then the "_joined_from_info" concept can go
         left_orm_info = getattr(left, "_joined_from_info", left_info)
         self._joined_from_info = right_info
-        if isinstance(onclause, util.string_types):
+        if isinstance(onclause, str):
             onclause = getattr(left_orm_info.entity, onclause)
         # ####
 
@@ -1870,7 +1870,7 @@ def with_parent(instance, prop, from_entity=None):
       .. versionadded:: 1.2
 
     """
-    if isinstance(prop, util.string_types):
+    if isinstance(prop, str):
         util.warn_deprecated_20(
             "Using strings to indicate relationship names in the ORM "
             "with_parent() function is deprecated and will be removed "
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index b57da3289..2c74dd523 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -10,7 +10,10 @@
 """
 
 
+import collections.abc as collections_abc
+from functools import reduce
 import itertools
+from itertools import zip_longest
 import operator
 import re
 
@@ -27,7 +30,6 @@ from .. import util
 from ..util import HasMemoized
 from ..util import hybridmethod
 
-
 coercions = None
 elements = None
 type_api = None
@@ -176,7 +178,7 @@ def _cloned_difference(a, b):
     )
 
 
-class _DialectArgView(util.collections_abc.MutableMapping):
+class _DialectArgView(collections_abc.MutableMapping):
     """A dictionary view of dialect-level arguments in the form
     <dialectname>_<argument_name>.
 
@@ -236,7 +238,7 @@ class _DialectArgView(util.collections_abc.MutableMapping):
         )
 
 
-class _DialectArgDict(util.collections_abc.MutableMapping):
+class _DialectArgDict(collections_abc.MutableMapping):
     """A dictionary view of dialect-level arguments for a specific
     dialect.
 
@@ -615,7 +617,7 @@ class _MetaOptions(type):
         return o1
 
 
-class Options(util.with_metaclass(_MetaOptions)):
+class Options(metaclass=_MetaOptions):
     """A cacheable option dictionary with defaults."""
 
     def __init__(self, **kw):
@@ -638,7 +640,7 @@ class Options(util.with_metaclass(_MetaOptions)):
     def __eq__(self, other):
         # TODO: very inefficient.  This is used only in test suites
         # right now.
-        for a, b in util.zip_longest(self._cache_attrs, other._cache_attrs):
+        for a, b in zip_longest(self._cache_attrs, other._cache_attrs):
             if getattr(self, a) != getattr(other, b):
                 return False
         return True
@@ -1238,7 +1240,7 @@ class ColumnCollection:
         try:
             return self._index[key]
         except KeyError as err:
-            if isinstance(key, util.int_types):
+            if isinstance(key, int):
                 util.raise_(IndexError(key), replace_context=err)
             else:
                 raise
@@ -1251,7 +1253,7 @@ class ColumnCollection:
 
     def __contains__(self, key):
         if key not in self._index:
-            if not isinstance(key, util.string_types):
+            if not isinstance(key, str):
                 raise exc.ArgumentError(
                     "__contains__ requires a string argument"
                 )
@@ -1263,7 +1265,7 @@ class ColumnCollection:
         """Compare this :class:`_expression.ColumnCollection` to another
         based on the names of the keys"""
 
-        for l, r in util.zip_longest(self, other):
+        for l, r in zip_longest(self, other):
             if l is not r:
                 return False
         else:
@@ -1359,7 +1361,7 @@ class ColumnCollection:
     def contains_column(self, col):
         """Checks if a column object exists in this collection"""
         if col not in self._colset:
-            if isinstance(col, util.string_types):
+            if isinstance(col, str):
                 raise exc.ArgumentError(
                     "contains_column cannot be used with string arguments. "
                     "Use ``col_name in table.c`` instead."
@@ -1451,7 +1453,7 @@ class ColumnCollection:
                     # columns that have no reference to the target
                     # column (also occurs with CompoundSelect)
 
-                    col_distance = util.reduce(
+                    col_distance = reduce(
                         operator.add,
                         [
                             sc._annotations.get("weight", 1)
@@ -1459,7 +1461,7 @@ class ColumnCollection:
                             if sc.shares_lineage(column)
                         ],
                     )
-                    c_distance = util.reduce(
+                    c_distance = reduce(
                         operator.add,
                         [
                             sc._annotations.get("weight", 1)
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index f051ba12f..480d2c680 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -5,6 +5,7 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+import collections.abc as collections_abc
 import numbers
 import re
 
@@ -17,7 +18,6 @@ from .visitors import Visitable
 from .. import exc
 from .. import inspection
 from .. import util
-from ..util import collections_abc
 
 
 elements = None
@@ -224,7 +224,7 @@ def expect_col_expression_collection(role, expressions):
         column = None
 
         resolved = expect(role, expr)
-        if isinstance(resolved, util.string_types):
+        if isinstance(resolved, str):
             strname = resolved = expr
         else:
             cols = []
@@ -303,7 +303,7 @@ class _ReturnsStringKey:
     def _implicit_coercions(
         self, original_element, resolved, argname=None, **kw
     ):
-        if isinstance(original_element, util.string_types):
+        if isinstance(original_element, str):
             return original_element
         else:
             self._raise_for_expected(original_element, argname, resolved)
@@ -362,7 +362,7 @@ class _NoTextCoercion:
     __slots__ = ()
 
     def _literal_coercion(self, element, argname=None, **kw):
-        if isinstance(element, util.string_types) and issubclass(
+        if isinstance(element, str) and issubclass(
             elements.TextClause, self._role_class
         ):
             _no_text_coercion(element, argname)
@@ -380,7 +380,7 @@ class _CoerceLiterals:
         return _no_text_coercion(element, argname)
 
     def _literal_coercion(self, element, argname=None, **kw):
-        if isinstance(element, util.string_types):
+        if isinstance(element, str):
             if self._coerce_star and element == "*":
                 return elements.ColumnClause("*", is_literal=True)
             else:
@@ -542,7 +542,7 @@ class InElementImpl(RoleImpl):
 
     def _literal_coercion(self, element, expr, operator, **kw):
         if isinstance(element, collections_abc.Iterable) and not isinstance(
-            element, util.string_types
+            element, str
         ):
             non_literal_expressions = {}
             element = list(element)
@@ -729,7 +729,7 @@ class TruncatedLabelImpl(_StringOnly, RoleImpl):
     def _implicit_coercions(
         self, original_element, resolved, argname=None, **kw
     ):
-        if isinstance(original_element, util.string_types):
+        if isinstance(original_element, str):
             return resolved
         else:
             self._raise_for_expected(original_element, argname, resolved)
@@ -844,7 +844,7 @@ class StatementImpl(_CoerceLiterals, RoleImpl):
 
     def _post_coercion(self, resolved, original_element, argname=None, **kw):
         if resolved is not original_element and not isinstance(
-            original_element, util.string_types
+            original_element, str
         ):
             # use same method as Connection uses; this will later raise
             # ObjectNotExecutableError
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 29aa57faa..0dd61d675 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -22,12 +22,13 @@ To generate user-defined SQL strings, see
 :doc:`/ext/compiler`.
 
 """
-
 import collections
+import collections.abc as collections_abc
 import contextlib
 import itertools
 import operator
 import re
+from time import perf_counter
 
 from . import base
 from . import coercions
@@ -453,7 +454,7 @@ class Compiled:
                 self.string = self.preparer._render_schema_translates(
                     self.string, schema_translate_map
                 )
-        self._gen_time = util.perf_counter()
+        self._gen_time = perf_counter()
 
     def _execute_on_connection(
         self, connection, distilled_params, execution_options
@@ -505,7 +506,7 @@ class Compiled:
         return self.construct_params()
 
 
-class TypeCompiler(util.with_metaclass(util.EnsureKWArgType, object)):
+class TypeCompiler(metaclass=util.EnsureKWArgType):
     """Produces DDL specification for TypeEngine objects."""
 
     ensure_kwarg = r"visit_\w+"
@@ -2026,10 +2027,8 @@ class SQLCompiler(Compiled):
 
         elif typ_dialect_impl._is_tuple_type or (
             typ_dialect_impl._isnull
-            and isinstance(values[0], util.collections_abc.Sequence)
-            and not isinstance(
-                values[0], util.string_types + util.binary_types
-            )
+            and isinstance(values[0], collections_abc.Sequence)
+            and not isinstance(values[0], (str, bytes))
         ):
 
             replacement_expression = (
@@ -2077,10 +2076,8 @@ class SQLCompiler(Compiled):
 
         elif typ_dialect_impl._is_tuple_type or (
             typ_dialect_impl._isnull
-            and isinstance(values[0], util.collections_abc.Sequence)
-            and not isinstance(
-                values[0], util.string_types + util.binary_types
-            )
+            and isinstance(values[0], collections_abc.Sequence)
+            and not isinstance(values[0], (str, bytes))
         ):
             assert not typ_dialect_impl._is_array
             to_update = [
@@ -4355,7 +4352,7 @@ class DDLCompiler(Compiled):
             except exc.CompileError as ce:
                 util.raise_(
                     exc.CompileError(
-                        util.u("(in table '%s', column '%s'): %s")
+                        "(in table '%s', column '%s'): %s"
                         % (table.description, column.name, ce.args[0])
                     ),
                     from_=ce,
@@ -4628,7 +4625,7 @@ class DDLCompiler(Compiled):
 
     def get_column_default_string(self, column):
         if isinstance(column.server_default, schema.DefaultClause):
-            if isinstance(column.server_default.arg, util.string_types):
+            if isinstance(column.server_default.arg, str):
                 return self.sql_compiler.render_literal_value(
                     column.server_default.arg, sqltypes.STRINGTYPE
                 )
@@ -5126,14 +5123,14 @@ class IdentifierPreparer:
         return (
             lc_value in self.reserved_words
             or value[0] in self.illegal_initial_characters
-            or not self.legal_characters.match(util.text_type(value))
+            or not self.legal_characters.match(str(value))
             or (lc_value != value)
         )
 
     def _requires_quotes_illegal_chars(self, value):
         """Return True if the given identifier requires quoting, but
         not taking case convention into account."""
-        return not self.legal_characters.match(util.text_type(value))
+        return not self.legal_characters.match(str(value))
 
     def quote_schema(self, schema, force=None):
         """Conditionally quote a schema name.
diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index a9c9cb4c1..a313257ca 100644
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -373,7 +373,7 @@ def _scan_cols(
         cols = [
             stmt.table.c[key]
             for key in parameter_ordering
-            if isinstance(key, util.string_types) and key in stmt.table.c
+            if isinstance(key, str) and key in stmt.table.c
         ] + [c for c in stmt.table.c if c.key not in ordered_keys]
 
     else:
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index f732ff2b0..74e7df821 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -180,7 +180,7 @@ class DDLElement(roles.DDLRole, Executable, _DDLCompiles):
         self.state = state
 
     def _should_execute(self, target, bind, **kw):
-        if isinstance(self.dialect, util.string_types):
+        if isinstance(self.dialect, str):
             if self.dialect != bind.engine.name:
                 return False
         elif isinstance(self.dialect, (tuple, list, set)):
@@ -288,7 +288,7 @@ class DDL(DDLElement):
 
         """
 
-        if not isinstance(statement, util.string_types):
+        if not isinstance(statement, str):
             raise exc.ArgumentError(
                 "Expected a string or unicode SQL statement, got '%r'"
                 % statement
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index e45c6b888..93cd912a7 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -9,6 +9,8 @@ Provide :class:`_expression.Insert`, :class:`_expression.Update` and
 :class:`_expression.Delete`.
 
 """
+import collections.abc as collections_abc
+
 from sqlalchemy.types import NullType
 from . import coercions
 from . import roles
@@ -31,7 +33,6 @@ from .selectable import ReturnsRows
 from .visitors import InternalTraversal
 from .. import exc
 from .. import util
-from ..util import collections_abc
 
 
 class DMLState(CompileState):
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index ca65f2112..76633cdd8 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -10,8 +10,6 @@
 
 """
 
-from __future__ import unicode_literals
-
 import itertools
 import operator
 import re
@@ -3869,7 +3867,7 @@ class BinaryExpression(ColumnElement):
     ):
         # allow compatibility with libraries that
         # refer to BinaryExpression directly and pass strings
-        if isinstance(operator, util.string_types):
+        if isinstance(operator, str):
             operator = operators.custom_op(operator)
         self._orig = (left.__hash__(), right.__hash__())
         self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
@@ -4638,10 +4636,7 @@ class NamedColumn(ColumnElement):
 
     @util.memoized_property
     def description(self):
-        if util.py3k:
-            return self.name
-        else:
-            return self.name.encode("ascii", "backslashreplace")
+        return self.name
 
     @HasMemoized.memoized_attribute
     def _tq_key_label(self):
@@ -5068,7 +5063,7 @@ class ReleaseSavepointClause(_IdentifiedClause):
     __visit_name__ = "release_savepoint"
 
 
-class quoted_name(util.MemoizedSlots, util.text_type):
+class quoted_name(util.MemoizedSlots, str):
     """Represent a SQL identifier combined with quoting preferences.
 
     :class:`.quoted_name` is a Python unicode/str subclass which
@@ -5138,19 +5133,19 @@ class quoted_name(util.MemoizedSlots, util.text_type):
         return self
 
     def __reduce__(self):
-        return quoted_name, (util.text_type(self), self.quote)
+        return quoted_name, (str(self), self.quote)
 
     def _memoized_method_lower(self):
         if self.quote:
             return self
         else:
-            return util.text_type(self).lower()
+            return str(self).lower()
 
     def _memoized_method_upper(self):
         if self.quote:
             return self
         else:
-            return util.text_type(self).upper()
+            return str(self).upper()
 
 
 def _find_columns(clause):
@@ -5238,7 +5233,7 @@ class _truncated_label(quoted_name):
         return super(_truncated_label, cls).__new__(cls, value, quote)
 
     def __reduce__(self):
-        return self.__class__, (util.text_type(self), self.quote)
+        return self.__class__, (str(self), self.quote)
 
     def apply_map(self, map_):
         return self
@@ -5324,26 +5319,26 @@ class _anonymous_label(_truncated_label):
 
     def __add__(self, other):
         if "%" in other and not isinstance(other, _anonymous_label):
-            other = util.text_type(other).replace("%", "%%")
+            other = str(other).replace("%", "%%")
         else:
-            other = util.text_type(other)
+            other = str(other)
 
         return _anonymous_label(
             quoted_name(
-                util.text_type.__add__(self, other),
+                str.__add__(self, other),
                 self.quote,
             )
         )
 
     def __radd__(self, other):
         if "%" in other and not isinstance(other, _anonymous_label):
-            other = util.text_type(other).replace("%", "%%")
+            other = str(other).replace("%", "%%")
         else:
-            other = util.text_type(other)
+            other = str(other)
 
         return _anonymous_label(
             quoted_name(
-                util.text_type.__add__(other, self),
+                str.__add__(other, self),
                 self.quote,
             )
         )
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 31f4fab0d..901a3a77c 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -55,7 +55,7 @@ def register_function(identifier, fn, package="_default"):
     """
     reg = _registry[package]
 
-    identifier = util.text_type(identifier).lower()
+    identifier = str(identifier).lower()
 
     # Check if a function with the same identifier is registered.
     if identifier in reg:
@@ -909,7 +909,7 @@ class _GenericMeta(TraversibleType):
         super(_GenericMeta, cls).__init__(clsname, bases, clsdict)
 
 
-class GenericFunction(util.with_metaclass(_GenericMeta, Function)):
+class GenericFunction(Function, metaclass=_GenericMeta):
     """Define a 'generic' function.
 
     A generic function is a pre-established :class:`.Function`
diff --git a/lib/sqlalchemy/sql/lambdas.py b/lib/sqlalchemy/sql/lambdas.py
index 110260799..2256fb5a9 100644
--- a/lib/sqlalchemy/sql/lambdas.py
+++ b/lib/sqlalchemy/sql/lambdas.py
@@ -5,6 +5,7 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+import collections.abc as collections_abc
 import itertools
 import operator
 import sys
@@ -24,8 +25,6 @@ from .operators import ColumnOperators
 from .. import exc
 from .. import inspection
 from .. import util
-from ..util import collections_abc
-from ..util import compat
 
 _closure_per_cache_key = util.LRUCache(1000)
 
@@ -1111,7 +1110,7 @@ class AnalyzedFunction:
         code += "        return %s\n" % ", ".join("i%d" % i for i in argrange)
         code += "    return closure.__closure__"
         vars_ = {"o%d" % i: cell_values[i] for i in argrange}
-        compat.exec_(code, vars_, vars_)
+        exec(code, vars_, vars_)
         closure = vars_["make_cells"]()
 
         func = type(f)(
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index 675048cd0..6d45cd033 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -1394,7 +1394,7 @@ def _escaped_like_impl(fn, other, escape, autoescape):
         if escape is None:
             escape = "/"
 
-        if not isinstance(other, util.compat.string_types):
+        if not isinstance(other, str):
             raise TypeError("String value expected when autoescape=True")
 
         if escape not in ("%", "_"):
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 4bd49c468..eed2fbba1 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -28,8 +28,6 @@ Since these objects are part of the SQL expression language, they are usable
 as components in SQL expressions.
 
 """
-from __future__ import absolute_import
-
 import collections
 
 import sqlalchemy
@@ -1572,7 +1570,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
         type_ = kwargs.pop("type_", None)
         args = list(args)
         if args:
-            if isinstance(args[0], util.string_types):
+            if isinstance(args[0], str):
                 if name is not None:
                     raise exc.ArgumentError(
                         "May not pass name positionally and as a keyword."
@@ -1819,7 +1817,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
             )
 
         if self.index:
-            if isinstance(self.index, util.string_types):
+            if isinstance(self.index, str):
                 raise exc.ArgumentError(
                     "The 'index' keyword argument on Column is boolean only. "
                     "To create indexes with a specific name, create an "
@@ -1832,7 +1830,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
             )
 
         elif self.unique:
-            if isinstance(self.unique, util.string_types):
+            if isinstance(self.unique, str):
                 raise exc.ArgumentError(
                     "The 'unique' keyword argument on Column is boolean "
                     "only. To create unique constraints or indexes with a "
@@ -2125,7 +2123,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
 
         self._colspec = coercions.expect(roles.DDLReferredColumnRole, column)
 
-        if isinstance(self._colspec, util.string_types):
+        if isinstance(self._colspec, str):
             self._table_column = None
         else:
             self._table_column = self._colspec
@@ -2395,7 +2393,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
 
         """
 
-        if isinstance(self._colspec, util.string_types):
+        if isinstance(self._colspec, str):
 
             parenttable, tablekey, colname = self._resolve_col_tokens()
 
@@ -2472,7 +2470,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         table.foreign_keys.add(self)
         # set up remote ".column" attribute, or a note to pick it
         # up when the other Table/Column shows up
-        if isinstance(self._colspec, util.string_types):
+        if isinstance(self._colspec, str):
             parenttable, table_key, colname = self._resolve_col_tokens()
             fk_key = (table_key, colname)
             if table_key in parenttable.metadata.tables:
@@ -3068,9 +3066,7 @@ class DefaultClause(FetchedValue):
     has_argument = True
 
     def __init__(self, arg, for_update=False, _reflected=False):
-        util.assert_arg_type(
-            arg, (util.string_types[0], ClauseElement, TextClause), "arg"
-        )
+        util.assert_arg_type(arg, (str, ClauseElement, TextClause), "arg")
         super(DefaultClause, self).__init__(for_update)
         self.arg = arg
         self.reflected = _reflected
@@ -3264,7 +3260,7 @@ class ColumnCollectionMixin:
 
     def _col_expressions(self, table):
         return [
-            table.c[col] if isinstance(col, util.string_types) else col
+            table.c[col] if isinstance(col, str) else col
             for col in self._pending_colargs
         ]
 
@@ -4422,7 +4418,7 @@ class MetaData(SchemaItem):
             return "MetaData()"
 
     def __contains__(self, table_or_key):
-        if not isinstance(table_or_key, util.string_types):
+        if not isinstance(table_or_key, str):
             table_or_key = table_or_key.key
         return table_or_key in self.tables
 
@@ -4501,7 +4497,7 @@ class MetaData(SchemaItem):
     def _bind_to(self, bind):
         """Bind this MetaData to an Engine, Connection, string or URL."""
         url = util.preloaded.engine_url
-        if isinstance(bind, util.string_types + (url.URL,)):
+        if isinstance(bind, (str, url.URL)):
             self._bind = sqlalchemy.create_engine(bind)
         else:
             self._bind = bind
@@ -4838,7 +4834,7 @@ class ThreadLocalMetaData(MetaData):
     def _bind_to(self, bind):
         """Bind to a Connectable in the caller's thread."""
         url = util.preloaded.engine_url
-        if isinstance(bind, util.string_types + (url.URL,)):
+        if isinstance(bind, (str, url.URL)):
             try:
                 self.context._engine = self.__engines[bind]
             except KeyError:
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index a04b205b5..a77cd173b 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1624,10 +1624,7 @@ class AliasedReturnsRows(NoInit, FromClause):
         if isinstance(name, _anonymous_label):
             name = "anon_1"
 
-        if util.py3k:
-            return name
-        else:
-            return name.encode("ascii", "backslashreplace")
+        return name
 
     @property
     def original(self):
@@ -2728,10 +2725,7 @@ class TableClause(roles.DMLTableRole, Immutable, FromClause):
 
     @util.memoized_property
     def description(self):
-        if util.py3k:
-            return self.name
-        else:
-            return self.name.encode("ascii", "backslashreplace")
+        return self.name
 
     def append_column(self, c, **kw):
         existing = c.table
@@ -5267,9 +5261,7 @@ class Select(
                 isinstance(args[0], list)
                 or (
                     hasattr(args[0], "__iter__")
-                    and not isinstance(
-                        args[0], util.string_types + (ClauseElement,)
-                    )
+                    and not isinstance(args[0], (str, ClauseElement))
                     and inspect(args[0], raiseerr=False) is None
                     and not hasattr(args[0], "__clause_element__")
                 )
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 1687c9f29..8874f0a83 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -9,9 +9,11 @@
 
 """
 
+import collections.abc as collections_abc
 import datetime as dt
 import decimal
 import json
+import pickle
 
 from . import coercions
 from . import elements
@@ -38,10 +40,8 @@ from .. import exc
 from .. import inspection
 from .. import processors
 from .. import util
-from ..util import compat
 from ..util import langhelpers
 from ..util import OrderedDict
-from ..util import pickle
 
 
 class _LookupExpressionAdapter:
@@ -192,7 +192,7 @@ class String(Concatenable, TypeEngine):
 
     @property
     def python_type(self):
-        return util.text_type
+        return str
 
     def get_dbapi_type(self, dbapi):
         return dbapi.STRING
@@ -721,7 +721,7 @@ class _Binary(TypeEngine):
 
     @property
     def python_type(self):
-        return util.binary_type
+        return bytes
 
     # Python 3 - sqlite3 doesn't need the `Binary` conversion
     # here, though pg8000 does to indicate "bytea"
@@ -753,7 +753,7 @@ class _Binary(TypeEngine):
     def coerce_compared_value(self, op, value):
         """See :meth:`.TypeEngine.coerce_compared_value` for a description."""
 
-        if isinstance(value, util.string_types):
+        if isinstance(value, str):
             return self
         else:
             return super(_Binary, self).coerce_compared_value(op, value)
@@ -1328,9 +1328,7 @@ class Enum(Emulated, String, SchemaType):
             # here between an INSERT statement and a criteria used in a SELECT,
             # for now we're staying conservative w/ behavioral changes (perhaps
             # someone has a trigger that handles strings on INSERT)
-            if not self.validate_strings and isinstance(
-                elem, compat.string_types
-            ):
+            if not self.validate_strings and isinstance(elem, str):
                 return elem
             else:
                 util.raise_(
@@ -1513,8 +1511,7 @@ class PickleType(TypeDecorator):
 
         :param protocol: defaults to ``pickle.HIGHEST_PROTOCOL``.
 
-        :param pickler: defaults to cPickle.pickle or pickle.pickle if
-          cPickle is not available.  May be any object with
+        :param pickler: defaults to pickle.  May be any object with
           pickle-compatible ``dumps`` and ``loads`` methods.
 
         :param comparator: a 2-arg callable predicate used
@@ -2123,7 +2120,7 @@ class JSON(Indexable, TypeEngine):
             def process(value):
                 if int_processor and isinstance(value, int):
                     value = int_processor(value)
-                elif string_processor and isinstance(value, util.string_types):
+                elif string_processor and isinstance(value, str):
                     value = string_processor(value)
                 return value
 
@@ -2136,7 +2133,7 @@ class JSON(Indexable, TypeEngine):
             def process(value):
                 if int_processor and isinstance(value, int):
                     value = int_processor(value)
-                elif string_processor and isinstance(value, util.string_types):
+                elif string_processor and isinstance(value, str):
                     value = string_processor(value)
                 return value
 
@@ -2178,8 +2175,8 @@ class JSON(Indexable, TypeEngine):
         """Define comparison operations for :class:`_types.JSON`."""
 
         def _setup_getitem(self, index):
-            if not isinstance(index, util.string_types) and isinstance(
-                index, compat.collections_abc.Sequence
+            if not isinstance(index, str) and isinstance(
+                index, collections_abc.Sequence
             ):
                 index = coercions.expect(
                     roles.BinaryElementRole,
@@ -2982,15 +2979,10 @@ _type_map = {
     dt.time: Time(),
     dt.timedelta: Interval(),
     util.NoneType: NULLTYPE,
+    bytes: LargeBinary(),
+    str: Unicode(),
 }
 
-if util.py3k:
-    _type_map[bytes] = LargeBinary()  # noqa
-    _type_map[str] = Unicode()
-else:
-    _type_map[unicode] = Unicode()  # noqa
-    _type_map[str] = String()
-
 
 _type_map_get = _type_map.get
 
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 7973b535f..914a78dae 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -1,6 +1,8 @@
 from collections import deque
 from collections import namedtuple
+import collections.abc as collections_abc
 import itertools
+from itertools import zip_longest
 import operator
 
 from . import operators
@@ -8,9 +10,7 @@ from .visitors import ExtendedInternalTraversal
 from .visitors import InternalTraversal
 from .. import util
 from ..inspection import inspect
-from ..util import collections_abc
 from ..util import HasMemoized
-from ..util import py37
 
 SKIP_TRAVERSE = util.symbol("skip_traverse")
 COMPARE_FAILED = False
@@ -331,7 +331,7 @@ class CacheKey(namedtuple("CacheKey", ["key", "bindparams"])):
                 s1 = s1[idx]
                 s2 = s2[idx]
 
-            for idx, (e1, e2) in enumerate(util.zip_longest(s1, s2)):
+            for idx, (e1, e2) in enumerate(zip_longest(s1, s2)):
                 if idx < pickup_index:
                     continue
                 if e1 != e2:
@@ -669,38 +669,20 @@ class _CacheKey(ExtendedInternalTraversal):
         )
 
     def visit_dml_values(self, attrname, obj, parent, anon_map, bindparams):
-        if py37:
-            # in py37 we can assume two dictionaries created in the same
-            # insert ordering will retain that sorting
-            return (
-                attrname,
-                tuple(
-                    (
-                        k._gen_cache_key(anon_map, bindparams)
-                        if hasattr(k, "__clause_element__")
-                        else k,
-                        obj[k]._gen_cache_key(anon_map, bindparams),
-                    )
-                    for k in obj
-                ),
-            )
-        else:
-            expr_values = {k for k in obj if hasattr(k, "__clause_element__")}
-            if expr_values:
-                # expr values can't be sorted deterministically right now,
-                # so no cache
-                anon_map[NO_CACHE] = True
-                return ()
-
-            str_values = expr_values.symmetric_difference(obj)
-
-            return (
-                attrname,
-                tuple(
-                    (k, obj[k]._gen_cache_key(anon_map, bindparams))
-                    for k in sorted(str_values)
-                ),
-            )
+        # in py37 we can assume two dictionaries created in the same
+        # insert ordering will retain that sorting
+        return (
+            attrname,
+            tuple(
+                (
+                    k._gen_cache_key(anon_map, bindparams)
+                    if hasattr(k, "__clause_element__")
+                    else k,
+                    obj[k]._gen_cache_key(anon_map, bindparams),
+                )
+                for k in obj
+            ),
+        )
 
     def visit_dml_multi_values(
         self, attrname, obj, parent, anon_map, bindparams
@@ -1040,7 +1022,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
             for (
                 (left_attrname, left_visit_sym),
                 (right_attrname, right_visit_sym),
-            ) in util.zip_longest(
+            ) in zip_longest(
                 left._traverse_internals,
                 right._traverse_internals,
                 fillvalue=(None, None),
@@ -1098,7 +1080,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_has_cache_key_list(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for l, r in util.zip_longest(left, right, fillvalue=None):
+        for l, r in zip_longest(left, right, fillvalue=None):
             if l._gen_cache_key(self.anon_map[0], []) != r._gen_cache_key(
                 self.anon_map[1], []
             ):
@@ -1114,7 +1096,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_fromclause_canonical_column_collection(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for lcol, rcol in util.zip_longest(left, right, fillvalue=None):
+        for lcol, rcol in zip_longest(left, right, fillvalue=None):
             self.stack.append((lcol, rcol))
 
     def visit_fromclause_derived_column_collection(
@@ -1125,7 +1107,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_string_clauseelement_dict(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for lstr, rstr in util.zip_longest(
+        for lstr, rstr in zip_longest(
             sorted(left), sorted(right), fillvalue=None
         ):
             if lstr != rstr:
@@ -1135,23 +1117,23 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_clauseelement_tuples(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for ltup, rtup in util.zip_longest(left, right, fillvalue=None):
+        for ltup, rtup in zip_longest(left, right, fillvalue=None):
             if ltup is None or rtup is None:
                 return COMPARE_FAILED
 
-            for l, r in util.zip_longest(ltup, rtup, fillvalue=None):
+            for l, r in zip_longest(ltup, rtup, fillvalue=None):
                 self.stack.append((l, r))
 
     def visit_clauseelement_list(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for l, r in util.zip_longest(left, right, fillvalue=None):
+        for l, r in zip_longest(left, right, fillvalue=None):
             self.stack.append((l, r))
 
     def visit_clauseelement_tuple(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for l, r in util.zip_longest(left, right, fillvalue=None):
+        for l, r in zip_longest(left, right, fillvalue=None):
             self.stack.append((l, r))
 
     def _compare_unordered_sequences(self, seq1, seq2, **kw):
@@ -1174,7 +1156,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_fromclause_ordered_set(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for l, r in util.zip_longest(left, right, fillvalue=None):
+        for l, r in zip_longest(left, right, fillvalue=None):
             self.stack.append((l, r))
 
     def visit_string(
@@ -1256,7 +1238,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     def visit_prefix_sequence(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for (l_clause, l_str), (r_clause, r_str) in util.zip_longest(
+        for (l_clause, l_str), (r_clause, r_str) in zip_longest(
             left, right, fillvalue=(None, None)
         ):
             if l_str != r_str:
@@ -1271,7 +1253,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
         for (
             (l_target, l_onclause, l_from, l_flags),
             (r_target, r_onclause, r_from, r_flags),
-        ) in util.zip_longest(left, right, fillvalue=(None, None, None, None)):
+        ) in zip_longest(left, right, fillvalue=(None, None, None, None)):
             if l_flags != r_flags:
                 return COMPARE_FAILED
             self.stack.append((l_target, r_target))
@@ -1292,7 +1274,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
         right_keys = sorted(
             right, key=lambda elem: (elem[0].fullname, elem[1])
         )
-        for (ltable, ldialect), (rtable, rdialect) in util.zip_longest(
+        for (ltable, ldialect), (rtable, rdialect) in zip_longest(
             left_keys, right_keys, fillvalue=(None, None)
         ):
             if ldialect != rdialect:
@@ -1317,7 +1299,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     ):
         # sequence of tuple pairs
 
-        for (lk, lv), (rk, rv) in util.zip_longest(
+        for (lk, lv), (rk, rv) in zip_longest(
             left, right, fillvalue=(None, None)
         ):
             if not self._compare_dml_values_or_ce(lk, rk, **kw):
@@ -1349,7 +1331,7 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
                     return COMPARE_FAILED
         elif isinstance(right, collections_abc.Sequence):
             return COMPARE_FAILED
-        elif py37:
+        else:
             # dictionaries guaranteed to support insert ordering in
             # py37 so that we can compare the keys in order.  without
             # this, we can't compare SQL expression keys because we don't
@@ -1359,25 +1341,15 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
                     return COMPARE_FAILED
                 if not self._compare_dml_values_or_ce(lv, rv, **kw):
                     return COMPARE_FAILED
-        else:
-            for lk in left:
-                lv = left[lk]
-
-                if lk not in right:
-                    return COMPARE_FAILED
-                rv = right[lk]
-
-                if not self._compare_dml_values_or_ce(lv, rv, **kw):
-                    return COMPARE_FAILED
 
     def visit_dml_multi_values(
         self, attrname, left_parent, left, right_parent, right, **kw
     ):
-        for lseq, rseq in util.zip_longest(left, right, fillvalue=None):
+        for lseq, rseq in zip_longest(left, right, fillvalue=None):
             if lseq is None or rseq is None:
                 return COMPARE_FAILED
 
-            for ld, rd in util.zip_longest(lseq, rseq, fillvalue=None):
+            for ld, rd in zip_longest(lseq, rseq, fillvalue=None):
                 if (
                     self.visit_dml_values(
                         attrname, left_parent, ld, right_parent, rd, **kw
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index 01763f266..82b8f538a 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -981,9 +981,7 @@ class ExternalType:
         return NO_CACHE
 
 
-class UserDefinedType(
-    util.with_metaclass(VisitableCheckKWArg, ExternalType, TypeEngine)
-):
+class UserDefinedType(ExternalType, TypeEngine, metaclass=VisitableCheckKWArg):
     """Base for user defined types.
 
     This should be the base of new types.  Note that
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 6394c43a0..8aee6a97a 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -454,7 +454,7 @@ def bind_values(clause):
 
 
 def _quote_ddl_expr(element):
-    if isinstance(element, util.string_types):
+    if isinstance(element, str):
         element = element.replace("'", "''")
         return "'%s'" % element
     else:
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index deb92b081..b08080753 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -59,7 +59,7 @@ def _generate_compiler_dispatch(cls):
         cls._original_compiler_dispatch = cls._compiler_dispatch
         return
 
-    if not isinstance(visit_name, util.compat.string_types):
+    if not isinstance(visit_name, str):
         raise exc.InvalidRequestError(
             "__visit_name__ on class %s must be a string at the class level"
             % cls.__name__
@@ -114,7 +114,7 @@ class TraversibleType(type):
         super(TraversibleType, cls).__init__(clsname, bases, clsdict)
 
 
-class Traversible(util.with_metaclass(TraversibleType)):
+class Traversible(metaclass=TraversibleType):
     """Base class for visitable objects, applies the
     :class:`.visitors.TraversibleType` metaclass.
 
@@ -200,7 +200,7 @@ def _generate_dispatcher(visitor, internal_dispatch, method_name):
     return langhelpers._exec_code_in_env(meth_text, {}, method_name)
 
 
-class InternalTraversal(util.with_metaclass(_InternalTraversalType, object)):
+class InternalTraversal(metaclass=_InternalTraversalType):
     r"""Defines visitor symbols used for internal traversal.
 
     The :class:`.InternalTraversal` class is used in two ways.  One is that
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index d78e24181..cf7494f57 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -5,9 +5,9 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
+from unittest import mock
 
 from . import config
-from . import mock
 from .assertions import assert_raises
 from .assertions import assert_raises_context_ok
 from .assertions import assert_raises_message
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index ed634befe..234ab4b93 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -5,9 +5,8 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
-from __future__ import absolute_import
-
 import contextlib
+from itertools import filterfalse
 import re
 import sys
 import warnings
@@ -26,7 +25,6 @@ from .. import util
 from ..engine import default
 from ..engine import url
 from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
-from ..util import compat
 from ..util import decorator
 
 
@@ -58,7 +56,7 @@ def expect_warnings_on(db, *messages, **kw):
     """
     spec = db_spec(db)
 
-    if isinstance(db, util.string_types) and not spec(config._current):
+    if isinstance(db, str) and not spec(config._current):
         yield
     else:
         with expect_warnings(*messages, **kw):
@@ -144,7 +142,6 @@ def _expect_warnings(
     messages,
     regex=True,
     assert_=True,
-    py2konly=False,
     raise_on_any_unexpected=False,
 ):
 
@@ -210,7 +207,7 @@ def _expect_warnings(
             finally:
                 _SEEN = _FILTERS = _EXC_CLS = None
 
-                if assert_ and (not py2konly or not compat.py3k):
+                if assert_:
                     assert not seen, "Warnings were not seen: %s" % ", ".join(
                         "%r" % (s.pattern if regex else s) for s in seen
                     )
@@ -325,9 +322,6 @@ def _assert_proper_exception_context(exception):
 
     """
 
-    if not util.py3k:
-        return
-
     if (
         exception.__context__ is not exception.__cause__
         and not exception.__suppress_context__
@@ -386,14 +380,14 @@ def _expect_raises(except_cls, msg=None, check_context=False):
         if msg is not None:
             # I'm often pdbing here, and "err" above isn't
             # in scope, so assign the string explicitly
-            error_as_string = util.text_type(err)
+            error_as_string = str(err)
             assert re.search(msg, error_as_string, re.UNICODE), "%r !~ %s" % (
                 msg,
                 error_as_string,
             )
         if check_context and not are_we_already_in_a_traceback:
             _assert_proper_exception_context(err)
-        print(util.text_type(err).encode("utf-8"))
+        print(str(err).encode("utf-8"))
 
     # it's generally a good idea to not carry traceback objects outside
     # of the except: block, but in this case especially we seem to have
@@ -456,7 +450,7 @@ class AssertsCompiledSQL:
                 dialect.supports_default_metavalue = supports_default_metavalue
             elif dialect == "default_enhanced":
                 dialect = default.StrCompileDialect()
-            elif isinstance(dialect, util.string_types):
+            elif isinstance(dialect, str):
                 dialect = url.URL.create(dialect).get_dialect()()
 
         if default_schema_name:
@@ -553,21 +547,10 @@ class AssertsCompiledSQL:
         c = CheckCompilerAccess(clause).compile(dialect=dialect, **kw)
 
         param_str = repr(getattr(c, "params", {}))
-        if util.py3k:
-            param_str = param_str.encode("utf-8").decode("ascii", "ignore")
-            print(
-                ("\nSQL String:\n" + util.text_type(c) + param_str).encode(
-                    "utf-8"
-                )
-            )
-        else:
-            print(
-                "\nSQL String:\n"
-                + util.text_type(c).encode("utf-8")
-                + param_str
-            )
+        param_str = param_str.encode("utf-8").decode("ascii", "ignore")
+        print(("\nSQL String:\n" + str(c) + param_str).encode("utf-8"))
 
-        cc = re.sub(r"[\n\t]", "", util.text_type(c))
+        cc = re.sub(r"[\n\t]", "", str(c))
 
         eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect))
 
@@ -687,9 +670,7 @@ class AssertsExecutionResults:
         found = util.IdentitySet(result)
         expected = {immutabledict(e) for e in expected}
 
-        for wrong in util.itertools_filterfalse(
-            lambda o: isinstance(o, cls), found
-        ):
+        for wrong in filterfalse(lambda o: isinstance(o, cls), found):
             fail(
                 'Unexpected type "%s", expected "%s"'
                 % (type(wrong).__name__, cls.__name__)
diff --git a/lib/sqlalchemy/testing/assertsql.py b/lib/sqlalchemy/testing/assertsql.py
index 485a13f82..46fdf6b9b 100644
--- a/lib/sqlalchemy/testing/assertsql.py
+++ b/lib/sqlalchemy/testing/assertsql.py
@@ -10,7 +10,6 @@ import contextlib
 import re
 
 from .. import event
-from .. import util
 from ..engine import url
 from ..engine.default import DefaultDialect
 from ..schema import _DDLCompiles
@@ -123,7 +122,7 @@ class CompiledSQL(SQLMatchRule):
                 for_executemany=context.compiled.for_executemany,
                 schema_translate_map=map_,
             )
-        _received_statement = re.sub(r"[\n\t]", "", util.text_type(compiled))
+        _received_statement = re.sub(r"[\n\t]", "", str(compiled))
         parameters = execute_observed.parameters
 
         if not parameters:
diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 98de5df5c..e17c09be7 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -5,8 +5,6 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
-from __future__ import absolute_import
-
 import collections
 import re
 import warnings
diff --git a/lib/sqlalchemy/testing/entities.py b/lib/sqlalchemy/testing/entities.py
index 15b6388fb..a87d853c3 100644
--- a/lib/sqlalchemy/testing/entities.py
+++ b/lib/sqlalchemy/testing/entities.py
@@ -7,7 +7,6 @@
 
 import sqlalchemy as sa
 from .. import exc as sa_exc
-from ..util import compat
 
 _repr_stack = set()
 
@@ -87,9 +86,7 @@ class ComparableMixin:
                 except (AttributeError, sa_exc.UnboundExecutionError):
                     return False
 
-                if hasattr(value, "__iter__") and not isinstance(
-                    value, compat.string_types
-                ):
+                if hasattr(value, "__iter__") and not isinstance(value, str):
                     if hasattr(value, "__getitem__") and not hasattr(
                         value, "keys"
                     ):
diff --git a/lib/sqlalchemy/testing/exclusions.py b/lib/sqlalchemy/testing/exclusions.py
index 7b2343128..c20cd920c 100644
--- a/lib/sqlalchemy/testing/exclusions.py
+++ b/lib/sqlalchemy/testing/exclusions.py
@@ -201,7 +201,7 @@ class Predicate:
             )
         elif isinstance(predicate, tuple):
             return SpecPredicate(*predicate)
-        elif isinstance(predicate, util.string_types):
+        elif isinstance(predicate, str):
             tokens = re.match(
                 r"([\+\w]+)\s*(?:(>=|==|!=|<=|<|>)\s*([\d\.]+))?", predicate
             )
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py
index d5e8e376a..1d56d923e 100644
--- a/lib/sqlalchemy/testing/fixtures.py
+++ b/lib/sqlalchemy/testing/fixtures.py
@@ -429,7 +429,7 @@ class TablesTest(TestBase):
                     try:
                         conn.execute(table.delete())
                     except sa.exc.DBAPIError as ex:
-                        util.print_(
+                        print(
                             ("Error emptying table %s: %r" % (table, ex)),
                             file=sys.stderr,
                         )
@@ -483,7 +483,7 @@ class TablesTest(TestBase):
         for table, data in cls.fixtures().items():
             if len(data) < 2:
                 continue
-            if isinstance(table, util.string_types):
+            if isinstance(table, str):
                 table = cls.tables[table]
             headers[table] = data[0]
             rows[table] = data[1:]
@@ -654,7 +654,7 @@ class MappedTest(TablesTest, assertions.AssertsExecutionResults):
                 cls_registry[classname] = cls
                 type.__init__(cls, classname, bases, dict_)
 
-        class _Base(util.with_metaclass(FindFixture, object)):
+        class _Base(metaclass=FindFixture):
             pass
 
         class Basic(BasicEntity, _Base):
diff --git a/lib/sqlalchemy/testing/mock.py b/lib/sqlalchemy/testing/mock.py
deleted file mode 100644
index 8fe08a678..000000000
--- a/lib/sqlalchemy/testing/mock.py
+++ /dev/null
@@ -1,32 +0,0 @@
-# testing/mock.py
-# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: https://www.opensource.org/licenses/mit-license.php
-
-"""Import stub for mock library.
-"""
-from __future__ import absolute_import
-
-from ..util import py3k
-
-
-if py3k:
-    from unittest.mock import MagicMock
-    from unittest.mock import Mock
-    from unittest.mock import call
-    from unittest.mock import patch
-    from unittest.mock import ANY
-else:
-    try:
-        from mock import MagicMock  # noqa
-        from mock import Mock  # noqa
-        from mock import call  # noqa
-        from mock import patch  # noqa
-        from mock import ANY  # noqa
-    except ImportError:
-        raise ImportError(
-            "SQLAlchemy's test suite requires the "
-            "'mock' library as of 0.8.2."
-        )
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index b382e97f6..32ed2c315 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -13,35 +13,20 @@ created so that multiple test frameworks can be supported at once
 is pytest.
 
 """
-
-from __future__ import absolute_import
-
 import abc
+import configparser
 import logging
 import re
 import sys
 
+from sqlalchemy.testing import asyncio
+
 # flag which indicates we are in the SQLAlchemy testing suite,
 # and not that of Alembic or a third party dialect.
 bootstrapped_as_sqlalchemy = False
 
 log = logging.getLogger("sqlalchemy.testing.plugin_base")
 
-
-py3k = sys.version_info >= (3, 0)
-
-if py3k:
-    import configparser
-
-    ABC = abc.ABC
-else:
-    import ConfigParser as configparser
-    import collections as collections_abc  # noqa
-
-    class ABC:
-        __metaclass__ = abc.ABCMeta
-
-
 # late imports
 fixtures = None
 engines = None
@@ -393,8 +378,7 @@ def _init_symbols(options, file_config):
 
 @post
 def _set_disable_asyncio(opt, file_config):
-    if opt.disable_asyncio or not py3k:
-        from sqlalchemy.testing import asyncio
+    if opt.disable_asyncio:
 
         asyncio.ENABLE_ASYNCIO = False
 
@@ -756,7 +740,7 @@ def _setup_config(config_obj, ctx):
     config._current.push(config_obj, testing)
 
 
-class FixtureFunctions(ABC):
+class FixtureFunctions(abc.ABC):
     @abc.abstractmethod
     def skip_test_exception(self, *arg, **kw):
         raise NotImplementedError()
diff --git a/lib/sqlalchemy/testing/plugin/pytestplugin.py b/lib/sqlalchemy/testing/plugin/pytestplugin.py
index 36aaa5d2a..ba774b118 100644
--- a/lib/sqlalchemy/testing/plugin/pytestplugin.py
+++ b/lib/sqlalchemy/testing/plugin/pytestplugin.py
@@ -344,7 +344,6 @@ _current_class = None
 
 def pytest_runtest_setup(item):
     from sqlalchemy.testing import asyncio
-    from sqlalchemy.util import string_types
 
     if not isinstance(item, pytest.Function):
         return
@@ -382,7 +381,7 @@ def pytest_runtest_setup(item):
                             "__Original test failure__:\n"
                             + _current_report.longreprtext,
                         )
-                    elif e.args[-1] and isinstance(e.args[-1], string_types):
+                    elif e.args[-1] and isinstance(e.args[-1], str):
                         args = list(e.args)
                         args[-1] += (
                             "\n__Original test failure__:\n"
diff --git a/lib/sqlalchemy/testing/profiling.py b/lib/sqlalchemy/testing/profiling.py
index b6a6e75b1..2761d4987 100644
--- a/lib/sqlalchemy/testing/profiling.py
+++ b/lib/sqlalchemy/testing/profiling.py
@@ -279,7 +279,7 @@ def count_functions(variance=0.05):
     # ended = time.time()
     pr.disable()
 
-    # s = compat.StringIO()
+    # s = StringIO()
     stats = pstats.Stats(pr, stream=sys.stdout)
 
     # timespent = ended - began
diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py
index 15613957c..e51eb172e 100644
--- a/lib/sqlalchemy/testing/provision.py
+++ b/lib/sqlalchemy/testing/provision.py
@@ -9,7 +9,6 @@ from .. import inspect
 from ..engine import url as sa_url
 from ..sql import ddl
 from ..sql import schema
-from ..util import compat
 
 
 log = logging.getLogger(__name__)
@@ -34,7 +33,7 @@ class register:
         return decorate
 
     def __call__(self, cfg, *arg):
-        if isinstance(cfg, compat.string_types):
+        if isinstance(cfg, str):
             url = sa_url.make_url(cfg)
         elif isinstance(cfg, sa_url.URL):
             url = cfg
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 325c0a9bb..8cb72d163 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -16,7 +16,6 @@ to provide specific inclusion/exclusions.
 """
 
 import platform
-import sys
 
 from . import exclusions
 from . import only_on
@@ -1272,12 +1271,12 @@ class SuiteRequirements(Requirements):
     @property
     def threading_with_mock(self):
         """Mark tests that use threading and mock at the same time - stability
-        issues have been observed with coverage + python 3.3
+        issues have been observed with coverage
 
         """
         return exclusions.skip_if(
-            lambda config: util.py3k and config.options.has_coverage,
-            "Stability issues with coverage + py3k",
+            lambda config: config.options.has_coverage,
+            "Stability issues with coverage",
         )
 
     @property
@@ -1317,17 +1316,6 @@ class SuiteRequirements(Requirements):
         return exclusions.only_if(check_lib, "patch library needed")
 
     @property
-    def non_broken_pickle(self):
-        from sqlalchemy.util import pickle
-
-        return exclusions.only_if(
-            lambda: util.cpython
-            and pickle.__name__ == "cPickle"
-            or sys.version_info >= (3, 2),
-            "Needs cPickle+cPython or newer Python 3 pickle",
-        )
-
-    @property
     def predictable_gc(self):
         """target platform must remove all cycles unconditionally when
         gc.collect() is called, as well as clean out unreferenced subclasses.
@@ -1505,8 +1493,3 @@ class SuiteRequirements(Requirements):
         sequence. This should be false only for oracle.
         """
         return exclusions.open()
-
-    @property
-    def generic_classes(self):
-        "If X[Y] can be implemented with ``__class_getitem__``. py3.7+"
-        return exclusions.only_if(lambda: util.py37)
diff --git a/lib/sqlalchemy/testing/suite/test_results.py b/lib/sqlalchemy/testing/suite/test_results.py
index f470432d2..5ad68034b 100644
--- a/lib/sqlalchemy/testing/suite/test_results.py
+++ b/lib/sqlalchemy/testing/suite/test_results.py
@@ -14,7 +14,6 @@ from ... import sql
 from ... import String
 from ... import testing
 from ... import text
-from ... import util
 
 
 class RowFetchTest(fixtures.TablesTest):
@@ -306,7 +305,7 @@ class ServerSideCursorsTest(
     ):
         engine = self._fixture(engine_ss_arg)
         with engine.begin() as conn:
-            if isinstance(statement, util.string_types):
+            if isinstance(statement, str):
                 result = conn.exec_driver_sql(statement)
             else:
                 result = conn.execute(statement)
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index b5a3dca3a..c1228f5df 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -1,3 +1,4 @@
+import collections.abc as collections_abc
 import itertools
 
 from .. import AssertsCompiledSQL
@@ -32,11 +33,9 @@ from ... import true
 from ... import tuple_
 from ... import TupleType
 from ... import union
-from ... import util
 from ... import values
 from ...exc import DatabaseError
 from ...exc import ProgrammingError
-from ...util import collections_abc
 
 
 class CollateTest(fixtures.TablesTest):
@@ -131,7 +130,7 @@ class OrderByLabelTest(fixtures.TablesTest):
         ly = (func.lower(table.c.q) + table.c.p).label("ly")
         self._assert_result(
             select(lx, ly).order_by(lx, ly.desc()),
-            [(3, util.u("q1p3")), (5, util.u("q2p2")), (7, util.u("q3p1"))],
+            [(3, "q1p3"), (5, "q2p2"), (7, "q3p1")],
         )
 
     def test_plain_desc(self):
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 4a5396ed8..7989b1e39 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -39,10 +39,8 @@ from ... import type_coerce
 from ... import TypeDecorator
 from ... import Unicode
 from ... import UnicodeText
-from ... import util
 from ...orm import declarative_base
 from ...orm import Session
-from ...util import u
 
 
 class _LiteralRoundTripFixture:
@@ -93,7 +91,7 @@ class _LiteralRoundTripFixture:
 class _UnicodeFixture(_LiteralRoundTripFixture, fixtures.TestBase):
     __requires__ = ("unicode_data",)
 
-    data = u(
+    data = (
         "Alors vous imaginez ma 🐍 surprise, au lever du jour, "
         "quand une drôle de petite 🐍 voix m’a réveillé. Elle "
         "disait: « S’il vous plaît… dessine-moi 🐍 un mouton! »"
@@ -124,7 +122,7 @@ class _UnicodeFixture(_LiteralRoundTripFixture, fixtures.TestBase):
         row = connection.execute(select(unicode_table.c.unicode_data)).first()
 
         eq_(row, (self.data,))
-        assert isinstance(row[0], util.text_type)
+        assert isinstance(row[0], str)
 
     def test_round_trip_executemany(self, connection):
         unicode_table = self.tables.unicode_table
@@ -139,7 +137,7 @@ class _UnicodeFixture(_LiteralRoundTripFixture, fixtures.TestBase):
         ).fetchall()
         eq_(rows, [(self.data,) for i in range(1, 4)])
         for row in rows:
-            assert isinstance(row[0], util.text_type)
+            assert isinstance(row[0], str)
 
     def _test_null_strings(self, connection):
         unicode_table = self.tables.unicode_table
@@ -154,18 +152,16 @@ class _UnicodeFixture(_LiteralRoundTripFixture, fixtures.TestBase):
         unicode_table = self.tables.unicode_table
 
         connection.execute(
-            unicode_table.insert(), {"id": 1, "unicode_data": u("")}
+            unicode_table.insert(), {"id": 1, "unicode_data": ""}
         )
         row = connection.execute(select(unicode_table.c.unicode_data)).first()
-        eq_(row, (u(""),))
+        eq_(row, ("",))
 
     def test_literal(self, literal_round_trip):
         literal_round_trip(self.datatype, [self.data], [self.data])
 
     def test_literal_non_ascii(self, literal_round_trip):
-        literal_round_trip(
-            self.datatype, [util.u("réve🐍 illé")], [util.u("réve🐍 illé")]
-        )
+        literal_round_trip(self.datatype, ["réve🐍 illé"], ["réve🐍 illé"])
 
 
 class UnicodeVarcharTest(_UnicodeFixture, fixtures.TablesTest):
@@ -243,9 +239,7 @@ class TextTest(_LiteralRoundTripFixture, fixtures.TablesTest):
         literal_round_trip(Text, ["some text"], ["some text"])
 
     def test_literal_non_ascii(self, literal_round_trip):
-        literal_round_trip(
-            Text, [util.u("réve🐍 illé")], [util.u("réve🐍 illé")]
-        )
+        literal_round_trip(Text, ["réve🐍 illé"], ["réve🐍 illé"])
 
     def test_literal_quoting(self, literal_round_trip):
         data = """some 'text' hey "hi there" that's text"""
@@ -277,9 +271,7 @@ class StringTest(_LiteralRoundTripFixture, fixtures.TestBase):
         literal_round_trip(String(40), ["some text"], ["some text"])
 
     def test_literal_non_ascii(self, literal_round_trip):
-        literal_round_trip(
-            String(40), [util.u("réve🐍 illé")], [util.u("réve🐍 illé")]
-        )
+        literal_round_trip(String(40), ["réve🐍 illé"], ["réve🐍 illé"])
 
     def test_literal_quoting(self, literal_round_trip):
         data = """some 'text' hey "hi there" that's text"""
@@ -474,10 +466,7 @@ class IntegerTest(_LiteralRoundTripFixture, fixtures.TestBase):
 
             eq_(row, (data,))
 
-            if util.py3k:
-                assert isinstance(row[0], int)
-            else:
-                assert isinstance(row[0], (long, int))  # noqa
+            assert isinstance(row[0], int)
 
         return run
 
@@ -880,10 +869,10 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
             ("boolean", None),
             ("string", "some string"),
             ("string", None),
-            ("string", util.u("réve illé")),
+            ("string", "réve illé"),
             (
                 "string",
-                util.u("réve🐍 illé"),
+                "réve🐍 illé",
                 testing.requires.json_index_supplementary_unicode_element,
             ),
             ("integer", 15),
@@ -1080,8 +1069,8 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
         (-1.0,),
         (15.052,),
         ("a string",),
-        (util.u("réve illé"),),
-        (util.u("réve🐍 illé"),),
+        ("réve illé",),
+        ("réve🐍 illé",),
     )
     def test_single_element_round_trip(self, element):
         data_table = self.tables.data_table
@@ -1243,8 +1232,8 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
                 {
                     "name": "r1",
                     "data": {
-                        util.u("réve🐍 illé"): util.u("réve🐍 illé"),
-                        "data": {"k1": util.u("drôl🐍e")},
+                        "réve🐍 illé": "réve🐍 illé",
+                        "data": {"k1": "drôl🐍e"},
                     },
                 },
             )
@@ -1252,8 +1241,8 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
             eq_(
                 conn.scalar(select(self.tables.data_table.c.data)),
                 {
-                    util.u("réve🐍 illé"): util.u("réve🐍 illé"),
-                    "data": {"k1": util.u("drôl🐍e")},
+                    "réve🐍 illé": "réve🐍 illé",
+                    "data": {"k1": "drôl🐍e"},
                 },
             )
 
diff --git a/lib/sqlalchemy/testing/suite/test_unicode_ddl.py b/lib/sqlalchemy/testing/suite/test_unicode_ddl.py
index 1334eb8db..a0cd6734a 100644
--- a/lib/sqlalchemy/testing/suite/test_unicode_ddl.py
+++ b/lib/sqlalchemy/testing/suite/test_unicode_ddl.py
@@ -10,8 +10,6 @@ from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
-from sqlalchemy.util import u
-from sqlalchemy.util import ue
 
 
 class UnicodeSchemaTest(fixtures.TablesTest):
@@ -23,20 +21,20 @@ class UnicodeSchemaTest(fixtures.TablesTest):
         global t1, t2, t3
 
         t1 = Table(
-            u("unitable1"),
+            "unitable1",
             metadata,
-            Column(u("méil"), Integer, primary_key=True),
-            Column(ue("\u6e2c\u8a66"), Integer),
+            Column("méil", Integer, primary_key=True),
+            Column("\u6e2c\u8a66", Integer),
             test_needs_fk=True,
         )
         t2 = Table(
-            u("Unitéble2"),
+            "Unitéble2",
             metadata,
-            Column(u("méil"), Integer, primary_key=True, key="a"),
+            Column("méil", Integer, primary_key=True, key="a"),
             Column(
-                ue("\u6e2c\u8a66"),
+                "\u6e2c\u8a66",
                 Integer,
-                ForeignKey(u("unitable1.méil")),
+                ForeignKey("unitable1.méil"),
                 key="b",
             ),
             test_needs_fk=True,
@@ -45,55 +43,53 @@ class UnicodeSchemaTest(fixtures.TablesTest):
         # Few DBs support Unicode foreign keys
         if testing.against("sqlite"):
             t3 = Table(
-                ue("\u6e2c\u8a66"),
+                "\u6e2c\u8a66",
                 metadata,
                 Column(
-                    ue("\u6e2c\u8a66_id"),
+                    "\u6e2c\u8a66_id",
                     Integer,
                     primary_key=True,
                     autoincrement=False,
                 ),
                 Column(
-                    ue("unitable1_\u6e2c\u8a66"),
+                    "unitable1_\u6e2c\u8a66",
                     Integer,
-                    ForeignKey(ue("unitable1.\u6e2c\u8a66")),
+                    ForeignKey("unitable1.\u6e2c\u8a66"),
                 ),
+                Column("Unitéble2_b", Integer, ForeignKey("Unitéble2.b")),
                 Column(
-                    u("Unitéble2_b"), Integer, ForeignKey(u("Unitéble2.b"))
-                ),
-                Column(
-                    ue("\u6e2c\u8a66_self"),
+                    "\u6e2c\u8a66_self",
                     Integer,
-                    ForeignKey(ue("\u6e2c\u8a66.\u6e2c\u8a66_id")),
+                    ForeignKey("\u6e2c\u8a66.\u6e2c\u8a66_id"),
                 ),
                 test_needs_fk=True,
             )
         else:
             t3 = Table(
-                ue("\u6e2c\u8a66"),
+                "\u6e2c\u8a66",
                 metadata,
                 Column(
-                    ue("\u6e2c\u8a66_id"),
+                    "\u6e2c\u8a66_id",
                     Integer,
                     primary_key=True,
                     autoincrement=False,
                 ),
-                Column(ue("unitable1_\u6e2c\u8a66"), Integer),
-                Column(u("Unitéble2_b"), Integer),
-                Column(ue("\u6e2c\u8a66_self"), Integer),
+                Column("unitable1_\u6e2c\u8a66", Integer),
+                Column("Unitéble2_b", Integer),
+                Column("\u6e2c\u8a66_self", Integer),
                 test_needs_fk=True,
             )
 
     def test_insert(self, connection):
-        connection.execute(t1.insert(), {u("méil"): 1, ue("\u6e2c\u8a66"): 5})
-        connection.execute(t2.insert(), {u("a"): 1, u("b"): 1})
+        connection.execute(t1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(t2.insert(), {"a": 1, "b": 1})
         connection.execute(
             t3.insert(),
             {
-                ue("\u6e2c\u8a66_id"): 1,
-                ue("unitable1_\u6e2c\u8a66"): 5,
-                u("Unitéble2_b"): 1,
-                ue("\u6e2c\u8a66_self"): 1,
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
             },
         )
 
@@ -102,42 +98,42 @@ class UnicodeSchemaTest(fixtures.TablesTest):
         eq_(connection.execute(t3.select()).fetchall(), [(1, 5, 1, 1)])
 
     def test_col_targeting(self, connection):
-        connection.execute(t1.insert(), {u("méil"): 1, ue("\u6e2c\u8a66"): 5})
-        connection.execute(t2.insert(), {u("a"): 1, u("b"): 1})
+        connection.execute(t1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(t2.insert(), {"a": 1, "b": 1})
         connection.execute(
             t3.insert(),
             {
-                ue("\u6e2c\u8a66_id"): 1,
-                ue("unitable1_\u6e2c\u8a66"): 5,
-                u("Unitéble2_b"): 1,
-                ue("\u6e2c\u8a66_self"): 1,
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
             },
         )
 
         row = connection.execute(t1.select()).first()
-        eq_(row._mapping[t1.c[u("méil")]], 1)
-        eq_(row._mapping[t1.c[ue("\u6e2c\u8a66")]], 5)
+        eq_(row._mapping[t1.c["méil"]], 1)
+        eq_(row._mapping[t1.c["\u6e2c\u8a66"]], 5)
 
         row = connection.execute(t2.select()).first()
-        eq_(row._mapping[t2.c[u("a")]], 1)
-        eq_(row._mapping[t2.c[u("b")]], 1)
+        eq_(row._mapping[t2.c["a"]], 1)
+        eq_(row._mapping[t2.c["b"]], 1)
 
         row = connection.execute(t3.select()).first()
-        eq_(row._mapping[t3.c[ue("\u6e2c\u8a66_id")]], 1)
-        eq_(row._mapping[t3.c[ue("unitable1_\u6e2c\u8a66")]], 5)
-        eq_(row._mapping[t3.c[u("Unitéble2_b")]], 1)
-        eq_(row._mapping[t3.c[ue("\u6e2c\u8a66_self")]], 1)
+        eq_(row._mapping[t3.c["\u6e2c\u8a66_id"]], 1)
+        eq_(row._mapping[t3.c["unitable1_\u6e2c\u8a66"]], 5)
+        eq_(row._mapping[t3.c["Unitéble2_b"]], 1)
+        eq_(row._mapping[t3.c["\u6e2c\u8a66_self"]], 1)
 
     def test_reflect(self, connection):
-        connection.execute(t1.insert(), {u("méil"): 2, ue("\u6e2c\u8a66"): 7})
-        connection.execute(t2.insert(), {u("a"): 2, u("b"): 2})
+        connection.execute(t1.insert(), {"méil": 2, "\u6e2c\u8a66": 7})
+        connection.execute(t2.insert(), {"a": 2, "b": 2})
         connection.execute(
             t3.insert(),
             {
-                ue("\u6e2c\u8a66_id"): 2,
-                ue("unitable1_\u6e2c\u8a66"): 7,
-                u("Unitéble2_b"): 2,
-                ue("\u6e2c\u8a66_self"): 2,
+                "\u6e2c\u8a66_id": 2,
+                "unitable1_\u6e2c\u8a66": 7,
+                "Unitéble2_b": 2,
+                "\u6e2c\u8a66_self": 2,
             },
         )
 
@@ -146,42 +142,36 @@ class UnicodeSchemaTest(fixtures.TablesTest):
         tt2 = Table(t2.name, meta, autoload_with=connection)
         tt3 = Table(t3.name, meta, autoload_with=connection)
 
-        connection.execute(tt1.insert(), {u("méil"): 1, ue("\u6e2c\u8a66"): 5})
-        connection.execute(tt2.insert(), {u("méil"): 1, ue("\u6e2c\u8a66"): 1})
+        connection.execute(tt1.insert(), {"méil": 1, "\u6e2c\u8a66": 5})
+        connection.execute(tt2.insert(), {"méil": 1, "\u6e2c\u8a66": 1})
         connection.execute(
             tt3.insert(),
             {
-                ue("\u6e2c\u8a66_id"): 1,
-                ue("unitable1_\u6e2c\u8a66"): 5,
-                u("Unitéble2_b"): 1,
-                ue("\u6e2c\u8a66_self"): 1,
+                "\u6e2c\u8a66_id": 1,
+                "unitable1_\u6e2c\u8a66": 5,
+                "Unitéble2_b": 1,
+                "\u6e2c\u8a66_self": 1,
             },
         )
 
         eq_(
-            connection.execute(
-                tt1.select().order_by(desc(u("méil")))
-            ).fetchall(),
+            connection.execute(tt1.select().order_by(desc("méil"))).fetchall(),
             [(2, 7), (1, 5)],
         )
         eq_(
-            connection.execute(
-                tt2.select().order_by(desc(u("méil")))
-            ).fetchall(),
+            connection.execute(tt2.select().order_by(desc("méil"))).fetchall(),
             [(2, 2), (1, 1)],
         )
         eq_(
             connection.execute(
-                tt3.select().order_by(desc(ue("\u6e2c\u8a66_id")))
+                tt3.select().order_by(desc("\u6e2c\u8a66_id"))
             ).fetchall(),
             [(2, 7, 2, 2), (1, 5, 1, 1)],
         )
 
     def test_repr(self):
         meta = MetaData()
-        t = Table(
-            ue("\u6e2c\u8a66"), meta, Column(ue("\u6e2c\u8a66_id"), Integer)
-        )
+        t = Table("\u6e2c\u8a66", meta, Column("\u6e2c\u8a66_id", Integer))
         eq_(
             repr(t),
             (
diff --git a/lib/sqlalchemy/testing/warnings.py b/lib/sqlalchemy/testing/warnings.py
index 2c41337e4..c8e481a90 100644
--- a/lib/sqlalchemy/testing/warnings.py
+++ b/lib/sqlalchemy/testing/warnings.py
@@ -4,9 +4,6 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
-
-from __future__ import absolute_import
-
 import warnings
 
 from . import assertions
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index 8a18a584a..a82e040df 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -7,13 +7,11 @@
 
 
 from collections import defaultdict
-from contextlib import contextmanager
 from functools import partial
 from functools import update_wrapper
 
 from ._collections import coerce_generator_arg
 from ._collections import coerce_to_immutabledict
-from ._collections import collections_abc
 from ._collections import column_dict
 from ._collections import column_set
 from ._collections import EMPTY_DICT
@@ -47,15 +45,10 @@ from ._collections import WeakPopulateDict
 from ._collections import WeakSequence
 from ._preloaded import preload_module
 from ._preloaded import preloaded
-from .compat import ABC
 from .compat import arm
 from .compat import b
 from .compat import b64decode
 from .compat import b64encode
-from .compat import binary_type
-from .compat import binary_types
-from .compat import byte_buffer
-from .compat import callable
 from .compat import cmp
 from .compat import cpython
 from .compat import dataclass_fields
@@ -63,42 +56,18 @@ from .compat import decode_backslashreplace
 from .compat import dottedgetter
 from .compat import has_refcount_gc
 from .compat import inspect_getfullargspec
-from .compat import int_types
-from .compat import iterbytes
-from .compat import itertools_filter
-from .compat import itertools_filterfalse
 from .compat import local_dataclass_fields
 from .compat import namedtuple
 from .compat import next
-from .compat import nullcontext
 from .compat import osx
-from .compat import parse_qsl
-from .compat import perf_counter
-from .compat import pickle
-from .compat import print_
-from .compat import py37
 from .compat import py38
 from .compat import py39
-from .compat import py3k
 from .compat import pypy
-from .compat import quote_plus
 from .compat import raise_
 from .compat import raise_from_cause
-from .compat import reduce
 from .compat import reraise
-from .compat import string_types
-from .compat import StringIO
-from .compat import text_type
 from .compat import threading
-from .compat import timezone
-from .compat import TYPE_CHECKING
-from .compat import u
-from .compat import ue
-from .compat import unquote
-from .compat import unquote_plus
 from .compat import win32
-from .compat import with_metaclass
-from .compat import zip_longest
 from .concurrency import asyncio
 from .concurrency import await_fallback
 from .concurrency import await_only
diff --git a/lib/sqlalchemy/util/_collections.py b/lib/sqlalchemy/util/_collections.py
index 9efbeb7db..32e989fca 100644
--- a/lib/sqlalchemy/util/_collections.py
+++ b/lib/sqlalchemy/util/_collections.py
@@ -6,21 +6,14 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 """Collection classes and helpers."""
-
-from __future__ import absolute_import
-
+import collections.abc as collections_abc
+from itertools import filterfalse
 import operator
 import types
 import weakref
 
-from .compat import binary_types
-from .compat import collections_abc
-from .compat import itertools_filterfalse
-from .compat import py37
-from .compat import string_types
 from .compat import threading
 
-
 EMPTY_SET = frozenset()
 
 
@@ -245,107 +238,8 @@ def _ordered_dictionary_sort(d, key=None):
     d.update(items)
 
 
-if py37:
-    OrderedDict = dict
-    sort_dictionary = _ordered_dictionary_sort
-
-else:
-    # prevent sort_dictionary from being used against a plain dictionary
-    # for Python < 3.7
-
-    def sort_dictionary(d, key=None):
-        """Sort an OrderedDict in place."""
-
-        d._ordered_dictionary_sort(key=key)
-
-    class OrderedDict(dict):
-        """Dictionary that maintains insertion order.
-
-        Superseded by Python dict as of Python 3.7
-
-        """
-
-        __slots__ = ("_list",)
-
-        def _ordered_dictionary_sort(self, key=None):
-            _ordered_dictionary_sort(self, key=key)
-
-        def __reduce__(self):
-            return OrderedDict, (self.items(),)
-
-        def __init__(self, ____sequence=None, **kwargs):
-            self._list = []
-            if ____sequence is None:
-                if kwargs:
-                    self.update(**kwargs)
-            else:
-                self.update(____sequence, **kwargs)
-
-        def clear(self):
-            self._list = []
-            dict.clear(self)
-
-        def copy(self):
-            return self.__copy__()
-
-        def __copy__(self):
-            return OrderedDict(self)
-
-        def update(self, ____sequence=None, **kwargs):
-            if ____sequence is not None:
-                if hasattr(____sequence, "keys"):
-                    for key in ____sequence.keys():
-                        self.__setitem__(key, ____sequence[key])
-                else:
-                    for key, value in ____sequence:
-                        self[key] = value
-            if kwargs:
-                self.update(kwargs)
-
-        def setdefault(self, key, value):
-            if key not in self:
-                self.__setitem__(key, value)
-                return value
-            else:
-                return self.__getitem__(key)
-
-        def __iter__(self):
-            return iter(self._list)
-
-        def keys(self):
-            return list(self)
-
-        def values(self):
-            return [self[key] for key in self._list]
-
-        def items(self):
-            return [(key, self[key]) for key in self._list]
-
-        def __setitem__(self, key, obj):
-            if key not in self:
-                try:
-                    self._list.append(key)
-                except AttributeError:
-                    # work around Python pickle loads() with
-                    # dict subclass (seems to ignore __setstate__?)
-                    self._list = [key]
-            dict.__setitem__(self, key, obj)
-
-        def __delitem__(self, key):
-            dict.__delitem__(self, key)
-            self._list.remove(key)
-
-        def pop(self, key, *default):
-            present = key in self
-            value = dict.pop(self, key, *default)
-            if present:
-                self._list.remove(key)
-            return value
-
-        def popitem(self):
-            item = dict.popitem(self)
-            self._list.remove(item[0])
-            return item
+OrderedDict = dict
+sort_dictionary = _ordered_dictionary_sort
 
 
 class OrderedSet(set):
@@ -515,7 +409,7 @@ class IdentitySet:
 
         if len(self) > len(other):
             return False
-        for m in itertools_filterfalse(
+        for m in filterfalse(
             other._members.__contains__, iter(self._members.keys())
         ):
             return False
@@ -540,7 +434,7 @@ class IdentitySet:
         if len(self) < len(other):
             return False
 
-        for m in itertools_filterfalse(
+        for m in filterfalse(
             self._members.__contains__, iter(other._members.keys())
         ):
             return False
@@ -818,7 +712,7 @@ def to_list(x, default=None):
     if x is None:
         return default
     if not isinstance(x, collections_abc.Iterable) or isinstance(
-        x, string_types + binary_types
+        x, (str, bytes)
     ):
         return [x]
     elif isinstance(x, list):
diff --git a/lib/sqlalchemy/util/_compat_py3k.py b/lib/sqlalchemy/util/_compat_py3k.py
deleted file mode 100644
index cd9f3ebc3..000000000
--- a/lib/sqlalchemy/util/_compat_py3k.py
+++ /dev/null
@@ -1,67 +0,0 @@
-# util/_compat_py3k.py
-# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: https://www.opensource.org/licenses/mit-license.php
-
-from functools import wraps
-
-# vendored from py3.7
-
-
-class _AsyncGeneratorContextManager:
-    """Helper for @asynccontextmanager."""
-
-    def __init__(self, func, args, kwds):
-        self.gen = func(*args, **kwds)
-        self.func, self.args, self.kwds = func, args, kwds
-        doc = getattr(func, "__doc__", None)
-        if doc is None:
-            doc = type(self).__doc__
-        self.__doc__ = doc
-
-    async def __aenter__(self):
-        try:
-            return await self.gen.__anext__()
-        except StopAsyncIteration:
-            raise RuntimeError("generator didn't yield") from None
-
-    async def __aexit__(self, typ, value, traceback):
-        if typ is None:
-            try:
-                await self.gen.__anext__()
-            except StopAsyncIteration:
-                return
-            else:
-                raise RuntimeError("generator didn't stop")
-        else:
-            if value is None:
-                value = typ()
-            # See _GeneratorContextManager.__exit__ for comments on subtleties
-            # in this implementation
-            try:
-                await self.gen.athrow(typ, value, traceback)
-                raise RuntimeError("generator didn't stop after athrow()")
-            except StopAsyncIteration as exc:
-                return exc is not value
-            except RuntimeError as exc:
-                if exc is value:
-                    return False
-                if isinstance(value, (StopIteration, StopAsyncIteration)):
-                    if exc.__cause__ is value:
-                        return False
-                raise
-            except BaseException as exc:
-                if exc is not value:
-                    raise
-
-
-# using the vendored version in all cases at the moment to establish
-# full test coverage
-def asynccontextmanager(func):
-    @wraps(func)
-    def helper(*args, **kwds):
-        return _AsyncGeneratorContextManager(func, args, kwds)
-
-    return helper
diff --git a/lib/sqlalchemy/util/_concurrency_py3k.py b/lib/sqlalchemy/util/_concurrency_py3k.py
index 55fe87c6a..71ed01307 100644
--- a/lib/sqlalchemy/util/_concurrency_py3k.py
+++ b/lib/sqlalchemy/util/_concurrency_py3k.py
@@ -6,6 +6,7 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 import asyncio
+from contextvars import copy_context as _copy_context
 import sys
 from typing import Any
 from typing import Callable
@@ -13,22 +14,17 @@ from typing import Coroutine
 
 import greenlet
 
-from . import compat
 from .langhelpers import memoized_property
 from .. import exc
 
-if compat.py37:
-    try:
-        from contextvars import copy_context as _copy_context
+try:
 
-        # If greenlet.gr_context is present in current version of greenlet,
-        # it will be set with a copy of the current context on creation.
-        # Refs: https://github.com/python-greenlet/greenlet/pull/198
-        getattr(greenlet.greenlet, "gr_context")
-    except (ImportError, AttributeError):
-        _copy_context = None
-else:
-    _copy_context = None
+    # If greenlet.gr_context is present in current version of greenlet,
+    # it will be set with a copy of the current context on creation.
+    # Refs: https://github.com/python-greenlet/greenlet/pull/198
+    getattr(greenlet.greenlet, "gr_context")
+except (ImportError, AttributeError):
+    _copy_context = None  # noqa
 
 
 def is_exit_exception(e):
@@ -193,10 +189,7 @@ def get_event_loop():
     Python 3.10 deprecates get_event_loop() as a standalone.
 
     """
-    if compat.py37:
-        try:
-            return asyncio.get_running_loop()
-        except RuntimeError:
-            return asyncio.get_event_loop_policy().get_event_loop()
-    else:
-        return asyncio.get_event_loop()
+    try:
+        return asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.get_event_loop_policy().get_event_loop()
diff --git a/lib/sqlalchemy/util/_preloaded.py b/lib/sqlalchemy/util/_preloaded.py
index c8da9230a..ff9f5bdb3 100644
--- a/lib/sqlalchemy/util/_preloaded.py
+++ b/lib/sqlalchemy/util/_preloaded.py
@@ -12,8 +12,6 @@ runtime.
 
 import sys
 
-from . import compat
-
 
 class _ModuleRegistry:
     """Registry of modules to load in a package init file.
@@ -60,7 +58,7 @@ class _ModuleRegistry:
             if (
                 not path or module.startswith(path)
             ) and key not in self.__dict__:
-                compat.import_(module, globals(), locals())
+                __import__(module, globals(), locals())
                 self.__dict__[key] = sys.modules[module]
 
 
diff --git a/lib/sqlalchemy/util/compat.py b/lib/sqlalchemy/util/compat.py
index 75b2a9a8d..e1291de36 100644
--- a/lib/sqlalchemy/util/compat.py
+++ b/lib/sqlalchemy/util/compat.py
@@ -6,9 +6,9 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 """Handle Python version/platform incompatibilities."""
-
+import base64
 import collections
-import contextlib
+import dataclasses
 import inspect
 import operator
 import platform
@@ -16,19 +16,15 @@ import sys
 
 py39 = sys.version_info >= (3, 9)
 py38 = sys.version_info >= (3, 8)
-py37 = sys.version_info >= (3, 7)
-py3k = sys.version_info >= (3, 0)
 pypy = platform.python_implementation() == "PyPy"
-
-
 cpython = platform.python_implementation() == "CPython"
+
 win32 = sys.platform.startswith("win")
 osx = sys.platform.startswith("darwin")
 arm = "aarch" in platform.machine().lower()
 
 has_refcount_gc = bool(cpython)
 
-contextmanager = contextlib.contextmanager
 dottedgetter = operator.attrgetter
 namedtuple = collections.namedtuple
 next = next  # noqa
@@ -47,23 +43,6 @@ FullArgSpec = collections.namedtuple(
 )
 
 
-class nullcontext:
-    """Context manager that does no additional processing.
-
-    Vendored from Python 3.7.
-
-    """
-
-    def __init__(self, enter_result=None):
-        self.enter_result = enter_result
-
-    def __enter__(self):
-        return self.enter_result
-
-    def __exit__(self, *excinfo):
-        pass
-
-
 try:
     import threading
 except ImportError:
@@ -84,7 +63,7 @@ def inspect_getfullargspec(func):
 
     nargs = co.co_argcount
     names = co.co_varnames
-    nkwargs = co.co_kwonlyargcount if py3k else 0
+    nkwargs = co.co_kwonlyargcount
     args = list(names[:nargs])
     kwonlyargs = list(names[nargs : nargs + nkwargs])
 
@@ -103,8 +82,8 @@ def inspect_getfullargspec(func):
         varkw,
         func.__defaults__,
         kwonlyargs,
-        func.__kwdefaults__ if py3k else None,
-        func.__annotations__ if py3k else {},
+        func.__kwdefaults__,
+        func.__annotations__,
     )
 
 
@@ -122,364 +101,163 @@ def importlib_metadata_get(group):
         return ep.get(group, ())
 
 
-if py3k:
-    import base64
-    import builtins
-    import configparser
-    import itertools
-    import pickle
-
-    from functools import reduce
-    from io import BytesIO as byte_buffer
-    from io import StringIO
-    from itertools import zip_longest
-    from time import perf_counter
-    from urllib.parse import (
-        quote_plus,
-        unquote_plus,
-        parse_qsl,
-        quote,
-        unquote,
-    )
-
-    string_types = (str,)
-    binary_types = (bytes,)
-    binary_type = bytes
-    text_type = str
-    int_types = (int,)
-    iterbytes = iter
-    long_type = int
-
-    itertools_filterfalse = itertools.filterfalse
-    itertools_filter = filter
-    itertools_imap = map
-
-    exec_ = getattr(builtins, "exec")
-    import_ = getattr(builtins, "__import__")
-    print_ = getattr(builtins, "print")
-
-    def b(s):
-        return s.encode("latin-1")
+def b(s):
+    return s.encode("latin-1")
 
-    def b64decode(x):
-        return base64.b64decode(x.encode("ascii"))
 
-    def b64encode(x):
-        return base64.b64encode(x).decode("ascii")
+def b64decode(x):
+    return base64.b64decode(x.encode("ascii"))
 
-    def decode_backslashreplace(text, encoding):
-        return text.decode(encoding, errors="backslashreplace")
 
-    def cmp(a, b):
-        return (a > b) - (a < b)
+def b64encode(x):
+    return base64.b64encode(x).decode("ascii")
 
-    def raise_(
-        exception, with_traceback=None, replace_context=None, from_=False
-    ):
-        r"""implement "raise" with cause support.
 
-        :param exception: exception to raise
-        :param with_traceback: will call exception.with_traceback()
-        :param replace_context: an as-yet-unsupported feature.  This is
-         an exception object which we are "replacing", e.g., it's our
-         "cause" but we don't want it printed.    Basically just what
-         ``__suppress_context__`` does but we don't want to suppress
-         the enclosing context, if any.  So for now we make it the
-         cause.
-        :param from\_: the cause.  this actually sets the cause and doesn't
-         hope to hide it someday.
+def decode_backslashreplace(text, encoding):
+    return text.decode(encoding, errors="backslashreplace")
 
-        """
-        if with_traceback is not None:
-            exception = exception.with_traceback(with_traceback)
 
-        if from_ is not False:
-            exception.__cause__ = from_
-        elif replace_context is not None:
-            # no good solution here, we would like to have the exception
-            # have only the context of replace_context.__context__ so that the
-            # intermediary exception does not change, but we can't figure
-            # that out.
-            exception.__cause__ = replace_context
+def cmp(a, b):
+    return (a > b) - (a < b)
 
-        try:
-            raise exception
-        finally:
-            # credit to
-            # https://cosmicpercolator.com/2016/01/13/exception-leaks-in-python-2-and-3/
-            # as the __traceback__ object creates a cycle
-            del exception, replace_context, from_, with_traceback
 
-    def u(s):
-        return s
+def raise_(exception, with_traceback=None, replace_context=None, from_=False):
+    r"""implement "raise" with cause support.
 
-    def ue(s):
-        return s
+    :param exception: exception to raise
+    :param with_traceback: will call exception.with_traceback()
+    :param replace_context: an as-yet-unsupported feature.  This is
+        an exception object which we are "replacing", e.g., it's our
+        "cause" but we don't want it printed.    Basically just what
+        ``__suppress_context__`` does but we don't want to suppress
+        the enclosing context, if any.  So for now we make it the
+        cause.
+    :param from\_: the cause.  this actually sets the cause and doesn't
+        hope to hide it someday.
 
-    from typing import TYPE_CHECKING
-
-    # Unused. Kept for backwards compatibility.
-    callable = callable  # noqa
-
-    from abc import ABC
-
-    def _qualname(fn):
-        return fn.__qualname__
-
-
-else:
-    import base64
-    import ConfigParser as configparser  # noqa
-    import itertools
-
-    from StringIO import StringIO  # noqa
-    from cStringIO import StringIO as byte_buffer  # noqa
-    from itertools import izip_longest as zip_longest  # noqa
-    from time import clock as perf_counter  # noqa
-    from urllib import quote  # noqa
-    from urllib import quote_plus  # noqa
-    from urllib import unquote  # noqa
-    from urllib import unquote_plus  # noqa
-    from urlparse import parse_qsl  # noqa
-
-    from abc import ABCMeta
-
-    class ABC:
-        __metaclass__ = ABCMeta
+    """
+    if with_traceback is not None:
+        exception = exception.with_traceback(with_traceback)
+
+    if from_ is not False:
+        exception.__cause__ = from_
+    elif replace_context is not None:
+        # no good solution here, we would like to have the exception
+        # have only the context of replace_context.__context__ so that the
+        # intermediary exception does not change, but we can't figure
+        # that out.
+        exception.__cause__ = replace_context
 
     try:
-        import cPickle as pickle
-    except ImportError:
-        import pickle  # noqa
-
-    string_types = (basestring,)  # noqa
-    binary_types = (bytes,)
-    binary_type = str
-    text_type = unicode  # noqa
-    int_types = int, long  # noqa
-    long_type = long  # noqa
-
-    callable = callable  # noqa
-    cmp = cmp  # noqa
-    reduce = reduce  # noqa
-
-    b64encode = base64.b64encode
-    b64decode = base64.b64decode
-
-    itertools_filterfalse = itertools.ifilterfalse
-    itertools_filter = itertools.ifilter
-    itertools_imap = itertools.imap
-
-    def b(s):
-        return s
-
-    def exec_(func_text, globals_, lcl=None):
-        if lcl is None:
-            exec("exec func_text in globals_")
-        else:
-            exec("exec func_text in globals_, lcl")
-
-    def iterbytes(buf):
-        return (ord(byte) for byte in buf)
-
-    def import_(*args):
-        if len(args) == 4:
-            args = args[0:3] + ([str(arg) for arg in args[3]],)
-        return __import__(*args)
-
-    def print_(*args, **kwargs):
-        fp = kwargs.pop("file", sys.stdout)
-        if fp is None:
-            return
-        for arg in enumerate(args):
-            if not isinstance(arg, basestring):  # noqa
-                arg = str(arg)
-            fp.write(arg)
-
-    def u(s):
-        # this differs from what six does, which doesn't support non-ASCII
-        # strings - we only use u() with
-        # literal source strings, and all our source files with non-ascii
-        # in them (all are tests) are utf-8 encoded.
-        return unicode(s, "utf-8")  # noqa
-
-    def ue(s):
-        return unicode(s, "unicode_escape")  # noqa
-
-    def decode_backslashreplace(text, encoding):
-        try:
-            return text.decode(encoding)
-        except UnicodeDecodeError:
-            # regular "backslashreplace" for an incompatible encoding raises:
-            # "TypeError: don't know how to handle UnicodeDecodeError in
-            # error callback"
-            return repr(text)[1:-1].decode()
-
-    def safe_bytestring(text):
-        # py2k only
-        if not isinstance(text, string_types):
-            return unicode(text).encode(  # noqa: F821
-                "ascii", errors="backslashreplace"
-            )
-        elif isinstance(text, unicode):  # noqa: F821
-            return text.encode("ascii", errors="backslashreplace")
-        else:
-            return text
-
-    exec(
-        "def raise_(exception, with_traceback=None, replace_context=None, "
-        "from_=False):\n"
-        "    if with_traceback:\n"
-        "        raise type(exception), exception, with_traceback\n"
-        "    else:\n"
-        "        raise exception\n"
-    )
-
-    TYPE_CHECKING = False
-
-    def _qualname(meth):
-        """return __qualname__ equivalent for a method on a class"""
-
-        for cls in meth.im_class.__mro__:
-            if meth.__name__ in cls.__dict__:
-                break
-        else:
-            return meth.__name__
-
-        return "%s.%s" % (cls.__name__, meth.__name__)
-
-
-if py3k:
+        raise exception
+    finally:
+        # credit to
+        # https://cosmicpercolator.com/2016/01/13/exception-leaks-in-python-2-and-3/
+        # as the __traceback__ object creates a cycle
+        del exception, replace_context, from_, with_traceback
+
+
+def _formatannotation(annotation, base_module=None):
+    """vendored from python 3.7"""
+
+    if getattr(annotation, "__module__", None) == "typing":
+        return repr(annotation).replace("typing.", "")
+    if isinstance(annotation, type):
+        if annotation.__module__ in ("builtins", base_module):
+            return annotation.__qualname__
+        return annotation.__module__ + "." + annotation.__qualname__
+    return repr(annotation)
+
+
+def inspect_formatargspec(
+    args,
+    varargs=None,
+    varkw=None,
+    defaults=None,
+    kwonlyargs=(),
+    kwonlydefaults={},
+    annotations={},
+    formatarg=str,
+    formatvarargs=lambda name: "*" + name,
+    formatvarkw=lambda name: "**" + name,
+    formatvalue=lambda value: "=" + repr(value),
+    formatreturns=lambda text: " -> " + text,
+    formatannotation=_formatannotation,
+):
+    """Copy formatargspec from python 3.7 standard library.
+
+    Python 3 has deprecated formatargspec and requested that Signature
+    be used instead, however this requires a full reimplementation
+    of formatargspec() in terms of creating Parameter objects and such.
+    Instead of introducing all the object-creation overhead and having
+    to reinvent from scratch, just copy their compatibility routine.
+
+    Ultimately we would need to rewrite our "decorator" routine completely
+    which is not really worth it right now, until all Python 2.x support
+    is dropped.
 
-    def _formatannotation(annotation, base_module=None):
-        """vendored from python 3.7"""
-
-        if getattr(annotation, "__module__", None) == "typing":
-            return repr(annotation).replace("typing.", "")
-        if isinstance(annotation, type):
-            if annotation.__module__ in ("builtins", base_module):
-                return annotation.__qualname__
-            return annotation.__module__ + "." + annotation.__qualname__
-        return repr(annotation)
-
-    def inspect_formatargspec(
-        args,
-        varargs=None,
-        varkw=None,
-        defaults=None,
-        kwonlyargs=(),
-        kwonlydefaults={},
-        annotations={},
-        formatarg=str,
-        formatvarargs=lambda name: "*" + name,
-        formatvarkw=lambda name: "**" + name,
-        formatvalue=lambda value: "=" + repr(value),
-        formatreturns=lambda text: " -> " + text,
-        formatannotation=_formatannotation,
-    ):
-        """Copy formatargspec from python 3.7 standard library.
-
-        Python 3 has deprecated formatargspec and requested that Signature
-        be used instead, however this requires a full reimplementation
-        of formatargspec() in terms of creating Parameter objects and such.
-        Instead of introducing all the object-creation overhead and having
-        to reinvent from scratch, just copy their compatibility routine.
-
-        Ultimately we would need to rewrite our "decorator" routine completely
-        which is not really worth it right now, until all Python 2.x support
-        is dropped.
-
-        """
-
-        kwonlydefaults = kwonlydefaults or {}
-        annotations = annotations or {}
-
-        def formatargandannotation(arg):
-            result = formatarg(arg)
-            if arg in annotations:
-                result += ": " + formatannotation(annotations[arg])
-            return result
-
-        specs = []
-        if defaults:
-            firstdefault = len(args) - len(defaults)
-        for i, arg in enumerate(args):
-            spec = formatargandannotation(arg)
-            if defaults and i >= firstdefault:
-                spec = spec + formatvalue(defaults[i - firstdefault])
-            specs.append(spec)
+    """
 
-        if varargs is not None:
-            specs.append(formatvarargs(formatargandannotation(varargs)))
-        else:
-            if kwonlyargs:
-                specs.append("*")
+    kwonlydefaults = kwonlydefaults or {}
+    annotations = annotations or {}
 
-        if kwonlyargs:
-            for kwonlyarg in kwonlyargs:
-                spec = formatargandannotation(kwonlyarg)
-                if kwonlydefaults and kwonlyarg in kwonlydefaults:
-                    spec += formatvalue(kwonlydefaults[kwonlyarg])
-                specs.append(spec)
-
-        if varkw is not None:
-            specs.append(formatvarkw(formatargandannotation(varkw)))
-
-        result = "(" + ", ".join(specs) + ")"
-        if "return" in annotations:
-            result += formatreturns(formatannotation(annotations["return"]))
+    def formatargandannotation(arg):
+        result = formatarg(arg)
+        if arg in annotations:
+            result += ": " + formatannotation(annotations[arg])
         return result
 
+    specs = []
+    if defaults:
+        firstdefault = len(args) - len(defaults)
+    for i, arg in enumerate(args):
+        spec = formatargandannotation(arg)
+        if defaults and i >= firstdefault:
+            spec = spec + formatvalue(defaults[i - firstdefault])
+        specs.append(spec)
+
+    if varargs is not None:
+        specs.append(formatvarargs(formatargandannotation(varargs)))
+    else:
+        if kwonlyargs:
+            specs.append("*")
 
-else:
-    from inspect import formatargspec as _inspect_formatargspec
-
-    def inspect_formatargspec(*spec, **kw):
-        # convert for a potential FullArgSpec from compat.getfullargspec()
-        return _inspect_formatargspec(*spec[0:4], **kw)  # noqa
-
-
-# Fix deprecation of accessing ABCs straight from collections module
-# (which will stop working in 3.8).
-if py3k:
-    import collections.abc as collections_abc
-else:
-    import collections as collections_abc  # noqa
-
-
-if py37:
-    import dataclasses
+    if kwonlyargs:
+        for kwonlyarg in kwonlyargs:
+            spec = formatargandannotation(kwonlyarg)
+            if kwonlydefaults and kwonlyarg in kwonlydefaults:
+                spec += formatvalue(kwonlydefaults[kwonlyarg])
+            specs.append(spec)
 
-    def dataclass_fields(cls):
-        """Return a sequence of all dataclasses.Field objects associated
-        with a class."""
+    if varkw is not None:
+        specs.append(formatvarkw(formatargandannotation(varkw)))
 
-        if dataclasses.is_dataclass(cls):
-            return dataclasses.fields(cls)
-        else:
-            return []
+    result = "(" + ", ".join(specs) + ")"
+    if "return" in annotations:
+        result += formatreturns(formatannotation(annotations["return"]))
+    return result
 
-    def local_dataclass_fields(cls):
-        """Return a sequence of all dataclasses.Field objects associated with
-        a class, excluding those that originate from a superclass."""
 
-        if dataclasses.is_dataclass(cls):
-            super_fields = set()
-            for sup in cls.__bases__:
-                super_fields.update(dataclass_fields(sup))
-            return [
-                f for f in dataclasses.fields(cls) if f not in super_fields
-            ]
-        else:
-            return []
+def dataclass_fields(cls):
+    """Return a sequence of all dataclasses.Field objects associated
+    with a class."""
 
+    if dataclasses.is_dataclass(cls):
+        return dataclasses.fields(cls)
+    else:
+        return []
 
-else:
 
-    def dataclass_fields(cls):
-        return []
+def local_dataclass_fields(cls):
+    """Return a sequence of all dataclasses.Field objects associated with
+    a class, excluding those that originate from a superclass."""
 
-    def local_dataclass_fields(cls):
+    if dataclasses.is_dataclass(cls):
+        super_fields = set()
+        for sup in cls.__bases__:
+            super_fields.update(dataclass_fields(sup))
+        return [f for f in dataclasses.fields(cls) if f not in super_fields]
+    else:
         return []
 
 
@@ -497,134 +275,3 @@ def reraise(tp, value, tb=None, cause=None):
     r"""legacy.  use raise\_()"""
 
     raise_(value, with_traceback=tb, from_=cause)
-
-
-def with_metaclass(meta, *bases, **kw):
-    """Create a base class with a metaclass.
-
-    Drops the middle class upon creation.
-
-    Source: https://lucumr.pocoo.org/2013/5/21/porting-to-python-3-redux/
-
-    """
-
-    class metaclass(meta):
-        __call__ = type.__call__
-        __init__ = type.__init__
-
-        def __new__(cls, name, this_bases, d):
-            if this_bases is None:
-                cls = type.__new__(cls, name, (), d)
-            else:
-                cls = meta(name, bases, d)
-
-            if hasattr(cls, "__init_subclass__") and hasattr(
-                cls.__init_subclass__, "__func__"
-            ):
-                cls.__init_subclass__.__func__(cls, **kw)
-            return cls
-
-    return metaclass("temporary_class", None, {})
-
-
-if py3k:
-    from datetime import timezone
-else:
-    from datetime import datetime
-    from datetime import timedelta
-    from datetime import tzinfo
-
-    class timezone(tzinfo):
-        """Minimal port of python 3 timezone object"""
-
-        __slots__ = "_offset"
-
-        def __init__(self, offset):
-            if not isinstance(offset, timedelta):
-                raise TypeError("offset must be a timedelta")
-            if not self._minoffset <= offset <= self._maxoffset:
-                raise ValueError(
-                    "offset must be a timedelta "
-                    "strictly between -timedelta(hours=24) and "
-                    "timedelta(hours=24)."
-                )
-            self._offset = offset
-
-        def __eq__(self, other):
-            if type(other) != timezone:
-                return False
-            return self._offset == other._offset
-
-        def __hash__(self):
-            return hash(self._offset)
-
-        def __repr__(self):
-            return "sqlalchemy.util.%s(%r)" % (
-                self.__class__.__name__,
-                self._offset,
-            )
-
-        def __str__(self):
-            return self.tzname(None)
-
-        def utcoffset(self, dt):
-            return self._offset
-
-        def tzname(self, dt):
-            return self._name_from_offset(self._offset)
-
-        def dst(self, dt):
-            return None
-
-        def fromutc(self, dt):
-            if isinstance(dt, datetime):
-                if dt.tzinfo is not self:
-                    raise ValueError("fromutc: dt.tzinfo " "is not self")
-                return dt + self._offset
-            raise TypeError(
-                "fromutc() argument must be a datetime instance" " or None"
-            )
-
-        @staticmethod
-        def _timedelta_to_microseconds(timedelta):
-            """backport of timedelta._to_microseconds()"""
-            return (
-                timedelta.days * (24 * 3600) + timedelta.seconds
-            ) * 1000000 + timedelta.microseconds
-
-        @staticmethod
-        def _divmod_timedeltas(a, b):
-            """backport of timedelta.__divmod__"""
-
-            q, r = divmod(
-                timezone._timedelta_to_microseconds(a),
-                timezone._timedelta_to_microseconds(b),
-            )
-            return q, timedelta(0, 0, r)
-
-        @staticmethod
-        def _name_from_offset(delta):
-            if not delta:
-                return "UTC"
-            if delta < timedelta(0):
-                sign = "-"
-                delta = -delta
-            else:
-                sign = "+"
-            hours, rest = timezone._divmod_timedeltas(
-                delta, timedelta(hours=1)
-            )
-            minutes, rest = timezone._divmod_timedeltas(
-                rest, timedelta(minutes=1)
-            )
-            result = "UTC%s%02d:%02d" % (sign, hours, minutes)
-            if rest.seconds:
-                result += ":%02d" % (rest.seconds,)
-            if rest.microseconds:
-                result += ".%06d" % (rest.microseconds,)
-            return result
-
-        _maxoffset = timedelta(hours=23, minutes=59)
-        _minoffset = -_maxoffset
-
-    timezone.utc = timezone(timedelta(0))
diff --git a/lib/sqlalchemy/util/concurrency.py b/lib/sqlalchemy/util/concurrency.py
index 37ecfdbc3..3518f874d 100644
--- a/lib/sqlalchemy/util/concurrency.py
+++ b/lib/sqlalchemy/util/concurrency.py
@@ -5,30 +5,25 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
-from . import compat
 
 have_greenlet = False
 
-if compat.py3k:
-    try:
-        import greenlet  # noqa F401
-    except ImportError:
-        pass
-    else:
-        have_greenlet = True
-        from ._concurrency_py3k import await_only
-        from ._concurrency_py3k import await_fallback
-        from ._concurrency_py3k import greenlet_spawn
-        from ._concurrency_py3k import is_exit_exception
-        from ._concurrency_py3k import AsyncAdaptedLock
-        from ._concurrency_py3k import _util_async_run  # noqa F401
-        from ._concurrency_py3k import (
-            _util_async_run_coroutine_function,
-        )  # noqa F401, E501
-        from ._concurrency_py3k import asyncio  # noqa F401
-
-    # does not need greennlet, just Python 3
-    from ._compat_py3k import asynccontextmanager  # noqa F401
+try:
+    import greenlet  # noqa F401
+except ImportError:
+    pass
+else:
+    have_greenlet = True
+    from ._concurrency_py3k import await_only
+    from ._concurrency_py3k import await_fallback
+    from ._concurrency_py3k import greenlet_spawn
+    from ._concurrency_py3k import is_exit_exception
+    from ._concurrency_py3k import AsyncAdaptedLock
+    from ._concurrency_py3k import _util_async_run  # noqa F401
+    from ._concurrency_py3k import (
+        _util_async_run_coroutine_function,
+    )  # noqa F401, E501
+    from ._concurrency_py3k import asyncio  # noqa F401
 
 if not have_greenlet:
 
@@ -40,12 +35,9 @@ if not have_greenlet:
         if have_greenlet:
             return None
 
-        if not compat.py3k:
-            raise ValueError("Cannot use this function in py2.")
-        else:
-            raise ValueError(
-                "the greenlet library is required to use this function."
-            )
+        raise ValueError(
+            "the greenlet library is required to use this function."
+        )
 
     def is_exit_exception(e):  # noqa F811
         return not isinstance(e, Exception)
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index ac2c6e6a4..621941b43 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -28,8 +28,7 @@ from .. import exc
 
 
 def md5_hex(x):
-    if compat.py3k:
-        x = x.encode("utf-8")
+    x = x.encode("utf-8")
     m = hashlib.md5()
     m.update(x)
     return m.hexdigest()
@@ -72,14 +71,6 @@ class safe_reraise:
                     with_traceback=exc_tb,
                 )
         else:
-            if not compat.py3k and self._exc_info and self._exc_info[1]:
-                # emulate Py3K's behavior of telling us when an exception
-                # occurs in an exception handler.
-                warn(
-                    "An exception has occurred during handling of a "
-                    "previous exception.  The previous exception "
-                    "is:\n %s %s\n" % (self._exc_info[0], self._exc_info[1])
-                )
             self._exc_info = None  # remove potential circular references
             compat.raise_(value, with_traceback=traceback)
 
@@ -99,7 +90,7 @@ def walk_subclasses(cls):
 
 
 def string_or_unprintable(element):
-    if isinstance(element, compat.string_types):
+    if isinstance(element, str):
         return element
     else:
         try:
@@ -148,7 +139,7 @@ def _unique_symbols(used, *bases):
     for base in bases:
         pool = itertools.chain(
             (base,),
-            compat.itertools_imap(lambda i: base + str(i), range(1000)),
+            map(lambda i: base + str(i), range(1000)),
         )
         for sym in pool:
             if sym not in used:
@@ -346,7 +337,7 @@ class PluginLoader:
 
     def register(self, name, modulepath, objname):
         def load():
-            mod = compat.import_(modulepath)
+            mod = __import__(modulepath)
             for token in modulepath.split(".")[1:]:
                 mod = getattr(mod, token)
             return getattr(mod, objname)
@@ -516,7 +507,7 @@ def format_argspec_plus(fn, grouped=True):
        'apply_pos': '(self, a, b, c, **d)'}
 
     """
-    if compat.callable(fn):
+    if callable(fn):
         spec = compat.inspect_getfullargspec(fn)
     else:
         spec = fn
@@ -899,14 +890,8 @@ def class_hierarchy(cls):
             process.append(b)
             hier.add(b)
 
-        if compat.py3k:
-            if c.__module__ == "builtins" or not hasattr(c, "__subclasses__"):
-                continue
-        else:
-            if c.__module__ == "__builtin__" or not hasattr(
-                c, "__subclasses__"
-            ):
-                continue
+        if c.__module__ == "builtins" or not hasattr(c, "__subclasses__"):
+            continue
 
         for s in [_ for _ in c.__subclasses__() if _ not in hier]:
             process.append(s)
@@ -985,7 +970,7 @@ def monkeypatch_proxied_specials(
         )
 
         env = from_instance is not None and {name: from_instance} or {}
-        compat.exec_(py, env)
+        exec(py, env)
         try:
             env[method].__defaults__ = fn.__defaults__
         except AttributeError:
@@ -1073,7 +1058,7 @@ def as_interface(obj, cls=None, methods=None, required=None):
     for method, impl in dictlike_iteritems(obj):
         if method not in interface:
             raise TypeError("%r: unknown in this interface" % method)
-        if not compat.callable(impl):
+        if not callable(impl):
             raise TypeError("%r=%r is not callable" % (method, impl))
         setattr(AnonymousInterface, method, staticmethod(impl))
         found.add(method)
@@ -1230,7 +1215,7 @@ class MemoizedSlots:
 
 # from paste.deploy.converters
 def asbool(obj):
-    if isinstance(obj, compat.string_types):
+    if isinstance(obj, str):
         obj = obj.strip().lower()
         if obj in ["true", "yes", "on", "y", "t", "1"]:
             return True
@@ -1375,14 +1360,8 @@ def assert_arg_type(arg, argtype, name):
 def dictlike_iteritems(dictlike):
     """Return a (key, value) iterator for almost any dict-like object."""
 
-    if compat.py3k:
-        if hasattr(dictlike, "items"):
-            return list(dictlike.items())
-    else:
-        if hasattr(dictlike, "iteritems"):
-            return dictlike.iteritems()
-        elif hasattr(dictlike, "items"):
-            return iter(dictlike.items())
+    if hasattr(dictlike, "items"):
+        return list(dictlike.items())
 
     getter = getattr(dictlike, "__getitem__", getattr(dictlike, "get", None))
     if getter is None:
@@ -1458,7 +1437,7 @@ class hybridmethod:
 class _symbol(int):
     def __new__(self, name, doc=None, canonical=None):
         """Construct a new named symbol."""
-        assert isinstance(name, compat.string_types)
+        assert isinstance(name, str)
         if canonical is None:
             canonical = hash(name)
         v = int.__new__(_symbol, canonical)
@@ -1585,7 +1564,7 @@ def ellipses_string(value, len_=25):
         return value
 
 
-class _hash_limit_string(compat.text_type):
+class _hash_limit_string(str):
     """A string subclass that can only be hashed on a maximum amount
     of unique values.