Create initial future package, RemovedIn20Warning

Reorganization of Select() is the first major element
of the 2.0 restructuring.   In order to start this we need
to first create the new Select constructor and apply legacy
elements to the old one.    This in turn necessitates
starting up the RemovedIn20Warning concept which itself
need to refer to "sqlalchemy.future", so begin to establish
this basic framework.   Additionally, update the
DML constructors with the newer no-keyword style.  Remove
the use of the "pending deprecation" and fix Query.add_column()
deprecation which was not acting as deprecated.

Fixes: #4845
Fixes: #4648
Change-Id: I0c7a22b2841a985e1c379a0bb6c94089aae6264c

12 files changed, 351 insertions, 100 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 609a60f7c..1c47902ab 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1715,7 +1715,7 @@ class MSSQLCompiler(compiler.SQLCompiler):
             select = select._generate()
             select._mssql_visit = True
             select = (
-                select.column(
+                select.add_columns(
                     sql.func.ROW_NUMBER()
                     .over(order_by=_order_by_clauses)
                     .label("mssql_rn")
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 87e0baa58..8c69bf097 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -986,7 +986,7 @@ class OracleCompiler(compiler.SQLCompiler):
 
                     for elem in for_update.of:
                         if not select.selected_columns.contains_column(elem):
-                            select = select.column(elem)
+                            select = select.add_columns(elem)
 
                 # Wrap the middle select and add the hint
                 inner_subquery = select.alias()
@@ -1056,7 +1056,7 @@ class OracleCompiler(compiler.SQLCompiler):
                     limitselect._for_update_arg = for_update
                     select = limitselect
                 else:
-                    limitselect = limitselect.column(
+                    limitselect = limitselect.add_columns(
                         sql.literal_column("ROWNUM").label("ora_rn")
                     )
                     limitselect._oracle_visit = True
@@ -1069,7 +1069,7 @@ class OracleCompiler(compiler.SQLCompiler):
                                 limitselect_cols.corresponding_column(elem)
                                 is None
                             ):
-                                limitselect = limitselect.column(elem)
+                                limitselect = limitselect.add_columns(elem)
 
                     limit_subquery = limitselect.alias()
                     origselect_cols = orig_select.selected_columns
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 378890444..d900a74b8 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -670,6 +670,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
 
     # a hook for SQLite's translation of
     # result column names
+    # NOTE: pyhive is using this hook, can't remove it :(
     _translate_colname = None
 
     _expanded_parameters = util.immutabledict()
diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index c8e71dda5..cc096ad03 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -587,11 +587,24 @@ class NotSupportedError(DatabaseError):
 
 
 class SADeprecationWarning(DeprecationWarning):
-    """Issued once per usage of a deprecated API."""
+    """Issued for usage of deprecated APIs."""
+
+
+class RemovedIn20Warning(SADeprecationWarning):
+    """Issued for usage of APIs specifically deprecated in SQLAlchemy 2.0.
+
+    .. seealso::
+
+        :ref:`error_b8d9`.
+
+    """
 
 
 class SAPendingDeprecationWarning(PendingDeprecationWarning):
-    """Issued once per usage of a deprecated API."""
+    """A similar warning as :class:`.SADeprecationWarning`, this warning
+    is not used in modern versions of SQLAlchemy.
+
+    """
 
 
 class SAWarning(RuntimeWarning):
diff --git a/lib/sqlalchemy/future/__init__.py b/lib/sqlalchemy/future/__init__.py
new file mode 100644
index 000000000..808ef076a
--- /dev/null
+++ b/lib/sqlalchemy/future/__init__.py
@@ -0,0 +1,15 @@
+# sql/future/__init__.py
+# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+"""Future 2.0 API features for Core.
+
+"""
+
+from ..sql.selectable import Select
+from ..util.langhelpers import public_factory
+
+select = public_factory(Select._create_select, ".expression.select")
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 15319e049..f19ec5673 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -642,8 +642,6 @@ class Query(Generative):
         """Return the full SELECT statement represented by this
         :class:`.Query`, converted to a scalar subquery.
 
-        Analogous to :meth:`sqlalchemy.sql.expression.SelectBase.as_scalar`.
-
         """
         return self.scalar_subquery()
 
@@ -1477,18 +1475,15 @@ class Query(Generative):
         for c in column:
             _ColumnEntity(self, c)
 
-    @util.pending_deprecation(
-        "0.7",
-        ":meth:`.add_column` is superseded " "by :meth:`.add_columns`",
-        False,
+    @util.deprecated(
+        "1.4",
+        ":meth:`.Query.add_column` is deprecated and will be removed in a "
+        "future release.  Please use :meth:`.Query.add_columns`",
     )
     def add_column(self, column):
         """Add a column expression to the list of result columns to be
         returned.
 
-        Pending deprecation: :meth:`.add_column` will be superseded by
-        :meth:`.add_columns`.
-
         """
         return self.add_columns(column)
 
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index dddbadd63..6bada51dd 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -46,19 +46,82 @@ class UpdateBase(
     _prefixes = ()
     named_with_column = False
 
+    @classmethod
+    def _constructor_20_deprecations(cls, fn_name, clsname, names):
+
+        param_to_method_lookup = dict(
+            whereclause=(
+                "The :paramref:`.%(func)s.whereclause` parameter "
+                "will be removed "
+                "in SQLAlchemy 2.0.  Please refer to the "
+                ":meth:`.%(classname)s.where` method."
+            ),
+            values=(
+                "The :paramref:`.%(func)s.values` parameter will be removed "
+                "in SQLAlchemy 2.0.  Please refer to the "
+                ":meth:`.%(classname)s.values` method."
+            ),
+            bind=(
+                "The :paramref:`.%(func)s.bind` parameter will be removed in "
+                "SQLAlchemy 2.0.  Please use explicit connection execution."
+            ),
+            inline=(
+                "The :paramref:`.%(func)s.inline` parameter will be "
+                "removed in "
+                "SQLAlchemy 2.0.  Please use the "
+                ":meth:`.%(classname)s.inline` method."
+            ),
+            prefixes=(
+                "The :paramref:`.%(func)s.prefixes parameter will be "
+                "removed in "
+                "SQLAlchemy 2.0.  Please use the "
+                ":meth:`.%(classname)s.prefix_with` "
+                "method."
+            ),
+            return_defaults=(
+                "The :paramref:`.%(func)s.return_defaults` parameter will be "
+                "removed in SQLAlchemy 2.0.  Please use the "
+                ":meth:`.%(classname)s.return_defaults` method."
+            ),
+            returning=(
+                "The :paramref:`.%(func)s.returning` parameter will be "
+                "removed in SQLAlchemy 2.0.  Please use the "
+                ":meth:`.%(classname)s.returning`` method."
+            ),
+            preserve_parameter_order=(
+                "The :paramref:`%(func)s.preserve_parameter_order` parameter "
+                "will be removed in SQLAlchemy 2.0.   Use the "
+                ":meth:`.%(classname)s.ordered_values` method with a list "
+                "of tuples. "
+            ),
+        )
+
+        return util.deprecated_params(
+            **{
+                name: (
+                    "2.0",
+                    param_to_method_lookup[name]
+                    % {"func": fn_name, "classname": clsname},
+                )
+                for name in names
+            }
+        )
+
     def _generate_fromclause_column_proxies(self, fromclause):
         fromclause._columns._populate_separate_keys(
             col._make_proxy(fromclause) for col in self._returning
         )
 
-    def _process_colparams(self, parameters):
+    def _process_colparams(self, parameters, preserve_parameter_order=False):
         def process_single(p):
             if isinstance(p, (list, tuple)):
                 return dict((c.key, pval) for c, pval in zip(self.table.c, p))
             else:
                 return p
 
-        if self._preserve_parameter_order and parameters is not None:
+        if (
+            preserve_parameter_order or self._preserve_parameter_order
+        ) and parameters is not None:
             if not isinstance(parameters, list) or (
                 parameters and not isinstance(parameters[0], tuple)
             ):
@@ -492,6 +555,18 @@ class Insert(ValuesBase):
 
     _supports_multi_parameters = True
 
+    @ValuesBase._constructor_20_deprecations(
+        "insert",
+        "Insert",
+        [
+            "values",
+            "inline",
+            "bind",
+            "prefixes",
+            "returning",
+            "return_defaults",
+        ],
+    )
     def __init__(
         self,
         table,
@@ -549,7 +624,7 @@ class Insert(ValuesBase):
             :ref:`inserts_and_updates` - SQL Expression Tutorial
 
         """
-        ValuesBase.__init__(self, table, values, prefixes)
+        super(Insert, self).__init__(table, values, prefixes)
         self._bind = bind
         self.select = self.select_names = None
         self.include_insert_from_select_defaults = False
@@ -565,6 +640,25 @@ class Insert(ValuesBase):
             return ()
 
     @_generative
+    def inline(self):
+        """Make this :class:`.Insert` construct "inline" .
+
+        When set, no attempt will be made to retrieve the
+        SQL-generated default values to be provided within the statement;
+        in particular,
+        this allows SQL expressions to be rendered 'inline' within the
+        statement without the need to pre-execute them beforehand; for
+        backends that support "returning", this turns off the "implicit
+        returning" feature for the statement.
+
+
+        .. versionchanged:: 1.4 the :paramref:`.Insert.inline` parameter
+           is now superseded by the :meth:`.Insert.inline` method.
+
+        """
+        self.inline = True
+
+    @_generative
     def from_select(self, names, select, include_defaults=True):
         """Return a new :class:`.Insert` construct which represents
         an ``INSERT...FROM SELECT`` statement.
@@ -636,6 +730,20 @@ class Update(ValuesBase):
 
     __visit_name__ = "update"
 
+    @ValuesBase._constructor_20_deprecations(
+        "update",
+        "Update",
+        [
+            "whereclause",
+            "values",
+            "inline",
+            "bind",
+            "prefixes",
+            "returning",
+            "return_defaults",
+            "preserve_parameter_order",
+        ],
+    )
     def __init__(
         self,
         table,
@@ -761,8 +869,9 @@ class Update(ValuesBase):
 
 
         """
+
         self._preserve_parameter_order = preserve_parameter_order
-        ValuesBase.__init__(self, table, values, prefixes)
+        super(Update, self).__init__(table, values, prefixes)
         self._bind = bind
         self._returning = returning
         if whereclause is not None:
@@ -782,6 +891,62 @@ class Update(ValuesBase):
             return ()
 
     @_generative
+    def ordered_values(self, *args):
+        """Specify the VALUES clause of this UPDATE statement with an explicit
+        parameter ordering that will be maintained in the SET clause of the
+        resulting UPDATE statement.
+
+        E.g.::
+
+            stmt = table.update().ordered_values(
+                ("name", "ed"), ("ident": "foo")
+            )
+
+        .. seealso::
+
+           :ref:`updates_order_parameters` - full example of the
+           :paramref:`~sqlalchemy.sql.expression.update.preserve_parameter_order`
+           flag
+
+        .. versionchanged:: 1.4 The :meth:`.Update.ordered_values` method
+           supersedes the :paramref:`.update.preserve_parameter_order`
+           parameter, which will be removed in SQLAlchemy 2.0.
+
+        """
+        if self.select is not None:
+            raise exc.InvalidRequestError(
+                "This construct already inserts from a SELECT"
+            )
+
+        if self.parameters is None:
+            (
+                self.parameters,
+                self._has_multi_parameters,
+            ) = self._process_colparams(
+                list(args), preserve_parameter_order=True
+            )
+        else:
+            raise exc.ArgumentError(
+                "This statement already has values present"
+            )
+
+    @_generative
+    def inline(self):
+        """Make this :class:`.Update` construct "inline" .
+
+        When set, SQL defaults present on :class:`.Column` objects via the
+        ``default`` keyword will be compiled 'inline' into the statement and
+        not pre-executed.  This means that their values will not be available
+        in the dictionary returned from
+        :meth:`.ResultProxy.last_updated_params`.
+
+        .. versionchanged:: 1.4 the :paramref:`.update.inline` parameter
+           is now superseded by the :meth:`.Update.inline` method.
+
+        """
+        self.inline = True
+
+    @_generative
     def where(self, whereclause):
         """return a new update() construct with the given expression added to
         its WHERE clause, joined to the existing clause via AND, if any.
@@ -821,6 +986,11 @@ class Delete(UpdateBase):
 
     __visit_name__ = "delete"
 
+    @ValuesBase._constructor_20_deprecations(
+        "delete",
+        "Delete",
+        ["whereclause", "values", "bind", "prefixes", "returning"],
+    )
     def __init__(
         self,
         table,
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 068fc6809..137b1605a 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -341,7 +341,7 @@ class FunctionElement(Executable, ColumnElement, FromClause):
             s = select([function_element])
 
         """
-        s = Select([self])
+        s = Select._create_select(self)
         if self._execution_options:
             s = s.execution_options(**self._execution_options)
         return s
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index db743f408..b2ec32c13 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -369,7 +369,8 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
             col = list(self.primary_key)[0]
         else:
             col = list(self.columns)[0]
-        return Select(
+        return Select._create_select_from_fromclause(
+            self,
             [functions.func.count(col).label("tbl_row_count")],
             whereclause,
             from_obj=[self],
@@ -1018,7 +1019,11 @@ class Join(FromClause):
         """
         collist = [self.left, self.right]
 
-        return Select(collist, whereclause, from_obj=[self], **kwargs)
+        if whereclause is not None:
+            kwargs["whereclause"] = whereclause
+        return Select._create_select_from_fromclause(
+            self, collist, **kwargs
+        ).select_from(self)
 
     @property
     def bind(self):
@@ -1142,7 +1147,7 @@ class Join(FromClause):
                 full=self.full,
             )
         else:
-            return self.select(use_labels=True, correlate=False).alias(name)
+            return self.select().apply_labels().correlate(None).alias(name)
 
     @property
     def _hide_froms(self):
@@ -1155,6 +1160,21 @@ class Join(FromClause):
         return [self] + self.left._from_objects + self.right._from_objects
 
 
+class NoInit(object):
+    def __init__(self, *arg, **kw):
+        raise NotImplementedError(
+            "The %s class is not intended to be constructed "
+            "directly.  Please use the %s() standalone "
+            "function or the %s() method available from appropriate "
+            "selectable objects."
+            % (
+                self.__class__.__name__,
+                self.__class__.__name__.lower(),
+                self.__class__.__name__.lower(),
+            )
+        )
+
+
 # FromClause ->
 #   AliasedReturnsRows
 #        -> Alias   only for FromClause
@@ -1163,7 +1183,7 @@ class Join(FromClause):
 #        -> Lateral -> FromClause, but we accept SelectBase
 #           w/ non-deprecated coercion
 #        -> TableSample -> only for FromClause
-class AliasedReturnsRows(FromClause):
+class AliasedReturnsRows(NoInit, FromClause):
     """Base class of aliases against tables, subqueries, and other
     selectables."""
 
@@ -1175,19 +1195,6 @@ class AliasedReturnsRows(FromClause):
         ("name", InternalTraversal.dp_anon_name),
     ]
 
-    def __init__(self, *arg, **kw):
-        raise NotImplementedError(
-            "The %s class is not intended to be constructed "
-            "directly.  Please use the %s() standalone "
-            "function or the %s() method available from appropriate "
-            "selectable objects."
-            % (
-                self.__class__.__name__,
-                self.__class__.__name__.lower(),
-                self.__class__.__name__.lower(),
-            )
-        )
-
     @classmethod
     def _construct(cls, *arg, **kw):
         obj = cls.__new__(cls)
@@ -3046,7 +3053,7 @@ class DeprecatedSelectGenerations(object):
         :class:`.Select` object.
 
         """
-        self.column.non_generative(self, column)
+        self.add_columns.non_generative(self, column)
 
     @util.deprecated(
         "1.4",
@@ -3171,6 +3178,38 @@ class Select(
         + SupportsCloneAnnotations._traverse_internals
     )
 
+    @classmethod
+    def _create_select(cls, *entities):
+        self = cls.__new__(cls)
+        self._raw_columns = [
+            coercions.expect(roles.ColumnsClauseRole, ent)
+            for ent in util.to_list(entities)
+        ]
+
+        # this should all go away once Select is converted to have
+        # default state at the class level
+        self._auto_correlate = True
+        self._from_obj = util.OrderedSet()
+        self._whereclause = None
+        self._having = None
+
+        GenerativeSelect.__init__(self)
+
+        return self
+
+    @classmethod
+    def _create_select_from_fromclause(cls, target, entities, *arg, **kw):
+        if arg or kw:
+            util.warn_deprecated_20(
+                "Passing arguments to %s.select() is deprecated and "
+                "will be removed in SQLAlchemy 2.0.  Please use generative "
+                "methods such as select().where(), etc."
+                % (target.__class__.__name__,)
+            )
+            return Select(entities, *arg, **kw)
+        else:
+            return Select._create_select(*entities)
+
     @util.deprecated_params(
         autocommit=(
             "0.6",
@@ -3201,7 +3240,7 @@ class Select(
         suffixes=None,
         **kwargs
     ):
-        """Construct a new :class:`.Select`.
+        """Construct a new :class:`.Select` using the 1.x style API.
 
         Similar functionality is also available via the
         :meth:`.FromClause.select` method on any :class:`.FromClause`.
@@ -3387,6 +3426,13 @@ class Select(
             :meth:`.Select.apply_labels`
 
         """
+        util.warn_deprecated_20(
+            "The select() function in SQLAlchemy 2.0 will accept a "
+            "series of columns / tables and other entities only, "
+            "passed positionally. For forwards compatibility, use the "
+            "sqlalchemy.future.select() construct."
+        )
+
         self._auto_correlate = correlate
         if distinct is not False:
             self._distinct = True
@@ -3418,8 +3464,6 @@ class Select(
             self._raw_columns = []
             for c in columns:
                 c = coercions.expect(roles.ColumnsClauseRole, c)
-                if isinstance(c, ScalarSelect):
-                    c = c.self_group(against=operators.comma_op)
                 self._raw_columns.append(c)
         else:
             self._raw_columns = []
@@ -3446,7 +3490,6 @@ class Select(
 
         GenerativeSelect.__init__(self, **kwargs)
 
-    # @_memoized_property
     @property
     def _froms(self):
         # current roadblock to caching is two tests that test that the
@@ -3741,13 +3784,13 @@ class Select(
         )
 
     @_generative
-    def column(self, column):
-        """return a new select() construct with the given column expression
+    def add_columns(self, *columns):
+        """return a new select() construct with the given column expressions
             added to its columns clause.
 
             E.g.::
 
-                my_select = my_select.column(table.c.new_column)
+                my_select = my_select.add_columns(table.c.new_column)
 
             See the documentation for :meth:`.Select.with_only_columns`
             for guidelines on adding /replacing the columns of a
@@ -3755,12 +3798,32 @@ class Select(
 
         """
         self._reset_memoizations()
-        column = coercions.expect(roles.ColumnsClauseRole, column)
 
-        if isinstance(column, ScalarSelect):
-            column = column.self_group(against=operators.comma_op)
+        self._raw_columns = self._raw_columns + [
+            coercions.expect(roles.ColumnsClauseRole, column)
+            for column in columns
+        ]
+
+    @util.deprecated(
+        "1.4",
+        "The :meth:`.Select.column` method is deprecated and will "
+        "be removed in a future release.  Please use "
+        ":meth:`.Select.add_columns",
+    )
+    def column(self, column):
+        """return a new select() construct with the given column expression
+            added to its columns clause.
+
+            E.g.::
+
+                my_select = my_select.column(table.c.new_column)
+
+            See the documentation for :meth:`.Select.with_only_columns`
+            for guidelines on adding /replacing the columns of a
+            :class:`.Select` object.
 
-        self._raw_columns = self._raw_columns + [column]
+        """
+        return self.add_columns(column)
 
     @util.dependencies("sqlalchemy.sql.util")
     def reduce_columns(self, sqlutil, only_synonyms=True):
@@ -4340,7 +4403,9 @@ class Exists(UnaryExpression):
         return element.self_group(against=operators.exists)
 
     def select(self, whereclause=None, **params):
-        return Select([self], whereclause, **params)
+        if whereclause is not None:
+            params["whereclause"] = whereclause
+        return Select._create_select_from_fromclause(self, [self], **params)
 
     def correlate(self, *fromclause):
         e = self._clone()
diff --git a/lib/sqlalchemy/testing/warnings.py b/lib/sqlalchemy/testing/warnings.py
index 889ae27e3..08f543b47 100644
--- a/lib/sqlalchemy/testing/warnings.py
+++ b/lib/sqlalchemy/testing/warnings.py
@@ -31,6 +31,8 @@ def setup_filters():
         "ignore", category=DeprecationWarning, message=".*inspect.get.*argspec"
     )
 
+    warnings.filterwarnings("ignore", category=sa_exc.RemovedIn20Warning)
+
 
 def assert_warnings(fn, warning_msgs, regex=False):
     """Assert that each of the given warnings are emitted by fn.
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index 30e384027..d2428bf75 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -88,12 +88,12 @@ from .compat import win32  # noqa
 from .compat import with_metaclass  # noqa
 from .compat import zip_longest  # noqa
 from .deprecations import deprecated  # noqa
+from .deprecations import deprecated_20  # noqa
 from .deprecations import deprecated_cls  # noqa
 from .deprecations import deprecated_params  # noqa
 from .deprecations import inject_docstring_text  # noqa
-from .deprecations import pending_deprecation  # noqa
 from .deprecations import warn_deprecated  # noqa
-from .deprecations import warn_pending_deprecation  # noqa
+from .deprecations import warn_deprecated_20  # noqa
 from .langhelpers import add_parameter_text  # noqa
 from .langhelpers import as_interface  # noqa
 from .langhelpers import asbool  # noqa
diff --git a/lib/sqlalchemy/util/deprecations.py b/lib/sqlalchemy/util/deprecations.py
index 058fe0c71..0db2c72ae 100644
--- a/lib/sqlalchemy/util/deprecations.py
+++ b/lib/sqlalchemy/util/deprecations.py
@@ -22,8 +22,10 @@ def warn_deprecated(msg, stacklevel=3):
     warnings.warn(msg, exc.SADeprecationWarning, stacklevel=stacklevel)
 
 
-def warn_pending_deprecation(msg, stacklevel=3):
-    warnings.warn(msg, exc.SAPendingDeprecationWarning, stacklevel=stacklevel)
+def warn_deprecated_20(msg, stacklevel=3):
+    msg += "(Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
+
+    warnings.warn(msg, exc.RemovedIn20Warning, stacklevel=stacklevel)
 
 
 def deprecated_cls(version, message, constructor="__init__"):
@@ -41,7 +43,9 @@ def deprecated_cls(version, message, constructor="__init__"):
     return decorate
 
 
-def deprecated(version, message=None, add_deprecation_to_docstring=True):
+def deprecated(
+    version, message=None, add_deprecation_to_docstring=True, warning=None
+):
     """Decorates a function and issues a deprecation warning on use.
 
     :param version:
@@ -66,17 +70,33 @@ def deprecated(version, message=None, add_deprecation_to_docstring=True):
     if message is None:
         message = "Call to deprecated function %(func)s"
 
+    if warning is None:
+        warning = exc.SADeprecationWarning
+
     def decorate(fn):
         return _decorate_with_warning(
-            fn,
-            exc.SADeprecationWarning,
-            message % dict(func=fn.__name__),
-            header,
+            fn, warning, message % dict(func=fn.__name__), header
         )
 
     return decorate
 
 
+def deprecated_20(api_name, alternative=None, **kw):
+    message = (
+        "The %s() function/method is considered legacy as of the "
+        "1.x series of SQLAlchemy and will be removed in 2.0." % api_name
+    )
+
+    if alternative:
+        message += " " + alternative
+
+    message += " (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)"
+
+    return deprecated(
+        "2.0", message=message, warning=exc.RemovedIn20Warning, **kw
+    )
+
+
 def deprecated_params(**specs):
     """Decorates a function to warn on use of certain parameters.
 
@@ -94,8 +114,14 @@ def deprecated_params(**specs):
     """
 
     messages = {}
+    version_warnings = {}
     for param, (version, message) in specs.items():
         messages[param] = _sanitize_restructured_text(message)
+        version_warnings[param] = (
+            exc.RemovedIn20Warning
+            if version == "2.0"
+            else exc.SADeprecationWarning
+        )
 
     def decorate(fn):
         spec = compat.inspect_getfullargspec(fn)
@@ -115,14 +141,16 @@ def deprecated_params(**specs):
         @decorator
         def warned(fn, *args, **kwargs):
             for m in check_defaults:
-                if kwargs[m] != defaults[m]:
+                if (defaults[m] is None and kwargs[m] is not None) or (
+                    defaults[m] is not None and kwargs[m] != defaults[m]
+                ):
                     warnings.warn(
-                        messages[m], exc.SADeprecationWarning, stacklevel=3
+                        messages[m], version_warnings[m], stacklevel=3
                     )
             for m in check_kw:
                 if m in kwargs:
                     warnings.warn(
-                        messages[m], exc.SADeprecationWarning, stacklevel=3
+                        messages[m], version_warnings[m], stacklevel=3
                     )
 
             return fn(*args, **kwargs)
@@ -143,44 +171,6 @@ def deprecated_params(**specs):
     return decorate
 
 
-def pending_deprecation(
-    version, message=None, add_deprecation_to_docstring=True
-):
-    """Decorates a function and issues a pending deprecation warning on use.
-
-    :param version:
-      An approximate future version at which point the pending deprecation
-      will become deprecated.  Not used in messaging.
-
-    :param message:
-      If provided, issue message in the warning.  A sensible default
-      is used if not provided.
-
-    :param add_deprecation_to_docstring:
-      Default True.  If False, the wrapped function's __doc__ is left
-      as-is.  If True, the 'message' is prepended to the docs if
-      provided, or sensible default if message is omitted.
-    """
-
-    if add_deprecation_to_docstring:
-        header = ".. deprecated:: %s (pending) %s" % (version, (message or ""))
-    else:
-        header = None
-
-    if message is None:
-        message = "Call to deprecated function %(func)s"
-
-    def decorate(fn):
-        return _decorate_with_warning(
-            fn,
-            exc.SAPendingDeprecationWarning,
-            message % dict(func=fn.__name__),
-            header,
-        )
-
-    return decorate
-
-
 def deprecated_option_value(parameter_value, default_value, warning_text):
     if parameter_value is None:
         return default_value