Convert remaining ORM APIs to support 2.0 style

This is kind of a mixed bag of all kinds to help get us
to 1.4 betas.    The documentation stuff is a work in
progress.    Lots of other relatively small changes to
APIs and things.    More commits will follow to continue
improving the documentation and transitioning to the
1.4/2.0 hybrid documentation.  In particular some refinements
to Session usage models so that it can match Engine's
scoping / transactional patterns, and a decision to
start moving away from "subtransactions" completely.

* add select().from_statement() to produce FromStatement in an
  ORM context

* begin referring to select() that has "plugins" for the few edge
  cases where select() will have ORM-only behaviors

* convert dynamic.AppenderQuery to its own object that can use
  select(), though at the moment it uses Query to support legacy
  join calling forms.

* custom query classes for AppenderQuery are replaced by
  do_orm_execute() hooks for custom actions, a separate gerrit
  will document this

* add Session.get() to replace query.get()

* Deprecate session.begin->subtransaction.  propose within the
  test suite a hypothetical recipe for apps that rely on this
  pattern

* introduce Session construction level context manager,
  sessionmaker context manager, rewrite the whole top of the
  session_transaction.rst documentation.   Establish context manager
  patterns for Session that are identical to engine

* ensure same begin_nested() / commit() behavior as engine

* devise all new "join into an external transaction" recipe,
  add test support for it, add rules into Session so it
  just works, write new docs.  need to ensure this doesn't
  break anything

* vastly reduce the verbosity of lots of session docs as
  I dont think people read this stuff and it's difficult
  to keep current in any case

* constructs like case(), with_only_columns() really need
  to move to *columns, add a coercion rule to just change
  these.

* docs need changes everywhere I look.  in_() is not in
  the Core tutorial?  how do people even know about it?
  Remove tons of cruft from Select docs, etc.

* build a system for common ORM options like populate_existing
  and autoflush to populate from execution options.

* others?

Change-Id: Ia4bea0f804250e54d90b3884cf8aab8b66b82ecf

18 files changed, 1230 insertions, 836 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 6bc9588ad..2d672099b 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -759,6 +759,40 @@ class Connection(Connectable):
 
         return self._transaction is not None and self._transaction.is_active
 
+    def in_nested_transaction(self):
+        """Return True if a transaction is in progress."""
+        if self.__branch_from is not None:
+            return self.__branch_from.in_nested_transaction()
+
+        return (
+            self._nested_transaction is not None
+            and self._nested_transaction.is_active
+        )
+
+    def get_transaction(self):
+        """Return the current root transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+
+        if self.__branch_from is not None:
+            return self.__branch_from.get_transaction()
+
+        return self._transaction
+
+    def get_nested_transaction(self):
+        """Return the current nested transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+        if self.__branch_from is not None:
+
+            return self.__branch_from.get_nested_transaction()
+
+        return self._nested_transaction
+
     def _begin_impl(self, transaction):
         assert not self.__branch_from
 
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 7df17cf22..db546380e 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -282,7 +282,8 @@ class Result(InPlaceGenerative):
        updated usage model and calling facade for SQLAlchemy Core and
        SQLAlchemy ORM.   In Core, it forms the basis of the
        :class:`.CursorResult` object which replaces the previous
-       :class:`.ResultProxy` interface.
+       :class:`.ResultProxy` interface.   When using the ORM, a higher level
+       object called :class:`.ChunkedIteratorResult` is normally used.
 
     """
 
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index fc6623609..e642a83d5 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -234,7 +234,7 @@ class BakedQuery(object):
         # used by the Connection, which in itself is more expensive to
         # generate than what BakedQuery was able to provide in 1.3 and prior
 
-        if statement.compile_options._bake_ok:
+        if statement._compile_options._bake_ok:
             self._bakery[self._effective_key(session)] = (
                 query,
                 statement,
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index fabb095a2..32ec60322 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -90,8 +90,13 @@ def create_session(bind=None, **kwargs):
     create_session().
 
     """
+
+    if kwargs.get("future", False):
+        kwargs.setdefault("autocommit", False)
+    else:
+        kwargs.setdefault("autocommit", True)
+
     kwargs.setdefault("autoflush", False)
-    kwargs.setdefault("autocommit", True)
     kwargs.setdefault("expire_on_commit", False)
     return Session(bind=bind, **kwargs)
 
@@ -267,12 +272,15 @@ contains_alias = public_factory(AliasOption, ".orm.contains_alias")
 
 def __go(lcls):
     global __all__
+    global AppenderQuery
     from .. import util as sa_util  # noqa
     from . import dynamic  # noqa
     from . import events  # noqa
     from . import loading  # noqa
     import inspect as _inspect
 
+    from .dynamic import AppenderQuery
+
     __all__ = sorted(
         name
         for name, obj in lcls.items()
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index d5f001db1..55a6b4cd2 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -188,7 +188,7 @@ class ORMCompileState(CompileState):
         raise NotImplementedError()
 
     @classmethod
-    def get_column_descriptions(self, statement):
+    def get_column_descriptions(cls, statement):
         return _column_descriptions(statement)
 
     @classmethod
@@ -204,8 +204,14 @@ class ORMCompileState(CompileState):
         if is_reentrant_invoke:
             return statement, execution_options
 
-        load_options = execution_options.get(
-            "_sa_orm_load_options", QueryContext.default_load_options
+        (
+            load_options,
+            execution_options,
+        ) = QueryContext.default_load_options.from_execution_options(
+            "_sa_orm_load_options",
+            {"populate_existing", "autoflush", "yield_per"},
+            execution_options,
+            statement._execution_options,
         )
 
         bind_arguments["clause"] = statement
@@ -246,6 +252,7 @@ class ORMCompileState(CompileState):
         load_options = execution_options.get(
             "_sa_orm_load_options", QueryContext.default_load_options
         )
+
         querycontext = QueryContext(
             compile_state,
             statement,
@@ -304,7 +311,7 @@ class ORMFromStatementCompileState(ORMCompileState):
         self._primary_entity = None
 
         self.use_legacy_query_style = (
-            statement_container.compile_options._use_legacy_query_style
+            statement_container._compile_options._use_legacy_query_style
         )
         self.statement_container = self.select_statement = statement_container
         self.requested_statement = statement = statement_container.element
@@ -315,9 +322,9 @@ class ORMFromStatementCompileState(ORMCompileState):
 
         _QueryEntity.to_compile_state(self, statement_container._raw_columns)
 
-        self.compile_options = statement_container.compile_options
+        self.compile_options = statement_container._compile_options
 
-        self.current_path = statement_container.compile_options._current_path
+        self.current_path = statement_container._compile_options._current_path
 
         if toplevel and statement_container._with_options:
             self.attributes = {"_unbound_load_dedupes": set()}
@@ -416,8 +423,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
         # if we are a select() that was never a legacy Query, we won't
         # have ORM level compile options.
-        statement.compile_options = cls.default_compile_options.safe_merge(
-            statement.compile_options
+        statement._compile_options = cls.default_compile_options.safe_merge(
+            statement._compile_options
         )
 
         self = cls.__new__(cls)
@@ -434,20 +441,20 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         # indicates this select() came from Query.statement
         self.for_statement = (
             for_statement
-        ) = select_statement.compile_options._for_statement
+        ) = select_statement._compile_options._for_statement
 
         if not for_statement and not toplevel:
             # for subqueries, turn off eagerloads.
             # if "for_statement" mode is set, Query.subquery()
             # would have set this flag to False already if that's what's
             # desired
-            select_statement.compile_options += {
+            select_statement._compile_options += {
                 "_enable_eagerloads": False,
             }
 
         # generally if we are from Query or directly from a select()
         self.use_legacy_query_style = (
-            select_statement.compile_options._use_legacy_query_style
+            select_statement._compile_options._use_legacy_query_style
         )
 
         self._entities = []
@@ -457,15 +464,15 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         self._no_yield_pers = set()
 
         # legacy: only for query.with_polymorphic()
-        if select_statement.compile_options._with_polymorphic_adapt_map:
+        if select_statement._compile_options._with_polymorphic_adapt_map:
             self._with_polymorphic_adapt_map = dict(
-                select_statement.compile_options._with_polymorphic_adapt_map
+                select_statement._compile_options._with_polymorphic_adapt_map
             )
             self._setup_with_polymorphics()
 
         _QueryEntity.to_compile_state(self, select_statement._raw_columns)
 
-        self.compile_options = select_statement.compile_options
+        self.compile_options = select_statement._compile_options
 
         # determine label style.   we can make different decisions here.
         # at the moment, trying to see if we can always use DISAMBIGUATE_ONLY
@@ -479,7 +486,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         else:
             self.label_style = self.select_statement._label_style
 
-        self.current_path = select_statement.compile_options._current_path
+        self.current_path = select_statement._compile_options._current_path
 
         self.eager_order_by = ()
 
@@ -668,7 +675,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         self._polymorphic_adapters = {}
 
         compile_options = cls.default_compile_options.safe_merge(
-            query.compile_options
+            query._compile_options
         )
         # legacy: only for query.with_polymorphic()
         if compile_options._with_polymorphic_adapt_map:
@@ -711,6 +718,26 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
                 for elem in _select_iterables([element]):
                     yield elem
 
+    @classmethod
+    @util.preload_module("sqlalchemy.orm.query")
+    def from_statement(cls, statement, from_statement):
+        query = util.preloaded.orm_query
+
+        from_statement = coercions.expect(
+            roles.SelectStatementRole,
+            from_statement,
+            apply_propagate_attrs=statement,
+        )
+
+        stmt = query.FromStatement(statement._raw_columns, from_statement)
+        stmt.__dict__.update(
+            _with_options=statement._with_options,
+            _with_context_options=statement._with_context_options,
+            _execution_options=statement._execution_options,
+            _propagate_attrs=statement._propagate_attrs,
+        )
+        return stmt
+
     def _setup_with_polymorphics(self):
         # legacy: only for query.with_polymorphic()
         for ext_info, wp in self._with_polymorphic_adapt_map.items():
diff --git a/lib/sqlalchemy/orm/dynamic.py b/lib/sqlalchemy/orm/dynamic.py
index d15127563..7832152a2 100644
--- a/lib/sqlalchemy/orm/dynamic.py
+++ b/lib/sqlalchemy/orm/dynamic.py
@@ -23,7 +23,12 @@ from . import util as orm_util
 from .query import Query
 from .. import exc
 from .. import log
+from .. import sql
 from .. import util
+from ..engine import result as _result
+from ..sql import selectable
+from ..sql.base import _generative
+from ..sql.base import Generative
 
 
 @log.class_logger
@@ -74,7 +79,6 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         dispatch,
         target_mapper,
         order_by,
-        query_class=None,
         **kw
     ):
         super(DynamicAttributeImpl, self).__init__(
@@ -82,12 +86,7 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         )
         self.target_mapper = target_mapper
         self.order_by = order_by
-        if not query_class:
-            self.query_class = AppenderQuery
-        elif AppenderMixin in query_class.mro():
-            self.query_class = query_class
-        else:
-            self.query_class = mixin_user_query(query_class)
+        self.query_class = AppenderQuery
 
     def get(self, state, dict_, passive=attributes.PASSIVE_OFF):
         if not passive & attributes.SQL_OK:
@@ -259,15 +258,26 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         self.remove(state, dict_, value, initiator, passive=passive)
 
 
-class AppenderMixin(object):
-    query_class = None
+class AppenderQuery(Generative):
+    """A dynamic query that supports basic collection storage operations."""
 
     def __init__(self, attr, state):
-        super(AppenderMixin, self).__init__(attr.target_mapper, None)
+
+        # this can be select() except for aliased=True flag on join()
+        # and corresponding behaviors on select().
+        self._is_core = False
+        self._statement = Query([attr.target_mapper], None)
+
+        # self._is_core = True
+        # self._statement = sql.select(attr.target_mapper)._set_label_style(
+        #    selectable.LABEL_STYLE_TABLENAME_PLUS_COL
+        # )
+
+        self._autoflush = True
         self.instance = instance = state.obj()
         self.attr = attr
 
-        mapper = object_mapper(instance)
+        self.mapper = mapper = object_mapper(instance)
         prop = mapper._props[self.attr.key]
 
         if prop.secondary is not None:
@@ -277,29 +287,154 @@ class AppenderMixin(object):
             # is in the FROM.  So we purposely put the mapper selectable
             # in _from_obj[0] to ensure a user-defined join() later on
             # doesn't fail, and secondary is then in _from_obj[1].
-            self._from_obj = (prop.mapper.selectable, prop.secondary)
+            self._statement = self._statement.select_from(
+                prop.mapper.selectable, prop.secondary
+            )
 
-        self._where_criteria += (
+        self._statement = self._statement.where(
             prop._with_parent(instance, alias_secondary=False),
         )
 
         if self.attr.order_by:
 
-            if (
-                self._order_by_clauses is False
-                or self._order_by_clauses is None
-            ):
-                self._order_by_clauses = tuple(self.attr.order_by)
-            else:
-                self._order_by_clauses = self._order_by_clauses + tuple(
-                    self.attr.order_by
-                )
+            self._statement = self._statement.order_by(*self.attr.order_by)
+
+    @_generative
+    def autoflush(self, setting):
+        """Set autoflush to a specific setting.
+
+        Note that a Session with autoflush=False will
+        not autoflush, even if this flag is set to True at the
+        Query level.  Therefore this flag is usually used only
+        to disable autoflush for a specific Query.
+
+        """
+        self._autoflush = setting
+
+    @property
+    def statement(self):
+        """Return the Core statement represented by this
+        :class:`.AppenderQuery`.
+
+        """
+        if self._is_core:
+            return self._statement._set_label_style(
+                selectable.LABEL_STYLE_DISAMBIGUATE_ONLY
+            )
+
+        else:
+            return self._statement.statement
+
+    def filter(self, *criteria):
+        """A synonym for the :meth:`_orm.AppenderQuery.where` method."""
+
+        return self.where(*criteria)
+
+    @_generative
+    def where(self, *criteria):
+        r"""Apply the given WHERE criterion, using SQL expressions.
+
+        Equivalent to :meth:`.Select.where`.
+
+        """
+        self._statement = self._statement.where(*criteria)
+
+    @_generative
+    def order_by(self, *criteria):
+        r"""Apply the given ORDER BY criterion, using SQL expressions.
+
+        Equivalent to :meth:`.Select.order_by`.
+
+        """
+        self._statement = self._statement.order_by(*criteria)
+
+    @_generative
+    def filter_by(self, **kwargs):
+        r"""Apply the given filtering criterion using keyword expressions.
+
+        Equivalent to :meth:`.Select.filter_by`.
+
+        """
+        self._statement = self._statement.filter_by(**kwargs)
+
+    @_generative
+    def join(self, target, *props, **kwargs):
+        r"""Create a SQL JOIN against this
+        object's criterion.
+
+        Equivalent to :meth:`.Select.join`.
+        """
+
+        self._statement = self._statement.join(target, *props, **kwargs)
+
+    @_generative
+    def outerjoin(self, target, *props, **kwargs):
+        r"""Create a SQL LEFT OUTER JOIN against this
+        object's criterion.
+
+        Equivalent to :meth:`.Select.outerjoin`.
+
+        """
+
+        self._statement = self._statement.outerjoin(target, *props, **kwargs)
+
+    def scalar(self):
+        """Return the first element of the first result or None
+        if no rows present.  If multiple rows are returned,
+        raises MultipleResultsFound.
+
+        Equivalent to :meth:`_query.Query.scalar`.
+
+        .. versionadded:: 1.1.6
+
+        """
+        return self._iter().scalar()
+
+    def first(self):
+        """Return the first row.
+
+        Equivalent to :meth:`_query.Query.first`.
+
+        """
+
+        # replicates limit(1) behavior
+        if self._statement is not None:
+            return self._iter().first()
+        else:
+            return self.limit(1)._iter().first()
+
+    def one(self):
+        """Return exactly one result or raise an exception.
+
+        Equivalent to :meth:`_query.Query.one`.
+
+        """
+        return self._iter().one()
+
+    def one_or_none(self):
+        """Return one or zero results, or raise an exception for multiple
+        rows.
+
+        Equivalent to :meth:`_query.Query.one_or_none`.
+
+        .. versionadded:: 1.0.9
+
+        """
+        return self._iter().one_or_none()
+
+    def all(self):
+        """Return all rows.
+
+        Equivalent to :meth:`_query.Query.all`.
+
+        """
+        return self._iter().all()
 
     def session(self):
         sess = object_session(self.instance)
         if (
             sess is not None
-            and self.autoflush
+            and self._autoflush
             and sess.autoflush
             and self.instance in sess
         ):
@@ -311,17 +446,60 @@ class AppenderMixin(object):
 
     session = property(session, lambda s, x: None)
 
-    def __iter__(self):
+    def _execute(self, sess=None):
+        # note we're returning an entirely new Query class instance
+        # here without any assignment capabilities; the class of this
+        # query is determined by the session.
+        instance = self.instance
+        if sess is None:
+            sess = object_session(instance)
+            if sess is None:
+                raise orm_exc.DetachedInstanceError(
+                    "Parent instance %s is not bound to a Session, and no "
+                    "contextual session is established; lazy load operation "
+                    "of attribute '%s' cannot proceed"
+                    % (orm_util.instance_str(instance), self.attr.key)
+                )
+
+        result = sess.execute(self._statement, future=True)
+        result = result.scalars()
+
+        if result._attributes.get("filtered", False):
+            result = result.unique()
+
+        return result
+
+    def _iter(self):
         sess = self.session
         if sess is None:
-            return iter(
-                self.attr._get_collection_history(
-                    attributes.instance_state(self.instance),
-                    attributes.PASSIVE_NO_INITIALIZE,
-                ).added_items
-            )
+            instance = self.instance
+            state = attributes.instance_state(instance)
+
+            if state.detached:
+                raise orm_exc.DetachedInstanceError(
+                    "Parent instance %s is not bound to a Session, and no "
+                    "contextual session is established; lazy load operation "
+                    "of attribute '%s' cannot proceed"
+                    % (orm_util.instance_str(instance), self.attr.key)
+                )
+            else:
+                iterator = (
+                    (item,)
+                    for item in self.attr._get_collection_history(
+                        state, attributes.PASSIVE_NO_INITIALIZE,
+                    ).added_items
+                )
+
+                row_metadata = _result.SimpleResultMetaData(
+                    (self.mapper.class_.__name__,), [], _unique_filters=[id],
+                )
+
+                return _result.IteratorResult(row_metadata, iterator).scalars()
         else:
-            return iter(self._generate(sess))
+            return self._execute(sess)
+
+    def __iter__(self):
+        return iter(self._iter())
 
     def __getitem__(self, index):
         sess = self.session
@@ -331,9 +509,32 @@ class AppenderMixin(object):
                 attributes.PASSIVE_NO_INITIALIZE,
             ).indexed(index)
         else:
-            return self._generate(sess).__getitem__(index)
+            return orm_util._getitem(self, index)
+
+    def slice(self, start, stop):
+        """Computes the "slice" represented by
+        the given indices and apply as LIMIT/OFFSET.
+
+
+        """
+        limit_clause, offset_clause = orm_util._make_slice(
+            self._statement._limit_clause,
+            self._statement._offset_clause,
+            start,
+            stop,
+        )
+        self._statement = self._statement.limit(limit_clause).offset(
+            offset_clause
+        )
 
     def count(self):
+        """return the 'count'.
+
+        Equivalent to :meth:`_query.Query.count`.
+
+
+        """
+
         sess = self.session
         if sess is None:
             return len(
@@ -343,33 +544,10 @@ class AppenderMixin(object):
                 ).added_items
             )
         else:
-            return self._generate(sess).count()
-
-    def _generate(self, sess=None):
-        # note we're returning an entirely new Query class instance
-        # here without any assignment capabilities; the class of this
-        # query is determined by the session.
-        instance = self.instance
-        if sess is None:
-            sess = object_session(instance)
-            if sess is None:
-                raise orm_exc.DetachedInstanceError(
-                    "Parent instance %s is not bound to a Session, and no "
-                    "contextual session is established; lazy load operation "
-                    "of attribute '%s' cannot proceed"
-                    % (orm_util.instance_str(instance), self.attr.key)
-                )
-
-        if self.query_class:
-            query = self.query_class(self.attr.target_mapper, session=sess)
-        else:
-            query = sess.query(self.attr.target_mapper)
-
-        query._where_criteria = self._where_criteria
-        query._from_obj = self._from_obj
-        query._order_by_clauses = self._order_by_clauses
+            col = sql.func.count(sql.literal_column("*"))
 
-        return query
+            stmt = sql.select(col).select_from(self._statement.subquery())
+            return self.session.execute(stmt).scalar()
 
     def extend(self, iterator):
         for item in iterator:
@@ -397,16 +575,6 @@ class AppenderMixin(object):
         )
 
 
-class AppenderQuery(AppenderMixin, Query):
-    """A dynamic query that supports basic collection storage operations."""
-
-
-def mixin_user_query(cls):
-    """Return a new class with AppenderQuery functionality layered over."""
-    name = "Appender" + cls.__name__
-    return type(name, (AppenderMixin, cls), {"query_class": cls})
-
-
 class CollectionHistory(object):
     """Overrides AttributeHistory to receive append/remove events directly."""
 
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 55c2b79f5..8d1ae2e69 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -346,7 +346,7 @@ def load_on_pk_identity(
         load_options = QueryContext.default_load_options
 
     compile_options = ORMCompileState.default_compile_options.safe_merge(
-        q.compile_options
+        q._compile_options
     )
 
     if primary_key_identity is not None:
@@ -411,7 +411,7 @@ def load_on_pk_identity(
     # TODO: most of the compile_options that are not legacy only involve this
     # function, so try to see if handling of them can mostly be local to here
 
-    q.compile_options, load_options = _set_get_options(
+    q._compile_options, load_options = _set_get_options(
         compile_options,
         load_options,
         populate_existing=bool(refresh_state),
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 1b2779c00..a78af92b9 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1762,24 +1762,23 @@ class BulkUDCompileState(CompileState):
         if is_reentrant_invoke:
             return statement, execution_options
 
-        sync = execution_options.get("synchronize_session", None)
-        if sync is None:
-            sync = statement._execution_options.get(
-                "synchronize_session", None
-            )
-
-        update_options = execution_options.get(
+        (
+            update_options,
+            execution_options,
+        ) = BulkUDCompileState.default_update_options.from_execution_options(
             "_sa_orm_update_options",
-            BulkUDCompileState.default_update_options,
+            {"synchronize_session"},
+            execution_options,
+            statement._execution_options,
         )
 
+        sync = update_options._synchronize_session
         if sync is not None:
             if sync not in ("evaluate", "fetch", False):
                 raise sa_exc.ArgumentError(
                     "Valid strategies for session synchronization "
                     "are 'evaluate', 'fetch', False"
                 )
-            update_options += {"_synchronize_session": sync}
 
         bind_arguments["clause"] = statement
         try:
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index acc76094b..7bf69f99f 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -22,10 +22,10 @@ import itertools
 import operator
 import types
 
-from . import attributes
 from . import exc as orm_exc
 from . import interfaces
 from . import loading
+from . import util as orm_util
 from .base import _assertions
 from .context import _column_descriptions
 from .context import _legacy_determine_last_joined_entity
@@ -121,7 +121,7 @@ class Query(
     _legacy_setup_joins = ()
     _label_style = LABEL_STYLE_NONE
 
-    compile_options = ORMCompileState.default_compile_options
+    _compile_options = ORMCompileState.default_compile_options
 
     load_options = QueryContext.default_load_options
 
@@ -215,7 +215,7 @@ class Query(
             for elem in obj
         ]
 
-        self.compile_options += {"_set_base_alias": set_base_alias}
+        self._compile_options += {"_set_base_alias": set_base_alias}
         self._from_obj = tuple(fa)
 
     @_generative
@@ -254,7 +254,7 @@ class Query(
 
         self._from_obj = self._legacy_setup_joins = ()
         if self._statement is not None:
-            self.compile_options += {"_statement": None}
+            self._compile_options += {"_statement": None}
         self._where_criteria = ()
         self._distinct = False
 
@@ -320,7 +320,7 @@ class Query(
         if load_options:
             self.load_options += load_options
         if compile_options:
-            self.compile_options += compile_options
+            self._compile_options += compile_options
 
         return self
 
@@ -357,8 +357,8 @@ class Query(
         # passed into the execute process and wont generate its own cache
         # key; this will all occur in terms of the ORM-enabled Select.
         if (
-            not self.compile_options._set_base_alias
-            and not self.compile_options._with_polymorphic_adapt_map
+            not self._compile_options._set_base_alias
+            and not self._compile_options._with_polymorphic_adapt_map
         ):
             # if we don't have legacy top level aliasing features in use
             # then convert to a future select() directly
@@ -400,9 +400,9 @@ class Query(
                 if new_query is not None and new_query is not self:
                     self = new_query
                     if not fn._bake_ok:
-                        self.compile_options += {"_bake_ok": False}
+                        self._compile_options += {"_bake_ok": False}
 
-        compile_options = self.compile_options
+        compile_options = self._compile_options
         compile_options += {
             "_for_statement": for_statement,
             "_use_legacy_query_style": use_legacy_query_style,
@@ -413,21 +413,21 @@ class Query(
             stmt.__dict__.update(
                 _with_options=self._with_options,
                 _with_context_options=self._with_context_options,
-                compile_options=compile_options,
+                _compile_options=compile_options,
                 _execution_options=self._execution_options,
+                _propagate_attrs=self._propagate_attrs,
             )
-            stmt._propagate_attrs = self._propagate_attrs
         else:
             # Query / select() internal attributes are 99% cross-compatible
             stmt = Select.__new__(Select)
             stmt.__dict__.update(self.__dict__)
             stmt.__dict__.update(
                 _label_style=self._label_style,
-                compile_options=compile_options,
+                _compile_options=compile_options,
+                _propagate_attrs=self._propagate_attrs,
             )
             stmt.__dict__.pop("session", None)
 
-        stmt._propagate_attrs = self._propagate_attrs
         return stmt
 
     def subquery(
@@ -629,7 +629,7 @@ class Query(
         selectable, or when using :meth:`_query.Query.yield_per`.
 
         """
-        self.compile_options += {"_enable_eagerloads": value}
+        self._compile_options += {"_enable_eagerloads": value}
 
     @_generative
     def with_labels(self):
@@ -710,7 +710,7 @@ class Query(
         query intended for the deferred load.
 
         """
-        self.compile_options += {"_current_path": path}
+        self._compile_options += {"_current_path": path}
 
     # TODO: removed in 2.0
     @_generative
@@ -744,7 +744,7 @@ class Query(
             polymorphic_on=polymorphic_on,
         )
 
-        self.compile_options = self.compile_options.add_to_element(
+        self._compile_options = self._compile_options.add_to_element(
             "_with_polymorphic_adapt_map", ((entity, inspect(wp)),)
         )
 
@@ -818,6 +818,10 @@ class Query(
             {"stream_results": True, "max_row_buffer": count}
         )
 
+    @util.deprecated_20(
+        ":meth:`_orm.Query.get`",
+        alternative="The method is now available as :meth:`_orm.Session.get`",
+    )
     def get(self, ident):
         """Return an instance based on the given primary key identifier,
         or ``None`` if not found.
@@ -858,14 +862,6 @@ class Query(
         however, and will be used if the object is not
         yet locally present.
 
-        A lazy-loading, many-to-one attribute configured
-        by :func:`_orm.relationship`, using a simple
-        foreign-key-to-primary-key criterion, will also use an
-        operation equivalent to :meth:`_query.Query.get` in order to retrieve
-        the target value from the local identity map
-        before querying the database.  See :doc:`/orm/loading_relationships`
-        for further details on relationship loading.
-
         :param ident: A scalar, tuple, or dictionary representing the
          primary key.  For a composite (e.g. multiple column) primary key,
          a tuple or dictionary should be passed.
@@ -905,80 +901,22 @@ class Query(
 
         """
         self._no_criterion_assertion("get", order_by=False, distinct=False)
+
+        # we still implement _get_impl() so that baked query can override
+        # it
         return self._get_impl(ident, loading.load_on_pk_identity)
 
     def _get_impl(self, primary_key_identity, db_load_fn, identity_token=None):
-
-        # convert composite types to individual args
-        if hasattr(primary_key_identity, "__composite_values__"):
-            primary_key_identity = primary_key_identity.__composite_values__()
-
         mapper = self._only_full_mapper_zero("get")
-
-        is_dict = isinstance(primary_key_identity, dict)
-        if not is_dict:
-            primary_key_identity = util.to_list(
-                primary_key_identity, default=(None,)
-            )
-
-        if len(primary_key_identity) != len(mapper.primary_key):
-            raise sa_exc.InvalidRequestError(
-                "Incorrect number of values in identifier to formulate "
-                "primary key for query.get(); primary key columns are %s"
-                % ",".join("'%s'" % c for c in mapper.primary_key)
-            )
-
-        if is_dict:
-            try:
-                primary_key_identity = list(
-                    primary_key_identity[prop.key]
-                    for prop in mapper._identity_key_props
-                )
-
-            except KeyError as err:
-                util.raise_(
-                    sa_exc.InvalidRequestError(
-                        "Incorrect names of values in identifier to formulate "
-                        "primary key for query.get(); primary key attribute "
-                        "names are %s"
-                        % ",".join(
-                            "'%s'" % prop.key
-                            for prop in mapper._identity_key_props
-                        )
-                    ),
-                    replace_context=err,
-                )
-
-        if (
-            not self.load_options._populate_existing
-            and not mapper.always_refresh
-            and self._for_update_arg is None
-        ):
-
-            instance = self.session._identity_lookup(
-                mapper, primary_key_identity, identity_token=identity_token
-            )
-
-            if instance is not None:
-                self._get_existing_condition()
-                # reject calls for id in identity map but class
-                # mismatch.
-                if not issubclass(instance.__class__, mapper.class_):
-                    return None
-                return instance
-            elif instance is attributes.PASSIVE_CLASS_MISMATCH:
-                return None
-
-        # apply_labels() not strictly necessary, however this will ensure that
-        # tablename_colname style is used which at the moment is asserted
-        # in a lot of unit tests :)
-
-        statement = self._statement_20().apply_labels()
-        return db_load_fn(
-            self.session,
-            statement,
+        return self.session._get_impl(
+            mapper,
             primary_key_identity,
-            load_options=self.load_options,
+            db_load_fn,
+            populate_existing=self.load_options._populate_existing,
+            with_for_update=self._for_update_arg,
+            options=self._with_options,
+            identity_token=identity_token,
+            execution_options=self._execution_options,
         )
 
     @property
@@ -1000,7 +938,7 @@ class Query(
 
     @property
     def _current_path(self):
-        return self.compile_options._current_path
+        return self._compile_options._current_path
 
     @_generative
     def correlate(self, *fromclauses):
@@ -1375,7 +1313,7 @@ class Query(
 
     @_generative
     def _set_enable_single_crit(self, val):
-        self.compile_options += {"_enable_single_crit": val}
+        self._compile_options += {"_enable_single_crit": val}
 
     @_generative
     def _from_selectable(self, fromclause, set_entity_from=True):
@@ -1394,7 +1332,7 @@ class Query(
         ):
             self.__dict__.pop(attr, None)
         self._set_select_from([fromclause], set_entity_from)
-        self.compile_options += {
+        self._compile_options += {
             "_enable_single_crit": False,
             "_statement": None,
         }
@@ -1404,7 +1342,7 @@ class Query(
         # legacy.  see test/orm/test_froms.py for various
         # "oldstyle" tests that rely on this and the correspoinding
         # "newtyle" that do not.
-        self.compile_options += {"_orm_only_from_obj_alias": False}
+        self._compile_options += {"_orm_only_from_obj_alias": False}
 
     @util.deprecated(
         "1.4",
@@ -1517,7 +1455,7 @@ class Query(
         """
 
         opts = tuple(util.flatten_iterator(args))
-        if self.compile_options._current_path:
+        if self._compile_options._current_path:
             for opt in opts:
                 if opt._is_legacy_option:
                     opt.process_query_conditionally(self)
@@ -1641,6 +1579,14 @@ class Query(
         params = self.load_options._params.union(kwargs)
         self.load_options += {"_params": params}
 
+    def where(self, *criterion):
+        """A synonym for :meth:`.Query.filter`.
+
+        .. versionadded:: 1.4
+
+        """
+        return self.filter(*criterion)
+
     @_generative
     @_assertions(_no_statement_condition, _no_limit_offset)
     def filter(self, *criterion):
@@ -2204,6 +2150,7 @@ class Query(
             SQLAlchemy versions was the primary ORM-level joining interface.
 
         """
+
         aliased, from_joinpoint, isouter, full = (
             kwargs.pop("aliased", False),
             kwargs.pop("from_joinpoint", False),
@@ -2496,36 +2443,10 @@ class Query(
         """
 
         self._set_select_from([from_obj], True)
-        self.compile_options += {"_enable_single_crit": False}
+        self._compile_options += {"_enable_single_crit": False}
 
     def __getitem__(self, item):
-        if isinstance(item, slice):
-            start, stop, step = util.decode_slice(item)
-
-            if (
-                isinstance(stop, int)
-                and isinstance(start, int)
-                and stop - start <= 0
-            ):
-                return []
-
-            # perhaps we should execute a count() here so that we
-            # can still use LIMIT/OFFSET ?
-            elif (isinstance(start, int) and start < 0) or (
-                isinstance(stop, int) and stop < 0
-            ):
-                return list(self)[item]
-
-            res = self.slice(start, stop)
-            if step is not None:
-                return list(res)[None : None : item.step]
-            else:
-                return list(res)
-        else:
-            if item == -1:
-                return list(self)[-1]
-            else:
-                return list(self[item : item + 1])[0]
+        return orm_util._getitem(self, item)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2559,46 +2480,10 @@ class Query(
            :meth:`_query.Query.offset`
 
         """
-        # for calculated limit/offset, try to do the addition of
-        # values to offset in Python, howver if a SQL clause is present
-        # then the addition has to be on the SQL side.
-        if start is not None and stop is not None:
-            offset_clause = self._offset_or_limit_clause_asint_if_possible(
-                self._offset_clause
-            )
-            if offset_clause is None:
-                offset_clause = 0
 
-            if start != 0:
-                offset_clause = offset_clause + start
-
-            if offset_clause == 0:
-                self._offset_clause = None
-            else:
-                self._offset_clause = self._offset_or_limit_clause(
-                    offset_clause
-                )
-
-            self._limit_clause = self._offset_or_limit_clause(stop - start)
-
-        elif start is None and stop is not None:
-            self._limit_clause = self._offset_or_limit_clause(stop)
-        elif start is not None and stop is None:
-            offset_clause = self._offset_or_limit_clause_asint_if_possible(
-                self._offset_clause
-            )
-            if offset_clause is None:
-                offset_clause = 0
-
-            if start != 0:
-                offset_clause = offset_clause + start
-
-            if offset_clause == 0:
-                self._offset_clause = None
-            else:
-                self._offset_clause = self._offset_or_limit_clause(
-                    offset_clause
-                )
+        self._limit_clause, self._offset_clause = orm_util._make_slice(
+            self._limit_clause, self._offset_clause, start, stop
+        )
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2607,7 +2492,7 @@ class Query(
         ``Query``.
 
         """
-        self._limit_clause = self._offset_or_limit_clause(limit)
+        self._limit_clause = orm_util._offset_or_limit_clause(limit)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2616,31 +2501,7 @@ class Query(
         ``Query``.
 
         """
-        self._offset_clause = self._offset_or_limit_clause(offset)
-
-    def _offset_or_limit_clause(self, element, name=None, type_=None):
-        """Convert the given value to an "offset or limit" clause.
-
-        This handles incoming integers and converts to an expression; if
-        an expression is already given, it is passed through.
-
-        """
-        return coercions.expect(
-            roles.LimitOffsetRole, element, name=name, type_=type_
-        )
-
-    def _offset_or_limit_clause_asint_if_possible(self, clause):
-        """Return the offset or limit clause as a simple integer if possible,
-        else return the clause.
-
-        """
-        if clause is None:
-            return None
-        if hasattr(clause, "_limit_offset_value"):
-            value = clause._limit_offset_value
-            return util.asint(value)
-        else:
-            return clause
+        self._offset_clause = orm_util._offset_or_limit_clause(offset)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2723,7 +2584,7 @@ class Query(
             roles.SelectStatementRole, statement, apply_propagate_attrs=self
         )
         self._statement = statement
-        self.compile_options += {"_statement": statement}
+        self._compile_options += {"_statement": statement}
 
     def first(self):
         """Return the first result of this ``Query`` or
@@ -3088,110 +2949,22 @@ class Query(
             sess.query(User).filter(User.age == 25).\
                 delete(synchronize_session='evaluate')
 
-        .. warning:: The :meth:`_query.Query.delete`
-           method is a "bulk" operation,
-           which bypasses ORM unit-of-work automation in favor of greater
-           performance.  **Please read all caveats and warnings below.**
-
-        :param synchronize_session: chooses the strategy for the removal of
-            matched objects from the session. Valid values are:
-
-            ``False`` - don't synchronize the session. This option is the most
-            efficient and is reliable once the session is expired, which
-            typically occurs after a commit(), or explicitly using
-            expire_all(). Before the expiration, objects may still remain in
-            the session which were in fact deleted which can lead to confusing
-            results if they are accessed via get() or already loaded
-            collections.
-
-            ``'fetch'`` - performs a select query before the delete to find
-            objects that are matched by the delete query and need to be
-            removed from the session. Matched objects are removed from the
-            session.
+        .. warning::
 
-            ``'evaluate'`` - Evaluate the query's criteria in Python straight
-            on the objects in the session. If evaluation of the criteria isn't
-            implemented, an error is raised.
+            See the section :ref:`bulk_update_delete` for important caveats
+            and warnings, including limitations when using bulk UPDATE
+            and DELETE with mapper inheritance configurations.
 
-            The expression evaluator currently doesn't account for differing
-            string collations between the database and Python.
+        :param synchronize_session: chooses the strategy to update the
+         attributes on objects in the session.   See the section
+         :ref:`bulk_update_delete` for a discussion of these strategies.
 
         :return: the count of rows matched as returned by the database's
           "row count" feature.
 
-        .. warning:: **Additional Caveats for bulk query deletes**
-
-            * This method does **not work for joined
-              inheritance mappings**, since the **multiple table
-              deletes are not supported by SQL** as well as that the
-              **join condition of an inheritance mapper is not
-              automatically rendered**.  Care must be taken in any
-              multiple-table delete to first accommodate via some other means
-              how the related table will be deleted, as well as to
-              explicitly include the joining
-              condition between those tables, even in mappings where
-              this is normally automatic. E.g. if a class ``Engineer``
-              subclasses ``Employee``, a DELETE against the ``Employee``
-              table would look like::
-
-                    session.query(Engineer).\
-                        filter(Engineer.id == Employee.id).\
-                        filter(Employee.name == 'dilbert').\
-                        delete()
-
-              However the above SQL will not delete from the Engineer table,
-              unless an ON DELETE CASCADE rule is established in the database
-              to handle it.
-
-              Short story, **do not use this method for joined inheritance
-              mappings unless you have taken the additional steps to make
-              this feasible**.
-
-            * The polymorphic identity WHERE criteria is **not** included
-              for single- or
-              joined- table updates - this must be added **manually** even
-              for single table inheritance.
-
-            * The method does **not** offer in-Python cascading of
-              relationships - it is assumed that ON DELETE CASCADE/SET
-              NULL/etc. is configured for any foreign key references
-              which require it, otherwise the database may emit an
-              integrity violation if foreign key references are being
-              enforced.
-
-              After the DELETE, dependent objects in the
-              :class:`.Session` which were impacted by an ON DELETE
-              may not contain the current state, or may have been
-              deleted. This issue is resolved once the
-              :class:`.Session` is expired, which normally occurs upon
-              :meth:`.Session.commit` or can be forced by using
-              :meth:`.Session.expire_all`.  Accessing an expired
-              object whose row has been deleted will invoke a SELECT
-              to locate the row; when the row is not found, an
-              :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is
-              raised.
-
-            * The ``'fetch'`` strategy results in an additional
-              SELECT statement emitted and will significantly reduce
-              performance.
-
-            * The ``'evaluate'`` strategy performs a scan of
-              all matching objects within the :class:`.Session`; if the
-              contents of the :class:`.Session` are expired, such as
-              via a proceeding :meth:`.Session.commit` call, **this will
-              result in SELECT queries emitted for every matching object**.
-
-            * The :meth:`.MapperEvents.before_delete` and
-              :meth:`.MapperEvents.after_delete`
-              events **are not invoked** from this method.  Instead, the
-              :meth:`.SessionEvents.after_bulk_delete` method is provided to
-              act upon a mass DELETE of entity rows.
-
         .. seealso::
 
-            :meth:`_query.Query.update`
-
-            :ref:`inserts_and_updates` - Core SQL tutorial
+            :ref:`bulk_update_delete`
 
         """
 
@@ -3231,12 +3004,11 @@ class Query(
             sess.query(User).filter(User.age == 25).\
                 update({"age": User.age - 10}, synchronize_session='evaluate')
 
+        .. warning::
 
-        .. warning:: The :meth:`_query.Query.update`
-           method is a "bulk" operation,
-           which bypasses ORM unit-of-work automation in favor of greater
-           performance.  **Please read all caveats and warnings below.**
-
+            See the section :ref:`bulk_update_delete` for important caveats
+            and warnings, including limitations when using bulk UPDATE
+            and DELETE with mapper inheritance configurations.
 
         :param values: a dictionary with attributes names, or alternatively
          mapped attributes or SQL expressions, as keys, and literal
@@ -3248,31 +3020,9 @@ class Query(
          flag is passed to the :paramref:`.Query.update.update_args` dictionary
          as well.
 
-          .. versionchanged:: 1.0.0 - string names in the values dictionary
-             are now resolved against the mapped entity; previously, these
-             strings were passed as literal column names with no mapper-level
-             translation.
-
         :param synchronize_session: chooses the strategy to update the
-         attributes on objects in the session. Valid values are:
-
-            ``False`` - don't synchronize the session. This option is the most
-            efficient and is reliable once the session is expired, which
-            typically occurs after a commit(), or explicitly using
-            expire_all(). Before the expiration, updated objects may still
-            remain in the session with stale values on their attributes, which
-            can lead to confusing results.
-
-            ``'fetch'`` - performs a select query before the update to find
-            objects that are matched by the update query. The updated
-            attributes are expired on matched objects.
-
-            ``'evaluate'`` - Evaluate the Query's criteria in Python straight
-            on the objects in the session. If evaluation of the criteria isn't
-            implemented, an exception is raised.
-
-            The expression evaluator currently doesn't account for differing
-            string collations between the database and Python.
+         attributes on objects in the session.   See the section
+         :ref:`bulk_update_delete` for a discussion of these strategies.
 
         :param update_args: Optional dictionary, if present will be passed
          to the underlying :func:`_expression.update`
@@ -3281,70 +3031,14 @@ class Query(
          as ``mysql_limit``, as well as other special arguments such as
          :paramref:`~sqlalchemy.sql.expression.update.preserve_parameter_order`.
 
-         .. versionadded:: 1.0.0
-
         :return: the count of rows matched as returned by the database's
          "row count" feature.
 
-        .. warning:: **Additional Caveats for bulk query updates**
-
-            * The method does **not** offer in-Python cascading of
-              relationships - it is assumed that ON UPDATE CASCADE is
-              configured for any foreign key references which require
-              it, otherwise the database may emit an integrity
-              violation if foreign key references are being enforced.
-
-              After the UPDATE, dependent objects in the
-              :class:`.Session` which were impacted by an ON UPDATE
-              CASCADE may not contain the current state; this issue is
-              resolved once the :class:`.Session` is expired, which
-              normally occurs upon :meth:`.Session.commit` or can be
-              forced by using :meth:`.Session.expire_all`.
-
-            * The ``'fetch'`` strategy results in an additional
-              SELECT statement emitted and will significantly reduce
-              performance.
-
-            * The ``'evaluate'`` strategy performs a scan of
-              all matching objects within the :class:`.Session`; if the
-              contents of the :class:`.Session` are expired, such as
-              via a proceeding :meth:`.Session.commit` call, **this will
-              result in SELECT queries emitted for every matching object**.
-
-            * The method supports multiple table updates, as detailed
-              in :ref:`multi_table_updates`, and this behavior does
-              extend to support updates of joined-inheritance and
-              other multiple table mappings.  However, the **join
-              condition of an inheritance mapper is not
-              automatically rendered**. Care must be taken in any
-              multiple-table update to explicitly include the joining
-              condition between those tables, even in mappings where
-              this is normally automatic. E.g. if a class ``Engineer``
-              subclasses ``Employee``, an UPDATE of the ``Engineer``
-              local table using criteria against the ``Employee``
-              local table might look like::
-
-                    session.query(Engineer).\
-                        filter(Engineer.id == Employee.id).\
-                        filter(Employee.name == 'dilbert').\
-                        update({"engineer_type": "programmer"})
-
-            * The polymorphic identity WHERE criteria is **not** included
-              for single- or
-              joined- table updates - this must be added **manually**, even
-              for single table inheritance.
-
-            * The :meth:`.MapperEvents.before_update` and
-              :meth:`.MapperEvents.after_update`
-              events **are not invoked from this method**.  Instead, the
-              :meth:`.SessionEvents.after_bulk_update` method is provided to
-              act upon a mass UPDATE of entity rows.
 
         .. seealso::
 
-            :meth:`_query.Query.delete`
+            :ref:`bulk_update_delete`
 
-            :ref:`inserts_and_updates` - Core SQL tutorial
 
         """
 
@@ -3390,7 +3084,7 @@ class Query(
         """
 
         stmt = self._statement_20(for_statement=for_statement, **kw)
-        assert for_statement == stmt.compile_options._for_statement
+        assert for_statement == stmt._compile_options._for_statement
 
         # this chooses between ORMFromStatementCompileState and
         # ORMSelectCompileState.  We could also base this on
@@ -3422,7 +3116,7 @@ class FromStatement(SelectStatementGrouping, Executable):
 
     __visit_name__ = "orm_from_statement"
 
-    compile_options = ORMFromStatementCompileState.default_compile_options
+    _compile_options = ORMFromStatementCompileState.default_compile_options
 
     _compile_state_factory = ORMFromStatementCompileState.create_for_statement
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index f4f7374e4..3d2f26e0d 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -30,6 +30,7 @@ from .unitofwork import UOWTransaction
 from .. import engine
 from .. import exc as sa_exc
 from .. import future
+from .. import sql
 from .. import util
 from ..inspection import inspect
 from ..sql import coercions
@@ -288,7 +289,7 @@ class ORMExecuteState(util.MemoizedSlots):
         return self.statement._execution_options.union(self._execution_options)
 
     def _orm_compile_options(self):
-        opts = self.statement.compile_options
+        opts = self.statement._compile_options
         if isinstance(opts, context.ORMCompileState.default_compile_options):
             return opts
         else:
@@ -367,134 +368,53 @@ class ORMExecuteState(util.MemoizedSlots):
 class SessionTransaction(object):
     """A :class:`.Session`-level transaction.
 
-    :class:`.SessionTransaction` is a mostly behind-the-scenes object
-    not normally referenced directly by application code.   It coordinates
-    among multiple :class:`_engine.Connection` objects, maintaining a database
-    transaction for each one individually, committing or rolling them
-    back all at once.   It also provides optional two-phase commit behavior
-    which can augment this coordination operation.
-
-    The :attr:`.Session.transaction` attribute of :class:`.Session`
-    refers to the current :class:`.SessionTransaction` object in use, if any.
-    The :attr:`.SessionTransaction.parent` attribute refers to the parent
-    :class:`.SessionTransaction` in the stack of :class:`.SessionTransaction`
-    objects.  If this attribute is ``None``, then this is the top of the stack.
-    If non-``None``, then this :class:`.SessionTransaction` refers either
-    to a so-called "subtransaction" or a "nested" transaction.  A
-    "subtransaction" is a scoping concept that demarcates an inner portion
-    of the outermost "real" transaction.  A nested transaction, which
-    is indicated when the :attr:`.SessionTransaction.nested`
-    attribute is also True, indicates that this :class:`.SessionTransaction`
-    corresponds to a SAVEPOINT.
-
-    **Life Cycle**
-
-    A :class:`.SessionTransaction` is associated with a :class:`.Session` in
-    its default mode of ``autocommit=False`` whenever the "autobegin" process
-    takes place, associated with no database connections.  As the
-    :class:`.Session` is called upon to emit SQL on behalf of various
-    :class:`_engine.Engine` or :class:`_engine.Connection` objects,
-    a corresponding
-    :class:`_engine.Connection` and associated :class:`.Transaction`
-    is added to a
-    collection within the :class:`.SessionTransaction` object, becoming one of
-    the connection/transaction pairs maintained by the
-    :class:`.SessionTransaction`.  The start of a :class:`.SessionTransaction`
-    can be tracked using the :meth:`.SessionEvents.after_transaction_create`
-    event.
-
-    The lifespan of the :class:`.SessionTransaction` ends when the
-    :meth:`.Session.commit`, :meth:`.Session.rollback` or
-    :meth:`.Session.close` methods are called.  At this point, the
-    :class:`.SessionTransaction` removes its association with its parent
-    :class:`.Session`.   A :class:`.Session` that is in ``autocommit=False``
-    mode will create a new :class:`.SessionTransaction` to replace it when the
-    next "autobegin" event occurs, whereas a :class:`.Session` that's in
-    ``autocommit=True`` mode will remain without a :class:`.SessionTransaction`
-    until the :meth:`.Session.begin` method is called.  The end of a
-    :class:`.SessionTransaction` can be tracked using the
-    :meth:`.SessionEvents.after_transaction_end` event.
-
-    .. versionchanged:: 1.4 the :class:`.SessionTransaction` is not created
-       immediately within a :class:`.Session` when constructed or when the
-       previous transaction is removed, it instead is created when the
-       :class:`.Session` is next used.
-
-    **Nesting and Subtransactions**
-
-    Another detail of :class:`.SessionTransaction` behavior is that it is
-    capable of "nesting".  This means that the :meth:`.Session.begin` method
-    can be called while an existing :class:`.SessionTransaction` is already
-    present, producing a new :class:`.SessionTransaction` that temporarily
-    replaces the parent :class:`.SessionTransaction`.   When a
-    :class:`.SessionTransaction` is produced as nested, it assigns itself to
-    the :attr:`.Session.transaction` attribute, and it additionally will assign
-    the previous :class:`.SessionTransaction` to its :attr:`.Session.parent`
-    attribute.  The behavior is effectively a
-    stack, where :attr:`.Session.transaction` refers to the current head of
-    the stack, and the :attr:`.SessionTransaction.parent` attribute allows
-    traversal up the stack until :attr:`.SessionTransaction.parent` is
-    ``None``, indicating the top of the stack.
-
-    When the scope of :class:`.SessionTransaction` is ended via
-    :meth:`.Session.commit` or :meth:`.Session.rollback`, it restores its
-    parent :class:`.SessionTransaction` back onto the
-    :attr:`.Session.transaction` attribute.
-
-    The purpose of this stack is to allow nesting of
-    :meth:`.Session.rollback` or :meth:`.Session.commit` calls in context
-    with various flavors of :meth:`.Session.begin`. This nesting behavior
-    applies to when :meth:`.Session.begin_nested` is used to emit a
-    SAVEPOINT transaction, and is also used to produce a so-called
-    "subtransaction" which allows a block of code to use a
-    begin/rollback/commit sequence regardless of whether or not its enclosing
-    code block has begun a transaction.  The :meth:`.flush` method, whether
-    called explicitly or via autoflush, is the primary consumer of the
-    "subtransaction" feature, in that it wishes to guarantee that it works
-    within in a transaction block regardless of whether or not the
-    :class:`.Session` is in transactional mode when the method is called.
-
-    Note that the flush process that occurs within the "autoflush" feature
-    as well as when the :meth:`.Session.flush` method is used **always**
-    creates a :class:`.SessionTransaction` object.   This object is normally
-    a subtransaction, unless the :class:`.Session` is in autocommit mode
-    and no transaction exists at all, in which case it's the outermost
-    transaction.   Any event-handling logic or other inspection logic
-    needs to take into account whether a :class:`.SessionTransaction`
-    is the outermost transaction, a subtransaction, or a "nested" / SAVEPOINT
-    transaction.
+    :class:`.SessionTransaction` is produced from the
+    :meth:`_orm.Session.begin`
+    and :meth:`_orm.Session.begin_nested` methods.   It's largely an internal
+    object that in modern use provides a context manager for session
+    transactions.
 
-    .. seealso::
+    Documentation on interacting with :class:`_orm.SessionTransaction` is
+    at: :ref:`unitofwork_transaction`.
 
-        :meth:`.Session.rollback`
 
-        :meth:`.Session.commit`
+    .. versionchanged:: 1.4  The scoping and API methods to work with the
+       :class:`_orm.SessionTransaction` object directly have been simplified.
+
+    .. seealso::
+
+        :ref:`unitofwork_transaction`
 
         :meth:`.Session.begin`
 
         :meth:`.Session.begin_nested`
 
-        :attr:`.Session.is_active`
+        :meth:`.Session.rollback`
+
+        :meth:`.Session.commit`
 
-        :meth:`.SessionEvents.after_transaction_create`
+        :meth:`.Session.in_transaction`
 
-        :meth:`.SessionEvents.after_transaction_end`
+        :meth:`.Session.in_nested_transaction`
 
-        :meth:`.SessionEvents.after_commit`
+        :meth:`.Session.get_transaction`
 
-        :meth:`.SessionEvents.after_rollback`
+        :meth:`.Session.get_nested_transaction`
 
-        :meth:`.SessionEvents.after_soft_rollback`
 
     """
 
     _rollback_exception = None
 
-    def __init__(self, session, parent=None, nested=False, autobegin=False):
+    def __init__(
+        self, session, parent=None, nested=False, autobegin=False,
+    ):
         self.session = session
         self._connections = {}
         self._parent = parent
         self.nested = nested
+        if nested:
+            self._previous_nested_transaction = session._nested_transaction
         self._state = ACTIVE
         if not parent and nested:
             raise sa_exc.InvalidRequestError(
@@ -688,6 +608,8 @@ class SessionTransaction(object):
             return self._connections[bind][0]
 
         local_connect = False
+        should_commit = True
+
         if self._parent:
             conn = self._parent._connection_for_bind(bind, execution_options)
             if not self.nested:
@@ -712,11 +634,16 @@ class SessionTransaction(object):
                 transaction = conn.begin_twophase()
             elif self.nested:
                 transaction = conn.begin_nested()
-            else:
-                if conn._is_future and conn.in_transaction():
-                    transaction = conn._transaction
+            elif conn.in_transaction():
+                # if given a future connection already in a transaction, don't
+                # commit that transaction unless it is a savepoint
+                if conn.in_nested_transaction():
+                    transaction = conn.get_nested_transaction()
                 else:
-                    transaction = conn.begin()
+                    transaction = conn.get_transaction()
+                    should_commit = False
+            else:
+                transaction = conn.begin()
         except:
             # connection will not not be associated with this Session;
             # close it immediately so that it isn't closed under GC
@@ -729,7 +656,7 @@ class SessionTransaction(object):
             self._connections[conn] = self._connections[conn.engine] = (
                 conn,
                 transaction,
-                not bind_is_connection or not conn._is_future,
+                should_commit,
                 not bind_is_connection,
             )
             self.session.dispatch.after_begin(self.session, self, conn)
@@ -748,7 +675,7 @@ class SessionTransaction(object):
         if self._parent is None or self.nested:
             self.session.dispatch.before_commit(self.session)
 
-        stx = self.session.transaction
+        stx = self.session._transaction
         if stx is not self:
             for subtransaction in stx._iterate_self_and_parents(upto=self):
                 subtransaction.commit()
@@ -775,7 +702,7 @@ class SessionTransaction(object):
 
         self._state = PREPARED
 
-    def commit(self):
+    def commit(self, _to_root=False):
         self._assert_active(prepared_ok=True)
         if self._state is not PREPARED:
             self._prepare_impl()
@@ -793,12 +720,16 @@ class SessionTransaction(object):
             self._remove_snapshot()
 
         self.close()
+
+        if _to_root and self._parent:
+            return self._parent.commit(_to_root=True)
+
         return self._parent
 
-    def rollback(self, _capture_exception=False):
+    def rollback(self, _capture_exception=False, _to_root=False):
         self._assert_active(prepared_ok=True, rollback_ok=True)
 
-        stx = self.session.transaction
+        stx = self.session._transaction
         if stx is not self:
             for subtransaction in stx._iterate_self_and_parents(upto=self):
                 subtransaction.close()
@@ -849,20 +780,28 @@ class SessionTransaction(object):
 
         sess.dispatch.after_soft_rollback(sess, self)
 
+        if _to_root and self._parent:
+            return self._parent.rollback(_to_root=True)
         return self._parent
 
     def close(self, invalidate=False):
+        if self.nested:
+            self.session._nested_transaction = (
+                self._previous_nested_transaction
+            )
+
         self.session._transaction = self._parent
+
         if self._parent is None:
             for connection, transaction, should_commit, autoclose in set(
                 self._connections.values()
             ):
                 if invalidate:
                     connection.invalidate()
+                if should_commit and transaction.is_active:
+                    transaction.close()
                 if autoclose:
                     connection.close()
-                else:
-                    transaction.close()
 
         self._state = CLOSED
         self.session.dispatch.after_transaction_end(self.session, self)
@@ -924,6 +863,15 @@ class Session(_SessionClassMethods):
         "scalar",
     )
 
+    @util.deprecated_params(
+        autocommit=(
+            "2.0",
+            "The :paramref:`.Session.autocommit` parameter is deprecated "
+            "and will be removed in SQLAlchemy version 2.0.  Please use the "
+            ":paramref:`.Session.autobegin` parameter set to False to support "
+            "explicit use of the :meth:`.Session.begin` method.",
+        ),
+    )
     def __init__(
         self,
         bind=None,
@@ -1071,8 +1019,6 @@ class Session(_SessionClassMethods):
            :class:`.Session` dictionary will be local to that
            :class:`.Session`.
 
-           .. versionadded:: 0.9.0
-
         :param query_cls:  Class which should be used to create new Query
           objects, as returned by the :meth:`~.Session.query` method.
           Defaults to :class:`_query.Query`.
@@ -1096,13 +1042,23 @@ class Session(_SessionClassMethods):
         self._flushing = False
         self._warn_on_events = False
         self._transaction = None
+        self._nested_transaction = None
         self.future = future
         self.hash_key = _new_sessionid()
         self.autoflush = autoflush
-        self.autocommit = autocommit
         self.expire_on_commit = expire_on_commit
         self.enable_baked_queries = enable_baked_queries
 
+        if autocommit:
+            if future:
+                raise sa_exc.ArgumentError(
+                    "Cannot use autocommit mode with future=True.  "
+                    "use the autobegin flag."
+                )
+            self.autocommit = True
+        else:
+            self.autocommit = False
+
         self.twophase = twophase
         self._query_cls = query_cls if query_cls else query.Query
         if info:
@@ -1116,21 +1072,77 @@ class Session(_SessionClassMethods):
 
     connection_callable = None
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type_, value, traceback):
+        self.close()
+
     @property
+    @util.deprecated_20(
+        "The :attr:`_orm.Session.transaction` accessor is deprecated and "
+        "will be removed in SQLAlchemy version 2.0.  "
+        "For context manager use, use :meth:`_orm.Session.begin`.  To access "
+        "the current root transaction, use "
+        ":meth:`_orm.Session.get_transaction()"
+    )
     def transaction(self):
         """The current active or inactive :class:`.SessionTransaction`.
 
-        If this session is in "autobegin" mode and the transaction was not
-        begun, this accessor will implicitly begin the transaction.
+        May be None if no transaction has begun yet.
 
         .. versionchanged:: 1.4  the :attr:`.Session.transaction` attribute
-           is now a read-only descriptor that will automatically start a
-           transaction in "autobegin" mode if one is not present.
+           is now a read-only descriptor that also may return None if no
+           transaction has begun yet.
+
 
         """
-        self._autobegin()
+        if not self.future:
+            self._autobegin()
         return self._transaction
 
+    def in_transaction(self):
+        """Return True if this :class:`_orm.Session` has begun a transaction.
+
+        .. versionadded:: 1.4
+
+        .. seealso::
+
+            :attr:`_orm.Session.is_active`
+
+
+        """
+        return self._transaction is not None
+
+    def in_nested_transaction(self):
+        """Return True if this :class:`_orm.Session` has begun a nested
+        transaction, e.g. SAVEPOINT.
+
+        .. versionadded:: 1.4
+
+        """
+        return self._nested_transaction is not None
+
+    def get_transaction(self):
+        """Return the current root transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+        trans = self._transaction
+        while trans is not None and trans._parent is not None:
+            trans = trans._parent
+        return trans
+
+    def get_nested_transaction(self):
+        """Return the current nested transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+
+        return self._nested_transaction
+
     @util.memoized_property
     def info(self):
         """A user-modifiable dictionary.
@@ -1141,8 +1153,6 @@ class Session(_SessionClassMethods):
         here is always local to this :class:`.Session` and can be modified
         independently of all other :class:`.Session` objects.
 
-        .. versionadded:: 0.9.0
-
         """
         return {}
 
@@ -1153,7 +1163,17 @@ class Session(_SessionClassMethods):
 
         return False
 
-    def begin(self, subtransactions=False, nested=False):
+    @util.deprecated_params(
+        subtransactions=(
+            "2.0",
+            "The :paramref:`_orm.Session.begin.subtransactions` flag is "
+            "deprecated and "
+            "will be removed in SQLAlchemy version 2.0.  The "
+            ":attr:`_orm.Session.transaction` flag may "
+            "be checked for None before invoking :meth:`_orm.Session.begin`.",
+        )
+    )
+    def begin(self, subtransactions=False, nested=False, _subtrans=False):
         """Begin a transaction on this :class:`.Session`.
 
         .. warning::
@@ -1206,17 +1226,24 @@ class Session(_SessionClassMethods):
 
         """
 
+        if subtransactions and self.future:
+            raise NotImplementedError(
+                "subtransactions are not implemented in future "
+                "Session objects."
+            )
         if self._autobegin():
             if not subtransactions and not nested:
-                return
+                return self._transaction
 
         if self._transaction is not None:
-            if subtransactions or nested:
-                self._transaction = self._transaction._begin(nested=nested)
+            if subtransactions or _subtrans or nested:
+                trans = self._transaction._begin(nested=nested)
+                self._transaction = trans
+                if nested:
+                    self._nested_transaction = trans
             else:
                 raise sa_exc.InvalidRequestError(
-                    "A transaction is already begun.  Use "
-                    "subtransactions=True to allow subtransactions."
+                    "A transaction is already begun on this Session."
                 )
         else:
             self._transaction = SessionTransaction(self, nested=nested)
@@ -1265,7 +1292,7 @@ class Session(_SessionClassMethods):
         if self._transaction is None:
             pass
         else:
-            self._transaction.rollback()
+            self._transaction.rollback(_to_root=self.future)
 
     def commit(self):
         """Flush pending changes and commit the current transaction.
@@ -1299,7 +1326,7 @@ class Session(_SessionClassMethods):
             if not self._autobegin():
                 raise sa_exc.InvalidRequestError("No transaction is begun.")
 
-        self._transaction.commit()
+        self._transaction.commit(_to_root=self.future)
 
     def prepare(self):
         """Prepare the current transaction in progress for two phase commit.
@@ -1371,8 +1398,6 @@ class Session(_SessionClassMethods):
          present within the :class:`.Session`, a warning is emitted and
          the arguments are ignored.
 
-         .. versionadded:: 0.9.9
-
          .. seealso::
 
             :ref:`session_transaction_isolation`
@@ -1402,6 +1427,7 @@ class Session(_SessionClassMethods):
             )
 
         assert self._transaction is None
+        assert self.autocommit
         conn = engine.connect(**kw)
         if execution_options:
             conn = conn.execution_options(**execution_options)
@@ -1663,12 +1689,19 @@ class Session(_SessionClassMethods):
 
         This is a variant of :meth:`.Session.close` that will additionally
         ensure that the :meth:`_engine.Connection.invalidate`
-        method will be called
-        on all :class:`_engine.Connection` objects.  This can be called when
-        the database is known to be in a state where the connections are
-        no longer safe to be used.
+        method will be called on each :class:`_engine.Connection` object
+        that is currently in use for a transaction (typically there is only
+        one connection unless the :class:`_orm.Session` is used with
+        multiple engines).
 
-        E.g.::
+        This can be called when the database is known to be in a state where
+        the connections are no longer safe to be used.
+
+        Below illustrates a scenario when using `gevent
+        <http://www.gevent.org/>`_, which can produce ``Timeout`` exceptions
+        that may mean the underlying connection should be discarded::
+
+            import gevent
 
             try:
                 sess = Session()
@@ -1681,13 +1714,8 @@ class Session(_SessionClassMethods):
                 sess.rollback()
                 raise
 
-        This clears all items and ends any transaction in progress.
-
-        If this session were created with ``autocommit=False``, a new
-        transaction is immediately begun.  Note that this new transaction does
-        not use any connection resources until they are first needed.
-
-        .. versionadded:: 0.9.9
+        The method additionally does everything that :meth:`_orm.Session.close`
+        does, including that all ORM objects are expunged.
 
         """
         self._close_impl(invalidate=True)
@@ -2118,13 +2146,7 @@ class Session(_SessionClassMethods):
                 "A blank dictionary is ambiguous."
             )
 
-        if with_for_update is not None:
-            if with_for_update is True:
-                with_for_update = query.ForUpdateArg()
-            elif with_for_update:
-                with_for_update = query.ForUpdateArg(**with_for_update)
-            else:
-                with_for_update = None
+        with_for_update = query.ForUpdateArg._from_argument(with_for_update)
 
         stmt = future.select(object_mapper(instance))
         if (
@@ -2482,6 +2504,200 @@ class Session(_SessionClassMethods):
             for o, m, st_, dct_ in cascade_states:
                 self._delete_impl(st_, o, False)
 
+    def get(
+        self,
+        entity,
+        ident,
+        options=None,
+        populate_existing=False,
+        with_for_update=None,
+        identity_token=None,
+    ):
+        """Return an instance based on the given primary key identifier,
+        or ``None`` if not found.
+
+        E.g.::
+
+            my_user = session.get(User, 5)
+
+            some_object = session.get(VersionedFoo, (5, 10))
+
+            some_object = session.get(
+                VersionedFoo,
+                {"id": 5, "version_id": 10}
+            )
+
+        .. versionadded:: 1.4 Added :meth:`_orm.Session.get`, which is moved
+           from the now deprecated :meth:`_orm.Query.get` method.
+
+        :meth:`_orm.Session.get` is special in that it provides direct
+        access to the identity map of the :class:`.Session`.
+        If the given primary key identifier is present
+        in the local identity map, the object is returned
+        directly from this collection and no SQL is emitted,
+        unless the object has been marked fully expired.
+        If not present,
+        a SELECT is performed in order to locate the object.
+
+        :meth:`_orm.Session.get` also will perform a check if
+        the object is present in the identity map and
+        marked as expired - a SELECT
+        is emitted to refresh the object as well as to
+        ensure that the row is still present.
+        If not, :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is raised.
+
+        :param entity: a mapped class or :class:`.Mapper` indicating the
+         type of entity to be loaded.
+
+        :param ident: A scalar, tuple, or dictionary representing the
+         primary key.  For a composite (e.g. multiple column) primary key,
+         a tuple or dictionary should be passed.
+
+         For a single-column primary key, the scalar calling form is typically
+         the most expedient.  If the primary key of a row is the value "5",
+         the call looks like::
+
+            my_object = session.get(SomeClass, 5)
+
+         The tuple form contains primary key values typically in
+         the order in which they correspond to the mapped
+         :class:`_schema.Table`
+         object's primary key columns, or if the
+         :paramref:`_orm.Mapper.primary_key` configuration parameter were
+         used, in
+         the order used for that parameter. For example, if the primary key
+         of a row is represented by the integer
+         digits "5, 10" the call would look like::
+
+             my_object = session.get(SomeClass, (5, 10))
+
+         The dictionary form should include as keys the mapped attribute names
+         corresponding to each element of the primary key.  If the mapped class
+         has the attributes ``id``, ``version_id`` as the attributes which
+         store the object's primary key value, the call would look like::
+
+            my_object = session.get(SomeClass, {"id": 5, "version_id": 10})
+
+        :param options: optional sequence of loader options which will be
+         applied to the query, if one is emitted.
+
+        :param populate_existing: causes the method to unconditionally emit
+         a SQL query and refresh the object with the newly loaded data,
+         regardless of whether or not the object is already present.
+
+        :param with_for_update: optional boolean ``True`` indicating FOR UPDATE
+          should be used, or may be a dictionary containing flags to
+          indicate a more specific set of FOR UPDATE flags for the SELECT;
+          flags should match the parameters of
+          :meth:`_query.Query.with_for_update`.
+          Supersedes the :paramref:`.Session.refresh.lockmode` parameter.
+
+        :return: The object instance, or ``None``.
+
+        """
+        return self._get_impl(
+            entity,
+            ident,
+            loading.load_on_pk_identity,
+            options,
+            populate_existing=populate_existing,
+            with_for_update=with_for_update,
+            identity_token=identity_token,
+        )
+
+    def _get_impl(
+        self,
+        entity,
+        primary_key_identity,
+        db_load_fn,
+        options=None,
+        populate_existing=False,
+        with_for_update=None,
+        identity_token=None,
+        execution_options=None,
+    ):
+
+        # convert composite types to individual args
+        if hasattr(primary_key_identity, "__composite_values__"):
+            primary_key_identity = primary_key_identity.__composite_values__()
+
+        mapper = inspect(entity)
+
+        is_dict = isinstance(primary_key_identity, dict)
+        if not is_dict:
+            primary_key_identity = util.to_list(
+                primary_key_identity, default=(None,)
+            )
+
+        if len(primary_key_identity) != len(mapper.primary_key):
+            raise sa_exc.InvalidRequestError(
+                "Incorrect number of values in identifier to formulate "
+                "primary key for query.get(); primary key columns are %s"
+                % ",".join("'%s'" % c for c in mapper.primary_key)
+            )
+
+        if is_dict:
+            try:
+                primary_key_identity = list(
+                    primary_key_identity[prop.key]
+                    for prop in mapper._identity_key_props
+                )
+
+            except KeyError as err:
+                util.raise_(
+                    sa_exc.InvalidRequestError(
+                        "Incorrect names of values in identifier to formulate "
+                        "primary key for query.get(); primary key attribute "
+                        "names are %s"
+                        % ",".join(
+                            "'%s'" % prop.key
+                            for prop in mapper._identity_key_props
+                        )
+                    ),
+                    replace_context=err,
+                )
+
+        if (
+            not populate_existing
+            and not mapper.always_refresh
+            and with_for_update is None
+        ):
+
+            instance = self._identity_lookup(
+                mapper, primary_key_identity, identity_token=identity_token
+            )
+
+            if instance is not None:
+                # reject calls for id in identity map but class
+                # mismatch.
+                if not issubclass(instance.__class__, mapper.class_):
+                    return None
+                return instance
+            elif instance is attributes.PASSIVE_CLASS_MISMATCH:
+                return None
+
+        # apply_labels() not strictly necessary, however this will ensure that
+        # tablename_colname style is used which at the moment is asserted
+        # in a lot of unit tests :)
+
+        load_options = context.QueryContext.default_load_options
+
+        if populate_existing:
+            load_options += {"_populate_existing": populate_existing}
+        statement = sql.select(mapper).apply_labels()
+        if with_for_update is not None:
+            statement._for_update_arg = query.ForUpdateArg._from_argument(
+                with_for_update
+            )
+
+        if options:
+            statement = statement.options(*options)
+        if execution_options:
+            statement = statement.execution_options(**execution_options)
+        return db_load_fn(
+            self, statement, primary_key_identity, load_options=load_options,
+        )
+
     def merge(self, instance, load=True):
         """Copy the state of a given instance into a corresponding instance
         within this :class:`.Session`.
@@ -2629,7 +2845,7 @@ class Session(_SessionClassMethods):
                 new_instance = True
 
             elif key_is_persistent:
-                merged = self.query(mapper.class_).get(key[1])
+                merged = self.get(mapper.class_, key[1], identity_token=key[2])
 
         if merged is None:
             merged = mapper.class_manager.new_instance()
@@ -3021,9 +3237,7 @@ class Session(_SessionClassMethods):
         if not flush_context.has_work:
             return
 
-        flush_context.transaction = transaction = self.begin(
-            subtransactions=True
-        )
+        flush_context.transaction = transaction = self.begin(_subtrans=True)
         try:
             self._warn_on_events = True
             try:
@@ -3338,7 +3552,7 @@ class Session(_SessionClassMethods):
         mapper = _class_to_mapper(mapper)
         self._flushing = True
 
-        transaction = self.begin(subtransactions=True)
+        transaction = self.begin(_subtrans=True)
         try:
             if isupdate:
                 persistence._bulk_update(
@@ -3441,62 +3655,38 @@ class Session(_SessionClassMethods):
 
     @property
     def is_active(self):
-        """True if this :class:`.Session` is in "transaction mode" and
-        is not in "partial rollback" state.
-
-        The :class:`.Session` in its default mode of ``autocommit=False``
-        is essentially always in "transaction mode", in that a
-        :class:`.SessionTransaction` is associated with it as soon as
-        it is instantiated.  This :class:`.SessionTransaction` is immediately
-        replaced with a new one as soon as it is ended, due to a rollback,
-        commit, or close operation.
-
-        "Transaction mode" does *not* indicate whether
-        or not actual database connection resources are in use;  the
-        :class:`.SessionTransaction` object coordinates among zero or more
-        actual database transactions, and starts out with none, accumulating
-        individual DBAPI connections as different data sources are used
-        within its scope.   The best way to track when a particular
-        :class:`.Session` has actually begun to use DBAPI resources is to
-        implement a listener using the :meth:`.SessionEvents.after_begin`
-        method, which will deliver both the :class:`.Session` as well as the
-        target :class:`_engine.Connection` to a user-defined event listener.
-
-        The "partial rollback" state refers to when an "inner" transaction,
-        typically used during a flush, encounters an error and emits a
-        rollback of the DBAPI connection.  At this point, the
-        :class:`.Session` is in "partial rollback" and awaits for the user to
-        call :meth:`.Session.rollback`, in order to close out the
-        transaction stack.  It is in this "partial rollback" period that the
-        :attr:`.is_active` flag returns False.  After the call to
-        :meth:`.Session.rollback`, the :class:`.SessionTransaction` is
-        replaced with a new one and :attr:`.is_active` returns ``True`` again.
-
-        When a :class:`.Session` is used in ``autocommit=True`` mode, the
-        :class:`.SessionTransaction` is only instantiated within the scope
-        of a flush call, or when :meth:`.Session.begin` is called.  So
-        :attr:`.is_active` will always be ``False`` outside of a flush or
-        :meth:`.Session.begin` block in this mode, and will be ``True``
-        within the :meth:`.Session.begin` block as long as it doesn't enter
-        "partial rollback" state.
-
-        From all the above, it follows that the only purpose to this flag is
-        for application frameworks that wish to detect if a "rollback" is
-        necessary within a generic error handling routine, for
-        :class:`.Session` objects that would otherwise be in
-        "partial rollback" mode.  In a typical integration case, this is also
-        not necessary as it is standard practice to emit
-        :meth:`.Session.rollback` unconditionally within the outermost
-        exception catch.
-
-        To track the transactional state of a :class:`.Session` fully,
-        use event listeners, primarily the :meth:`.SessionEvents.after_begin`,
-        :meth:`.SessionEvents.after_commit`,
-        :meth:`.SessionEvents.after_rollback` and related events.
+        """True if this :class:`.Session` not in "partial rollback" state.
+
+        .. versionchanged:: 1.4 The :class:`_orm.Session` no longer begins
+           a new transaction immediately, so this attribute will be False
+           when the :class:`_orm.Session` is first instantiated.
+
+        "partial rollback" state typically indicates that the flush process
+        of the :class:`_orm.Session` has failed, and that the
+        :meth:`_orm.Session.rollback` method must be emitted in order to
+        fully roll back the transaction.
+
+        If this :class:`_orm.Session` is not in a transaction at all, the
+        :class:`_orm.Session` will autobegin when it is first used, so in this
+        case :attr:`_orm.Session.is_active` will return True.
+
+        Otherwise, if this :class:`_orm.Session` is within a transaction,
+        and that transaction has not been rolled back internally, the
+        :attr:`_orm.Session.is_active` will also return True.
+
+        .. seealso::
+
+            :ref:`faq_session_rollback`
+
+            :meth:`_orm.Session.in_transaction`
 
         """
-        self._autobegin()
-        return self._transaction and self._transaction.is_active
+        if self.autocommit:
+            return (
+                self._transaction is not None and self._transaction.is_active
+            )
+        else:
+            return self._transaction is None or self._transaction.is_active
 
     identity_map = None
     """A mapping of object identities to objects themselves.
@@ -3576,36 +3766,84 @@ class sessionmaker(_SessionClassMethods):
 
     e.g.::
 
-        # global scope
-        Session = sessionmaker(autoflush=False)
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
 
-        # later, in a local scope, create and use a session:
-        sess = Session()
+        # an Engine, which the Session will use for connection
+        # resources
+        engine = create_engine('postgresql://scott:tiger@localhost/')
 
-    Any keyword arguments sent to the constructor itself will override the
-    "configured" keywords::
+        Session = sessionmaker(engine)
 
-        Session = sessionmaker()
+        with Session() as session:
+            session.add(some_object)
+            session.add(some_other_object)
+            session.commit()
+
+    Context manager use is optional; otherwise, the returned
+    :class:`_orm.Session` object may be closed explicitly via the
+    :meth:`_orm.Session.close` method.   Using a
+    ``try:/finally:`` block is optional, however will ensure that the close
+    takes place even if there are database errors::
+
+        session = Session()
+        try:
+            session.add(some_object)
+            session.add(some_other_object)
+            session.commit()
+        finally:
+            session.close()
+
+    :class:`.sessionmaker` acts as a factory for :class:`_orm.Session`
+    objects in the same way as an :class:`_engine.Engine` acts as a factory
+    for :class:`_engine.Connection` objects.  In this way it also includes
+    a :meth:`_orm.sessionmaker.begin` method, that provides a context
+    manager which both begins and commits a transaction, as well as closes
+    out the :class:`_orm.Session` when complete, rolling back the transaction
+    if any errors occur::
+
+        Session = sessionmaker(engine)
+
+        wih Session.begin() as session:
+            session.add(some_object)
+            session.add(some_other_object)
+        # commits transaction, closes session
+
+    .. versionadded:: 1.4
+
+    When calling upon :class:`_orm.sessionmaker` to construct a
+    :class:`_orm.Session`, keyword arguments may also be passed to the
+    method; these arguments will override that of the globally configured
+    parameters.  Below we use a :class:`_orm.sessionmaker` bound to a certain
+    :class:`_engine.Engine` to produce a :class:`_orm.Session` that is instead
+    bound to a specific :class:`_engine.Connection` procured from that engine::
+
+        Session = sessionmaker(engine)
 
         # bind an individual session to a connection
-        sess = Session(bind=connection)
 
-    The class also includes a method :meth:`.configure`, which can
-    be used to specify additional keyword arguments to the factory, which
-    will take effect for subsequent :class:`.Session` objects generated.
-    This is usually used to associate one or more :class:`_engine.Engine`
-    objects
-    with an existing :class:`.sessionmaker` factory before it is first
-    used::
+        with engine.connect() as connection:
+            with Session(bind=connection) as session:
+                # work with session
+
+    The class also includes a method :meth:`_orm.sessionmaker.configure`, which
+    can be used to specify additional keyword arguments to the factory, which
+    will take effect for subsequent :class:`.Session` objects generated. This
+    is usually used to associate one or more :class:`_engine.Engine` objects
+    with an existing
+    :class:`.sessionmaker` factory before it is first used::
 
-        # application starts
+        # application starts, sessionmaker does not have
+        # an engine bound yet
         Session = sessionmaker()
 
-        # ... later
+        # ... later, when an engine URL is read from a configuration
+        # file or other events allow the engine to be created
         engine = create_engine('sqlite:///foo.db')
         Session.configure(bind=engine)
 
         sess = Session()
+        # work with session
 
     .. seealso::
 
@@ -3646,8 +3884,6 @@ class sessionmaker(_SessionClassMethods):
          replaced, when the ``info`` parameter is specified to the specific
          :class:`.Session` construction operation.
 
-         .. versionadded:: 0.9.0
-
         :param \**kw: all other keyword arguments are passed to the
          constructor of newly created :class:`.Session` objects.
 
@@ -3663,6 +3899,29 @@ class sessionmaker(_SessionClassMethods):
         # events can be associated with it specifically.
         self.class_ = type(class_.__name__, (class_,), {})
 
+    @util.contextmanager
+    def begin(self):
+        """Produce a context manager that both provides a new
+        :class:`_orm.Session` as well as a transaction that commits.
+
+
+        e.g.::
+
+            Session = sessionmaker(some_engine)
+
+            with Session.begin() as session:
+                session.add(some_object)
+
+            # commits transaction, closes session
+
+        .. versionadded:: 1.4
+
+
+        """
+        with self() as session:
+            with session.begin():
+                yield session
+
     def __call__(self, **local_kw):
         """Produce a new :class:`.Session` object using the configuration
         established in this :class:`.sessionmaker`.
@@ -3806,8 +4065,6 @@ def make_transient_to_detached(instance):
     call to :meth:`.Session.merge` in that a given persistent state
     can be manufactured without any SQL calls.
 
-    .. versionadded:: 0.9.5
-
     .. seealso::
 
         :func:`.make_transient`
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 53cc99ccd..db82f0b74 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -677,7 +677,7 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
                         self._equated_columns[c] = self._equated_columns[col]
 
             self.logger.info(
-                "%s will use query.get() to " "optimize instance loads", self
+                "%s will use Session.get() to " "optimize instance loads", self
             )
 
     def init_class_attribute(self, mapper):
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index e04c54497..68ffa2393 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -1621,3 +1621,108 @@ def randomize_unitofwork():
     topological.set = (
         unitofwork.set
     ) = session.set = mapper.set = dependency.set = RandomSet
+
+
+def _offset_or_limit_clause(element, name=None, type_=None):
+    """Convert the given value to an "offset or limit" clause.
+
+    This handles incoming integers and converts to an expression; if
+    an expression is already given, it is passed through.
+
+    """
+    return coercions.expect(
+        roles.LimitOffsetRole, element, name=name, type_=type_
+    )
+
+
+def _offset_or_limit_clause_asint_if_possible(clause):
+    """Return the offset or limit clause as a simple integer if possible,
+    else return the clause.
+
+    """
+    if clause is None:
+        return None
+    if hasattr(clause, "_limit_offset_value"):
+        value = clause._limit_offset_value
+        return util.asint(value)
+    else:
+        return clause
+
+
+def _make_slice(limit_clause, offset_clause, start, stop):
+    """Compute LIMIT/OFFSET in terms of slice start/end
+    """
+
+    # for calculated limit/offset, try to do the addition of
+    # values to offset in Python, however if a SQL clause is present
+    # then the addition has to be on the SQL side.
+    if start is not None and stop is not None:
+        offset_clause = _offset_or_limit_clause_asint_if_possible(
+            offset_clause
+        )
+        if offset_clause is None:
+            offset_clause = 0
+
+        if start != 0:
+            offset_clause = offset_clause + start
+
+        if offset_clause == 0:
+            offset_clause = None
+        else:
+            offset_clause = _offset_or_limit_clause(offset_clause)
+
+        limit_clause = _offset_or_limit_clause(stop - start)
+
+    elif start is None and stop is not None:
+        limit_clause = _offset_or_limit_clause(stop)
+    elif start is not None and stop is None:
+        offset_clause = _offset_or_limit_clause_asint_if_possible(
+            offset_clause
+        )
+        if offset_clause is None:
+            offset_clause = 0
+
+        if start != 0:
+            offset_clause = offset_clause + start
+
+        if offset_clause == 0:
+            offset_clause = None
+        else:
+            offset_clause = _offset_or_limit_clause(offset_clause)
+
+    return limit_clause, offset_clause
+
+
+def _getitem(iterable_query, item):
+    """calculate __getitem__ in terms of an iterable query object
+    that also has a slice() method.
+
+    """
+
+    if isinstance(item, slice):
+        start, stop, step = util.decode_slice(item)
+
+        if (
+            isinstance(stop, int)
+            and isinstance(start, int)
+            and stop - start <= 0
+        ):
+            return []
+
+        # perhaps we should execute a count() here so that we
+        # can still use LIMIT/OFFSET ?
+        elif (isinstance(start, int) and start < 0) or (
+            isinstance(stop, int) and stop < 0
+        ):
+            return list(iterable_query)[item]
+
+        res = iterable_query.slice(start, stop)
+        if step is not None:
+            return list(res)[None : None : item.step]
+        else:
+            return list(res)
+    else:
+        if item == -1:
+            return list(iterable_query)[-1]
+        else:
+            return list(iterable_query[item : item + 1])[0]
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 4bc6d8280..36a8151d3 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -571,6 +571,26 @@ class Options(util.with_metaclass(_MetaOptions)):
         o1.__dict__.update(other)
         return o1
 
+    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):
+            if getattr(self, a) != getattr(other, b):
+                return False
+        return True
+
+    def __repr__(self):
+        # TODO: fairly inefficient, used only in debugging right now.
+
+        return "%s(%s)" % (
+            self.__class__.__name__,
+            ", ".join(
+                "%s=%r" % (k, self.__dict__[k])
+                for k in self._cache_attrs
+                if k in self.__dict__
+            ),
+        )
+
     @hybridmethod
     def add_to_element(self, name, value):
         return self + {name: getattr(self, name) + value}
@@ -610,6 +630,60 @@ class Options(util.with_metaclass(_MetaOptions)):
             )
         return cls + d
 
+    @classmethod
+    def from_execution_options(
+        cls, key, attrs, exec_options, statement_exec_options
+    ):
+        """"process Options argument in terms of execution options.
+
+
+        e.g.::
+
+            (
+                load_options,
+                execution_options,
+            ) = QueryContext.default_load_options.from_execution_options(
+                "_sa_orm_load_options",
+                {
+                    "populate_existing",
+                    "autoflush",
+                    "yield_per"
+                },
+                execution_options,
+                statement._execution_options,
+            )
+
+        get back the Options and refresh "_sa_orm_load_options" in the
+        exec options dict w/ the Options as well
+
+        """
+
+        # common case is that no options we are looking for are
+        # in either dictionary, so cancel for that first
+        check_argnames = attrs.intersection(
+            set(exec_options).union(statement_exec_options)
+        )
+
+        existing_options = exec_options.get(key, cls)
+
+        if check_argnames:
+            result = {}
+            for argname in check_argnames:
+                local = "_" + argname
+                if argname in exec_options:
+                    result[local] = exec_options[argname]
+                elif argname in statement_exec_options:
+                    result[local] = statement_exec_options[argname]
+
+            new_options = existing_options + result
+            exec_options = util.immutabledict().merge_with(
+                exec_options, {key: new_options}
+            )
+            return new_options, exec_options
+
+        else:
+            return existing_options, exec_options
+
 
 class CacheableOptions(Options, HasCacheKey):
     @hybridmethod
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index be412c770..588c485ae 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -52,6 +52,18 @@ def _document_text_coercion(paramname, meth_rst, param_rst):
     )
 
 
+def _expression_collection_was_a_list(attrname, fnname, args):
+    if args and isinstance(args[0], (list, set)) and len(args) == 1:
+        util.warn_deprecated_20(
+            'The "%s" argument to %s() is now passed as a series of '
+            "positional "
+            "elements, rather than as a list. " % (attrname, fnname)
+        )
+        return args[0]
+    else:
+        return args
+
+
 def expect(role, element, apply_propagate_attrs=None, argname=None, **kw):
     if (
         role.allows_lambda
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 6ce505412..c7e5aabcc 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -2573,10 +2573,8 @@ class Case(ColumnElement):
         stmt = select([users_table]).\
                     where(
                         case(
-                            [
-                                (users_table.c.name == 'wendy', 'W'),
-                                (users_table.c.name == 'jack', 'J')
-                            ],
+                            (users_table.c.name == 'wendy', 'W'),
+                            (users_table.c.name == 'jack', 'J'),
                             else_='E'
                         )
                     )
@@ -2597,7 +2595,10 @@ class Case(ColumnElement):
         ("else_", InternalTraversal.dp_clauseelement),
     ]
 
-    def __init__(self, whens, value=None, else_=None):
+    # TODO: for Py2k removal, this will be:
+    # def __init__(self, *whens, value=None, else_=None):
+
+    def __init__(self, *whens, **kw):
         r"""Produce a ``CASE`` expression.
 
         The ``CASE`` construct in SQL is a conditional object that
@@ -2612,10 +2613,8 @@ class Case(ColumnElement):
             stmt = select([users_table]).\
                         where(
                             case(
-                                [
-                                    (users_table.c.name == 'wendy', 'W'),
-                                    (users_table.c.name == 'jack', 'J')
-                                ],
+                                (users_table.c.name == 'wendy', 'W'),
+                                (users_table.c.name == 'jack', 'J'),
                                 else_='E'
                             )
                         )
@@ -2660,16 +2659,14 @@ class Case(ColumnElement):
             from sqlalchemy import case, literal_column
 
             case(
-                [
-                    (
-                        orderline.c.qty > 100,
-                        literal_column("'greaterthan100'")
-                    ),
-                    (
-                        orderline.c.qty > 10,
-                        literal_column("'greaterthan10'")
-                    )
-                ],
+                (
+                    orderline.c.qty > 100,
+                    literal_column("'greaterthan100'")
+                ),
+                (
+                    orderline.c.qty > 10,
+                    literal_column("'greaterthan10'")
+                ),
                 else_=literal_column("'lessthan10'")
             )
 
@@ -2683,19 +2680,23 @@ class Case(ColumnElement):
                 ELSE 'lessthan10'
             END
 
-        :param whens: The criteria to be compared against,
+        :param \*whens: The criteria to be compared against,
          :paramref:`.case.whens` accepts two different forms, based on
          whether or not :paramref:`.case.value` is used.
 
+         .. versionchanged:: 1.4 the :func:`_sql.case`
+            function now accepts the series of WHEN conditions positionally;
+            passing the expressions within a list is deprecated.
+
          In the first form, it accepts a list of 2-tuples; each 2-tuple
          consists of ``(<sql expression>, <value>)``, where the SQL
          expression is a boolean expression and "value" is a resulting value,
          e.g.::
 
-            case([
+            case(
                 (users_table.c.name == 'wendy', 'W'),
                 (users_table.c.name == 'jack', 'J')
-            ])
+            )
 
          In the second form, it accepts a Python dictionary of comparison
          values mapped to a resulting value; this form requires
@@ -2720,11 +2721,23 @@ class Case(ColumnElement):
 
         """
 
+        if "whens" in kw:
+            util.warn_deprecated_20(
+                'The "whens" argument to case() is now passed as a series of '
+                "positional "
+                "elements, rather than as a list. "
+            )
+            whens = kw.pop("whens")
+        else:
+            whens = coercions._expression_collection_was_a_list(
+                "whens", "case", whens
+            )
         try:
             whens = util.dictlike_iteritems(whens)
         except TypeError:
             pass
 
+        value = kw.pop("value", None)
         if value is not None:
             whenlist = [
                 (
@@ -2760,11 +2773,16 @@ class Case(ColumnElement):
 
         self.type = type_
         self.whens = whenlist
+
+        else_ = kw.pop("else_", None)
         if else_ is not None:
             self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
         else:
             self.else_ = None
 
+        if kw:
+            raise TypeError("unknown arguments: %s" % (", ".join(sorted(kw))))
+
     @property
     def _from_objects(self):
         return list(
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index 85db88345..2d369cdf8 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -169,9 +169,6 @@ class Operators(object):
           :class:`.Boolean`, and those that do not will be of the same
           type as the left-hand operand.
 
-          .. versionadded:: 1.2.0b3 - added the
-             :paramref:`.Operators.op.return_type` argument.
-
         .. seealso::
 
             :ref:`types_operators`
@@ -194,8 +191,6 @@ class Operators(object):
         :paramref:`.Operators.op.is_comparison`
         flag with True.
 
-        .. versionadded:: 1.2.0b3
-
         .. seealso::
 
             :meth:`.Operators.op`
@@ -723,15 +718,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.startswith.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.startswith.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
@@ -811,15 +797,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.endswith.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.endswith.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
@@ -899,15 +876,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.contains.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.contains.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 12fcc00c3..1155c273b 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2225,6 +2225,17 @@ class ForUpdateArg(ClauseElement):
         ("skip_locked", InternalTraversal.dp_boolean),
     ]
 
+    @classmethod
+    def _from_argument(cls, with_for_update):
+        if isinstance(with_for_update, ForUpdateArg):
+            return with_for_update
+        elif with_for_update in (None, False):
+            return None
+        elif with_for_update is True:
+            return ForUpdateArg()
+        else:
+            return ForUpdateArg(**with_for_update)
+
     def __eq__(self, other):
         return (
             isinstance(other, ForUpdateArg)
@@ -2699,6 +2710,12 @@ class SelectStatementGrouping(GroupedElement, SelectBase):
 
 
 class DeprecatedSelectBaseGenerations(object):
+    """A collection of methods available on :class:`_sql.Select` and
+    :class:`_sql.CompoundSelect`, these are all **deprecated** methods as they
+    modify the object in-place.
+
+    """
+
     @util.deprecated(
         "1.4",
         "The :meth:`_expression.GenerativeSelect.append_order_by` "
@@ -2740,9 +2757,6 @@ class DeprecatedSelectBaseGenerations(object):
         as it
         provides standard :term:`method chaining`.
 
-        .. seealso::
-
-            :meth:`_expression.GenerativeSelect.group_by`
 
         """
         self.group_by.non_generative(self, *clauses)
@@ -3353,6 +3367,12 @@ class CompoundSelect(HasCompileState, GenerativeSelect):
 
 
 class DeprecatedSelectGenerations(object):
+    """A collection of methods available on :class:`_sql.Select`, these
+    are all **deprecated** methods as they modify the :class:`_sql.Select`
+    object in -place.
+
+    """
+
     @util.deprecated(
         "1.4",
         "The :meth:`_expression.Select.append_correlation` "
@@ -3377,7 +3397,7 @@ class DeprecatedSelectGenerations(object):
         "1.4",
         "The :meth:`_expression.Select.append_column` method is deprecated "
         "and will be removed in a future release.  Use the generative "
-        "method :meth:`_expression.Select.column`.",
+        "method :meth:`_expression.Select.add_columns`.",
     )
     def append_column(self, column):
         """Append the given column expression to the columns clause of this
@@ -3388,14 +3408,10 @@ class DeprecatedSelectGenerations(object):
             my_select.append_column(some_table.c.new_column)
 
         This is an **in-place** mutation method; the
-        :meth:`_expression.Select.column` method is preferred,
+        :meth:`_expression.Select.add_columns` method is preferred,
         as it provides standard
         :term:`method chaining`.
 
-        See the documentation for :meth:`_expression.Select.with_only_columns`
-        for guidelines on adding /replacing the columns of a
-        :class:`_expression.Select` object.
-
         """
         self.add_columns.non_generative(self, column)
 
@@ -3501,6 +3517,21 @@ class SelectState(util.MemoizedSlots, CompileState):
 
         self.columns_plus_names = statement._generate_columns_plus_names(True)
 
+    @classmethod
+    def _plugin_not_implemented(cls):
+        raise NotImplementedError(
+            "The default SELECT construct without plugins does not "
+            "implement this method."
+        )
+
+    @classmethod
+    def get_column_descriptions(cls, statement):
+        cls._plugin_not_implemented()
+
+    @classmethod
+    def from_statement(cls, statement, from_statement):
+        cls._plugin_not_implemented()
+
     def _get_froms(self, statement):
         seen = set()
         froms = []
@@ -3805,6 +3836,15 @@ class Select(
 ):
     """Represents a ``SELECT`` statement.
 
+    The :class:`_sql.Select` object is normally constructed using the
+    :func:`_sql.select` function.  See that function for details.
+
+    .. seealso::
+
+        :func:`_sql.select`
+
+        :ref:`coretutorial_selecting` - in the Core tutorial
+
     """
 
     __visit_name__ = "select"
@@ -3821,7 +3861,7 @@ class Select(
     _from_obj = ()
     _auto_correlate = True
 
-    compile_options = SelectState.default_select_compile_options
+    _compile_options = SelectState.default_select_compile_options
 
     _traverse_internals = (
         [
@@ -3851,7 +3891,7 @@ class Select(
     )
 
     _cache_key_traversal = _traverse_internals + [
-        ("compile_options", InternalTraversal.dp_has_cache_key)
+        ("_compile_options", InternalTraversal.dp_has_cache_key)
     ]
 
     @classmethod
@@ -4274,12 +4314,35 @@ class Select(
     @property
     def column_descriptions(self):
         """Return a 'column descriptions' structure which may be
-        plugin-specific.
+        :term:`plugin-specific`.
 
         """
         meth = SelectState.get_plugin_class(self).get_column_descriptions
         return meth(self)
 
+    def from_statement(self, statement):
+        """Apply the columns which this :class:`.Select` would select
+        onto another statement.
+
+        This operation is :term:`plugin-specific` and will raise a not
+        supported exception if this :class:`_sql.Select` does not select from
+        plugin-enabled entities.
+
+
+        The statement is typically either a :func:`_expression.text` or
+        :func:`_expression.select` construct, and should return the set of
+        columns appropriate to the entities represented by this
+        :class:`.Select`.
+
+        .. seealso::
+
+            :ref:`orm_tutorial_literal_sql` - usage examples in the
+            ORM tutorial
+
+        """
+        meth = SelectState.get_plugin_class(self).from_statement
+        return meth(self, statement)
+
     @_generative
     def join(self, target, onclause=None, isouter=False, full=False):
         r"""Create a SQL JOIN against this :class:`_expresson.Select`
@@ -4550,7 +4613,7 @@ class Select(
         )
 
     @_generative
-    def with_only_columns(self, columns):
+    def with_only_columns(self, *columns):
         r"""Return a new :func:`_expression.select` construct with its columns
         clause replaced with the given columns.
 
@@ -4558,65 +4621,26 @@ class Select(
         :func:`_expression.select` had been called with the given columns
         clause.   I.e. a statement::
 
-            s = select([table1.c.a, table1.c.b])
-            s = s.with_only_columns([table1.c.b])
+            s = select(table1.c.a, table1.c.b)
+            s = s.with_only_columns(table1.c.b)
 
         should be exactly equivalent to::
 
-            s = select([table1.c.b])
-
-        This means that FROM clauses which are only derived
-        from the column list will be discarded if the new column
-        list no longer contains that FROM::
-
-            >>> table1 = table('t1', column('a'), column('b'))
-            >>> table2 = table('t2', column('a'), column('b'))
-            >>> s1 = select([table1.c.a, table2.c.b])
-            >>> print(s1)
-            SELECT t1.a, t2.b FROM t1, t2
-            >>> s2 = s1.with_only_columns([table2.c.b])
-            >>> print(s2)
-            SELECT t2.b FROM t1
-
-        The preferred way to maintain a specific FROM clause
-        in the construct, assuming it won't be represented anywhere
-        else (i.e. not in the WHERE clause, etc.) is to set it using
-        :meth:`_expression.Select.select_from`::
-
-            >>> s1 = select([table1.c.a, table2.c.b]).\
-            ...         select_from(table1.join(table2,
-            ...                 table1.c.a==table2.c.a))
-            >>> s2 = s1.with_only_columns([table2.c.b])
-            >>> print(s2)
-            SELECT t2.b FROM t1 JOIN t2 ON t1.a=t2.a
-
-        Care should also be taken to use the correct set of column objects
-        passed to :meth:`_expression.Select.with_only_columns`.
-        Since the method is
-        essentially equivalent to calling the :func:`_expression.select`
-        construct in the first place with the given columns, the columns passed
-        to :meth:`_expression.Select.with_only_columns`
-        should usually be a subset of
-        those which were passed to the :func:`_expression.select`
-        construct, not those which are available from the ``.c`` collection of
-        that :func:`_expression.select`.  That is::
-
-            s = select([table1.c.a, table1.c.b]).select_from(table1)
-            s = s.with_only_columns([table1.c.b])
-
-        and **not**::
-
-            # usually incorrect
-            s = s.with_only_columns([s.c.b])
-
-        The latter would produce the SQL::
-
-            SELECT b
-            FROM (SELECT t1.a AS a, t1.b AS b
-            FROM t1), t1
-
-        Since the :func:`_expression.select` construct is essentially
-        being asked to select both from ``table1`` as well as itself.
+            s = select(table1.c.b)
+
+        Note that this will also dynamically alter the FROM clause of the
+        statement if it is not explicitly stated.  To maintain the FROM
+        clause, ensure the :meth:`_sql.Select.select_from` method is
+        used appropriately::
+
+            s = select(table1.c.a, table2.c.b)
+            s = s.select_from(table2.c.b).with_only_columns(table1.c.a)
+
+        :param \*columns: column expressions to be used.
+
+         .. versionchanged:: 1.4 the :meth:`_sql.Select.with_only_columns`
+            method accepts the list of column expressions positionally;
+            passing the expressions as a list is deprecateed.
 
         """
 
@@ -4626,7 +4650,9 @@ class Select(
         self._assert_no_memoizations()
 
         rc = []
-        for c in columns:
+        for c in coercions._expression_collection_was_a_list(
+            "columns", "Select.with_only_columns", columns
+        ):
             c = coercions.expect(roles.ColumnsClauseRole, c,)
             # TODO: why are we doing this here?
             if isinstance(c, ScalarSelect):
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index 1ce59431e..ecc6a4ab8 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -404,6 +404,9 @@ class AssertsCompiledSQL(object):
 
         from sqlalchemy import orm
 
+        if isinstance(clause, orm.dynamic.AppenderQuery):
+            clause = clause._statement
+
         if isinstance(clause, orm.Query):
             compile_state = clause._compile_state()
             compile_state.statement._label_style = (