- New improvements to the :func:`.text` construct, including

more flexible ways to set up bound parameters and return types;
in particular, a :func:`.text` can now be turned into a full
FROM-object, embeddable in other statements as an alias or CTE
using the new method :meth:`.TextClause.columns`.
[ticket:2877]

4 files changed, 482 insertions, 193 deletions

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 3ba3957d6..0c252089c 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -586,17 +586,10 @@ class SQLCompiler(Compiled):
         return text
 
     def visit_textclause(self, textclause, **kwargs):
-        if textclause.typemap is not None:
-            for colname, type_ in textclause.typemap.items():
-                self.result_map[colname
-                                if self.dialect.case_sensitive
-                                else colname.lower()] = \
-                                (colname, None, type_)
-
         def do_bindparam(m):
             name = m.group(1)
-            if name in textclause.bindparams:
-                return self.process(textclause.bindparams[name])
+            if name in textclause._bindparams:
+                return self.process(textclause._bindparams[name])
             else:
                 return self.bindparam_string(name, **kwargs)
 
@@ -606,6 +599,33 @@ class SQLCompiler(Compiled):
              self.post_process_text(textclause.text))
         )
 
+    def visit_text_as_from(self, taf, iswrapper=False,
+                                compound_index=0, force_result_map=False,
+                                asfrom=False,
+                                parens=True, **kw):
+
+        toplevel = not self.stack
+        entry = self._default_stack_entry if toplevel else self.stack[-1]
+
+        populate_result_map = force_result_map or (
+                                        compound_index == 0 and (
+                                            toplevel or \
+                                            entry['iswrapper']
+                                        )
+                                    )
+
+        if populate_result_map:
+            for c in taf.c:
+                self._add_to_result_map(
+                        c.key, c.key, (c,), c.type
+                )
+
+        text = self.process(taf.element, **kw)
+        if asfrom and parens:
+            text = "(%s)" % text
+        return text
+
+
     def visit_null(self, expr, **kw):
         return 'NULL'
 
@@ -726,6 +746,7 @@ class SQLCompiler(Compiled):
     def function_argspec(self, func, **kwargs):
         return func.clause_expr._compiler_dispatch(self, **kwargs)
 
+
     def visit_compound_select(self, cs, asfrom=False,
                             parens=True, compound_index=0, **kwargs):
         toplevel = not self.stack
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index f349923ae..adf51a425 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -18,6 +18,8 @@ from .visitors import Visitable, cloned_traverse, traverse
 from .annotation import Annotated
 import itertools
 from .base import Executable, PARSE_AUTOCOMMIT, Immutable, NO_ARG
+from .base import _generative, Generative
+
 import re
 import operator
 
@@ -805,6 +807,17 @@ class BindParameter(ColumnElement):
         else:
             self.type = type_
 
+    def _with_value(self, value):
+        """Return a copy of this :class:`.BindParameter` with the given value set."""
+        cloned = self._clone()
+        cloned.value = value
+        cloned.callable = None
+        cloned.required = False
+        if cloned.type is type_api.NULLTYPE:
+            cloned.type = type_api._type_map.get(type(value),
+                        type_api.NULLTYPE)
+        return cloned
+
     @property
     def effective_value(self):
         """Return the value of this bound parameter,
@@ -896,12 +909,24 @@ class TextClause(Executable, ClauseElement):
 
     def __init__(
                     self,
-                    text='',
-                    bind=None,
-                    bindparams=None,
-                    typemap=None,
-                    autocommit=None):
-        """Construct a new :class:`.TextClause` clause.
+                    text,
+                    bind=None):
+        self._bind = bind
+        self._bindparams = {}
+
+        def repl(m):
+            self._bindparams[m.group(1)] = BindParameter(m.group(1))
+            return ':%s' % m.group(1)
+
+        # scan the string and search for bind parameter names, add them
+        # to the list of bindparams
+        self.text = self._bind_params_regex.sub(repl, text)
+
+    @classmethod
+    def _create_text(self, text, bind=None, bindparams=None,
+                                    typemap=None, autocommit=None):
+        """Construct a new :class:`.TextClause` clause, representing
+        a textual SQL string directly.
 
         E.g.::
 
@@ -915,7 +940,9 @@ class TextClause(Executable, ClauseElement):
         execution options, as well as
         bind parameter and result-column typing behavior, allowing
         SQLAlchemy type constructs to play a role when executing
-        a statement that is specified literally.
+        a statement that is specified literally.  The construct can also
+        be provided with a ``.c`` collection of column elements, allowing
+        it to be embedded in other SQL expression constructs as a subquery.
 
         Bind parameters are specified by name, using the format ``:name``.
         E.g.::
@@ -923,48 +950,47 @@ class TextClause(Executable, ClauseElement):
             t = text("SELECT * FROM users WHERE id=:user_id")
             result = connection.execute(t, user_id=12)
 
-        To invoke SQLAlchemy typing logic for bind parameters, the
-        ``bindparams`` list allows specification of :func:`bindparam`
-        constructs which specify the type for a given name::
+        For SQL statements where a colon is required verbatim, as within
+        an inline string, use a backslash to escape::
 
-            t = text("SELECT id FROM users WHERE updated_at>:updated",
-                        bindparams=[bindparam('updated', DateTime())]
-                    )
+            t = text("SELECT * FROM users WHERE name='\\:username'")
 
-        Typing during result row processing is also an important concern.
-        Result column types
-        are specified using the ``typemap`` dictionary, where the keys
-        match the names of columns.  These names are taken from what
-        the DBAPI returns as ``cursor.description``::
-
-            t = text("SELECT id, name FROM users",
-                    typemap={
-                        'id':Integer,
-                        'name':Unicode
-                    }
-            )
+        The :class:`.TextClause` construct includes methods which can
+        provide information about the bound parameters as well as the column
+        values which would be returned from the textual statement, assuming
+        it's an executable SELECT type of statement.  The :meth:`.TextClause.bindparams`
+        method is used to provide bound parameter detail, and
+        :meth:`.TextClause.columns` method allows specification of
+        return columns including names and types::
+
+            t = text("SELECT * FROM users WHERE id=:user_id").\\
+                    bindparams(user_id=7).\\
+                    columns(id=Integer, name=String)
 
-        The :func:`text` construct is used internally for most cases when
+            for id, name in connection.execute(t):
+                print(id, name)
+
+        The :func:`.text` construct is used internally in cases when
         a literal string is specified for part of a larger query, such as
-        within :func:`select()`, :func:`update()`,
-        :func:`insert()` or :func:`delete()`.   In those cases, the same
+        when a string is specified to the :meth:`.Select.where` method of
+        :class:`.Select`.  In those cases, the same
         bind parameter syntax is applied::
 
             s = select([users.c.id, users.c.name]).where("id=:user_id")
             result = connection.execute(s, user_id=12)
 
-        Using :func:`text` explicitly usually implies the construction
+        Using :func:`.text` explicitly usually implies the construction
         of a full, standalone statement.   As such, SQLAlchemy refers
         to it as an :class:`.Executable` object, and it supports
         the :meth:`Executable.execution_options` method.  For example,
-        a :func:`text` construct that should be subject to "autocommit"
+        a :func:`.text` construct that should be subject to "autocommit"
         can be set explicitly so using the ``autocommit`` option::
 
             t = text("EXEC my_procedural_thing()").\\
                     execution_options(autocommit=True)
 
         Note that SQLAlchemy's usual "autocommit" behavior applies to
-        :func:`text` constructs - that is, statements which begin
+        :func:`.text` constructs implicitly - that is, statements which begin
         with a phrase such as ``INSERT``, ``UPDATE``, ``DELETE``,
         or a variety of other phrases specific to certain backends, will
         be eligible for autocommit if no transaction is in progress.
@@ -982,53 +1008,228 @@ class TextClause(Executable, ClauseElement):
           an optional connection or engine to be used for this text query.
 
         :param bindparams:
-          a list of :func:`bindparam()` instances which can be used to define
-          the types and/or initial values for the bind parameters within
-          the textual statement; the keynames of the bindparams must match
-          those within the text of the statement.  The types will be used
-          for pre-processing on bind values.
+          Deprecated.  A list of :func:`.bindparam` instances used to
+          provide information about parameters embedded in the statement.
+          This argument now invokes the :meth:`.TextClause.bindparams`
+          method on the construct before returning it.  E.g.::
+
+              stmt = text("SELECT * FROM table WHERE id=:id",
+                        bindparams=[bindparam('id', value=5, type_=Integer)])
+
+          Is equivalent to::
+
+              stmt = text("SELECT * FROM table WHERE id=:id").\\
+                        bindparams(bindparam('id', value=5, type_=Integer))
+
+          .. deprecated:: 0.9.0 the :meth:`.TextClause.bindparams` method
+             supersedes the ``bindparams`` argument to :func:`.text`.
 
         :param typemap:
-          a dictionary mapping the names of columns represented in the
-          columns clause of a ``SELECT`` statement  to type objects,
+          Deprecated.  A dictionary mapping the names of columns
+          represented in the columns clause of a ``SELECT`` statement
+          to type objects,
           which will be used to perform post-processing on columns within
-          the result set.   This argument applies to any expression
-          that returns result sets.
+          the result set.  This parameter now invokes the :meth:`.TextClause.columns`
+          method, which returns a :class:`.TextAsFrom` construct that gains
+          a ``.c`` collection and can be embedded in other expressions.  E.g.::
 
-        """
+              stmt = text("SELECT * FROM table",
+                            typemap={'id': Integer, 'name': String},
+                        )
 
-        self._bind = bind
-        self.bindparams = {}
-        self.typemap = typemap
+          Is equivalent to::
+
+              stmt = text("SELECT * FROM table").columns(id=Integer, name=String)
+
+          Or alternatively::
+
+              from sqlalchemy.sql import column
+              stmt = text("SELECT * FROM table").columns(
+                                    column('id', Integer),
+                                    column('name', String)
+                                )
+
+          .. deprecated:: 0.9.0 the :meth:`.TextClause.columns` method
+             supersedes the ``typemap`` argument to :func:`.text`.
+
+        """
+        stmt = TextClause(text, bind=bind)
+        if bindparams:
+            stmt = stmt.bindparams(*bindparams)
+        if typemap:
+            stmt = stmt.columns(**typemap)
         if autocommit is not None:
             util.warn_deprecated('autocommit on text() is deprecated.  '
-                                 'Use .execution_options(autocommit=Tru'
-                                 'e)')
-            self._execution_options = \
-                self._execution_options.union(
-                    {'autocommit': autocommit})
-        if typemap is not None:
-            for key in typemap:
-                typemap[key] = type_api.to_instance(typemap[key])
+                                 'Use .execution_options(autocommit=True)')
+            stmt = stmt.execution_options(autocommit=autocommit)
 
-        def repl(m):
-            self.bindparams[m.group(1)] = BindParameter(m.group(1))
-            return ':%s' % m.group(1)
+        return stmt
 
-        # scan the string and search for bind parameter names, add them
-        # to the list of bindparams
+    @_generative
+    def bindparams(self, *binds, **names_to_values):
+        """Establish the values and/or types of bound parameters within
+        this :class:`.TextClause` construct.
 
-        self.text = self._bind_params_regex.sub(repl, text)
-        if bindparams is not None:
-            for b in bindparams:
-                self.bindparams[b.key] = b
+        Given a text construct such as::
+
+            from sqlalchemy import text
+            stmt = text("SELECT id, name FROM user WHERE name=:name "
+                        "AND timestamp=:timestamp")
+
+        the :meth:`.TextClause.bindparams` method can be used to establish
+        the initial value of ``:name`` and ``:timestamp``,
+        using simple keyword arguments::
+
+            stmt = stmt.bindparams(name='jack',
+                        timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5))
+
+        Where above, new :class:`.BindParameter` objects
+        will be generated with the names ``name`` and ``timestamp``, and
+        values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
+        respectively.  The types will be
+        inferred from the values given, in this case :class:`.String` and
+        :class:`.DateTime`.
+
+        When specific typing behavior is needed, the positional ``*binds``
+        argument can be used in which to specify :func:`.bindparam` constructs
+        directly.  These constructs must include at least the ``key`` argument,
+        then an optional value and type::
+
+            from sqlalchemy import bindparam
+            stmt = stmt.bindparams(
+                            bindparam('name', value='jack', type_=String),
+                            bindparam('timestamp', type_=DateTime)
+                        )
+
+        Above, we specified the type of :class:`.DateTime` for the ``timestamp``
+        bind, and the type of :class:`.String` for the ``name`` bind.  In
+        the case of ``name`` we also set the default value of ``"jack"``.
+
+        Additional bound parameters can be supplied at statement execution
+        time, e.g.::
+
+            result = connection.execute(stmt,
+                        timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5))
+
+        The :meth:`.TextClause.bindparams` method can be called repeatedly, where
+        it will re-use existing :class:`.BindParameter` objects to add new information.
+        For example, we can call :meth:`.TextClause.bindparams` first with
+        typing information, and a second time with value information, and it
+        will be combined::
+
+            stmt = text("SELECT id, name FROM user WHERE name=:name "
+                        "AND timestamp=:timestamp")
+            stmt = stmt.bindparams(
+                            bindparam('name', type_=String),
+                            bindparam('timestamp', type_=DateTime)
+                        )
+            stmt = stmt.bindparams(
+                            name='jack',
+                            timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
+                        )
+
+
+        .. versionadded:: 0.9.0 The :meth:`.TextClause.bindparams` method supersedes
+           the argument ``bindparams`` passed to :func:`~.expression.text`.
+
+
+        """
+        self._bindparams = new_params = self._bindparams.copy()
+
+        for bind in binds:
+            try:
+                existing = new_params[bind.key]
+            except KeyError:
+                raise exc.ArgumentError(
+                            "This text() construct doesn't define a "
+                            "bound parameter named %r" % bind.key)
+            else:
+                new_params[existing.key] = bind
+
+        for key, value in names_to_values.items():
+            try:
+                existing = new_params[key]
+            except KeyError:
+                raise exc.ArgumentError(
+                            "This text() construct doesn't define a "
+                            "bound parameter named %r" % key)
+            else:
+                new_params[key] = existing._with_value(value)
+
+
+
+    @util.dependencies('sqlalchemy.sql.selectable')
+    def columns(self, selectable, *cols, **types):
+        """Turn this :class:`.Text` object into a :class:`.FromClause`
+        object that can be embedded into another statement.
+
+        This function essentially bridges the gap between an entirely
+        textual SELECT statement and the SQL expression language concept
+        of a "selectable"::
+
+            from sqlalchemy.sql import column, text
+
+            stmt = text("SELECT id, name FROM some_table")
+            stmt = stmt.columns(column('id'), column('name')).alias('st')
+
+            stmt = select([mytable]).\\
+                    select_from(
+                        mytable.join(stmt, mytable.c.name == stmt.c.name)
+                    ).where(stmt.c.id > 5)
+
+        Above, we used untyped :func:`.column` elements.  These can also have
+        types specified, which will impact how the column behaves in expressions
+        as well as determining result set behavior::
+
+            stmt = text("SELECT id, name, timestamp FROM some_table")
+            stmt = stmt.columns(
+                        column('id', Integer),
+                        column('name', Unicode),
+                        column('timestamp', DateTime)
+                    )
+
+            for id, name, timestamp in connection.execute(stmt):
+                print(id, name, timestamp)
+
+        Keyword arguments allow just the names and types of columns to be specified,
+        where the :func:`.column` elements will be generated automatically::
+
+            stmt = text("SELECT id, name, timestamp FROM some_table")
+            stmt = stmt.columns(
+                        id=Integer,
+                        name=Unicode,
+                        timestamp=DateTime
+                    )
+
+            for id, name, timestamp in connection.execute(stmt):
+                print(id, name, timestamp)
+
+        The :meth:`.TextClause.columns` method provides a direct
+        route to calling :meth:`.FromClause.alias` as well as :meth:`.SelectBase.cte`
+        against a textual SELECT statement::
+
+            stmt = stmt.columns(id=Integer, name=String).cte('st')
+
+            stmt = select([sometable]).where(sometable.c.id == stmt.c.id)
+
+        .. versionadded:: 0.9.0 :func:`.text` can now be converted into a fully
+           featured "selectable" construct using the :meth:`.TextClause.columns`
+           method.  This method supersedes the ``typemap`` argument to
+           :func:`.text`.
+
+        """
+
+        col_by_name = dict(
+            (col.key, col) for col in cols
+        )
+        for key, type_ in types.items():
+            col_by_name[key] = ColumnClause(key, type_)
+
+        return selectable.TextAsFrom(self, col_by_name.values())
 
     @property
     def type(self):
-        if self.typemap is not None and len(self.typemap) == 1:
-            return list(self.typemap)[0]
-        else:
-            return type_api.NULLTYPE
+        return type_api.NULLTYPE
 
     @property
     def comparator(self):
@@ -1041,11 +1242,11 @@ class TextClause(Executable, ClauseElement):
             return self
 
     def _copy_internals(self, clone=_clone, **kw):
-        self.bindparams = dict((b.key, clone(b, **kw))
-                               for b in self.bindparams.values())
+        self._bindparams = dict((b.key, clone(b, **kw))
+                               for b in self._bindparams.values())
 
     def get_children(self, **kwargs):
-        return list(self.bindparams.values())
+        return list(self._bindparams.values())
 
 
 class Null(ColumnElement):
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index d1f019482..26a025ddb 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -46,7 +46,7 @@ from .base import ColumnCollection, Generative, Executable, \
 from .selectable import Alias, Join, Select, Selectable, TableClause, \
         CompoundSelect, CTE, FromClause, FromGrouping, SelectBase, \
         alias, \
-        subquery, HasPrefixes, Exists, ScalarSelect
+        subquery, HasPrefixes, Exists, ScalarSelect, TextAsFrom
 
 
 from .dml import Insert, Update, Delete, UpdateBase, ValuesBase
@@ -60,7 +60,7 @@ and_ = public_factory(BooleanClauseList.and_, ".expression.and_")
 or_ = public_factory(BooleanClauseList.or_, ".expression.or_")
 bindparam = public_factory(BindParameter, ".expression.bindparam")
 select = public_factory(Select, ".expression.select")
-text = public_factory(TextClause, ".expression.tet")
+text = public_factory(TextClause._create_text, ".expression.text")
 table = public_factory(TableClause, ".expression.table")
 column = public_factory(ColumnClause, ".expression.column")
 over = public_factory(Over, ".expression.over")
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 28c757a66..9fb99a4cd 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1219,100 +1219,16 @@ class ForUpdateArg(ClauseElement):
         else:
             self.of = None
 
-class SelectBase(Executable, FromClause):
-    """Base class for :class:`.Select` and :class:`.CompoundSelect`."""
-
-    _order_by_clause = ClauseList()
-    _group_by_clause = ClauseList()
-    _limit = None
-    _offset = None
-    _for_update_arg = None
-
-    def __init__(self,
-            use_labels=False,
-            for_update=False,
-            limit=None,
-            offset=None,
-            order_by=None,
-            group_by=None,
-            bind=None,
-            autocommit=None):
-        self.use_labels = use_labels
-
-        if for_update is not False:
-            self._for_update_arg = ForUpdateArg.parse_legacy_select(for_update)
-
-        if autocommit is not None:
-            util.warn_deprecated('autocommit on select() is '
-                                 'deprecated.  Use .execution_options(a'
-                                 'utocommit=True)')
-            self._execution_options = \
-                self._execution_options.union(
-                  {'autocommit': autocommit})
-        if limit is not None:
-            self._limit = util.asint(limit)
-        if offset is not None:
-            self._offset = util.asint(offset)
-        self._bind = bind
-
-        if order_by is not None:
-            self._order_by_clause = ClauseList(*util.to_list(order_by))
-        if group_by is not None:
-            self._group_by_clause = ClauseList(*util.to_list(group_by))
-
-    @property
-    def for_update(self):
-        """Provide legacy dialect support for the ``for_update`` attribute.
-        """
-        if self._for_update_arg is not None:
-            return self._for_update_arg.legacy_for_update_value
-        else:
-            return None
-
-    @for_update.setter
-    def for_update(self, value):
-        self._for_update_arg = ForUpdateArg.parse_legacy_select(value)
 
-    @_generative
-    def with_for_update(self, nowait=False, read=False, of=None):
-        """Specify a ``FOR UPDATE`` clause for this :class:`.SelectBase`.
-
-        E.g.::
-
-            stmt = select([table]).with_for_update(nowait=True)
-
-        On a database like Postgresql or Oracle, the above would render a
-        statement like::
-
-            SELECT table.a, table.b FROM table FOR UPDATE NOWAIT
-
-        on other backends, the ``nowait`` option is ignored and instead
-        would produce::
-
-            SELECT table.a, table.b FROM table FOR UPDATE
-
-        When called with no arguments, the statement will render with
-        the suffix ``FOR UPDATE``.   Additional arguments can then be
-        provided which allow for common database-specific
-        variants.
-
-        :param nowait: boolean; will render ``FOR UPDATE NOWAIT`` on Oracle and
-         Postgresql dialects.
+class SelectBase(Executable, FromClause):
+    """Base class for SELECT statements.
 
-        :param read: boolean; will render ``LOCK IN SHARE MODE`` on MySQL,
-         ``FOR SHARE`` on Postgresql.  On Postgresql, when combined with
-         ``nowait``, will render ``FOR SHARE NOWAIT``.
 
-        :param of: SQL expression or list of SQL expression elements
-         (typically :class:`.Column` objects or a compatible expression) which
-         will render into a ``FOR UPDATE OF`` clause; supported by PostgreSQL
-         and Oracle.  May render as a table or as a column depending on
-         backend.
+    This includes :class:`.Select`, :class:`.CompoundSelect` and
+    :class:`.TextAsFrom`.
 
-        .. versionadded:: 0.9.0b2
 
-        """
-        self._for_update_arg = ForUpdateArg(nowait=nowait, read=read, of=of)
+    """
 
     def as_scalar(self):
         """return a 'scalar' representation of this selectable, which can be
@@ -1327,18 +1243,6 @@ class SelectBase(Executable, FromClause):
         """
         return ScalarSelect(self)
 
-    @_generative
-    def apply_labels(self):
-        """return a new selectable with the 'use_labels' flag set to True.
-
-        This will result in column expressions being generated using labels
-        against their table name, such as "SELECT somecolumn AS
-        tablename_somecolumn". This allows selectables which contain multiple
-        FROM clauses to produce a unique set of column names regardless of
-        name conflicts among the individual FROM clauses.
-
-        """
-        self.use_labels = True
 
     def label(self, name):
         """return a 'scalar' representation of this selectable, embedded as a
@@ -1493,6 +1397,132 @@ class SelectBase(Executable, FromClause):
         s._reset_exported()
         return s
 
+    @property
+    def _from_objects(self):
+        return [self]
+
+class GenerativeSelect(SelectBase):
+    """Base class for SELECT statements where additional elements can be
+    added.
+
+    This serves as the base for :class:`.Select` and :class:`.CompoundSelect`
+    where elements such as ORDER BY, GROUP BY can be added and column rendering
+    can be controlled.  Compare to :class:`.TextAsFrom`, which, while it
+    subclasses :class:`.SelectBase` and is also a SELECT construct, represents
+    a fixed textual string which cannot be altered at this level, only
+    wrapped as a subquery.
+
+    .. versionadded:: 0.9.0b2 :class:`.GenerativeSelect` was added to
+       provide functionality specific to :class:`.Select` and :class:`.CompoundSelect`
+       while allowing :class:`.SelectBase` to be used for other SELECT-like
+       objects, e.g. :class:`.TextAsFrom`.
+
+    """
+    _order_by_clause = ClauseList()
+    _group_by_clause = ClauseList()
+    _limit = None
+    _offset = None
+    _for_update_arg = None
+
+    def __init__(self,
+            use_labels=False,
+            for_update=False,
+            limit=None,
+            offset=None,
+            order_by=None,
+            group_by=None,
+            bind=None,
+            autocommit=None):
+        self.use_labels = use_labels
+
+        if for_update is not False:
+            self._for_update_arg = ForUpdateArg.parse_legacy_select(for_update)
+
+        if autocommit is not None:
+            util.warn_deprecated('autocommit on select() is '
+                                 'deprecated.  Use .execution_options(a'
+                                 'utocommit=True)')
+            self._execution_options = \
+                self._execution_options.union(
+                  {'autocommit': autocommit})
+        if limit is not None:
+            self._limit = util.asint(limit)
+        if offset is not None:
+            self._offset = util.asint(offset)
+        self._bind = bind
+
+        if order_by is not None:
+            self._order_by_clause = ClauseList(*util.to_list(order_by))
+        if group_by is not None:
+            self._group_by_clause = ClauseList(*util.to_list(group_by))
+
+    @property
+    def for_update(self):
+        """Provide legacy dialect support for the ``for_update`` attribute.
+        """
+        if self._for_update_arg is not None:
+            return self._for_update_arg.legacy_for_update_value
+        else:
+            return None
+
+    @for_update.setter
+    def for_update(self, value):
+        self._for_update_arg = ForUpdateArg.parse_legacy_select(value)
+
+    @_generative
+    def with_for_update(self, nowait=False, read=False, of=None):
+        """Specify a ``FOR UPDATE`` clause for this :class:`.GenerativeSelect`.
+
+        E.g.::
+
+            stmt = select([table]).with_for_update(nowait=True)
+
+        On a database like Postgresql or Oracle, the above would render a
+        statement like::
+
+            SELECT table.a, table.b FROM table FOR UPDATE NOWAIT
+
+        on other backends, the ``nowait`` option is ignored and instead
+        would produce::
+
+            SELECT table.a, table.b FROM table FOR UPDATE
+
+        When called with no arguments, the statement will render with
+        the suffix ``FOR UPDATE``.   Additional arguments can then be
+        provided which allow for common database-specific
+        variants.
+
+        :param nowait: boolean; will render ``FOR UPDATE NOWAIT`` on Oracle and
+         Postgresql dialects.
+
+        :param read: boolean; will render ``LOCK IN SHARE MODE`` on MySQL,
+         ``FOR SHARE`` on Postgresql.  On Postgresql, when combined with
+         ``nowait``, will render ``FOR SHARE NOWAIT``.
+
+        :param of: SQL expression or list of SQL expression elements
+         (typically :class:`.Column` objects or a compatible expression) which
+         will render into a ``FOR UPDATE OF`` clause; supported by PostgreSQL
+         and Oracle.  May render as a table or as a column depending on
+         backend.
+
+        .. versionadded:: 0.9.0b2
+
+        """
+        self._for_update_arg = ForUpdateArg(nowait=nowait, read=read, of=of)
+
+    @_generative
+    def apply_labels(self):
+        """return a new selectable with the 'use_labels' flag set to True.
+
+        This will result in column expressions being generated using labels
+        against their table name, such as "SELECT somecolumn AS
+        tablename_somecolumn". This allows selectables which contain multiple
+        FROM clauses to produce a unique set of column names regardless of
+        name conflicts among the individual FROM clauses.
+
+        """
+        self.use_labels = True
+
     @_generative
     def limit(self, limit):
         """return a new selectable with the given LIMIT criterion
@@ -1537,7 +1567,7 @@ class SelectBase(Executable, FromClause):
         The criterion will be appended to any pre-existing ORDER BY criterion.
 
         This is an **in-place** mutation method; the
-        :meth:`~.SelectBase.order_by` method is preferred, as it provides standard
+        :meth:`~.GenerativeSelect.order_by` method is preferred, as it provides standard
         :term:`method chaining`.
 
         """
@@ -1554,7 +1584,7 @@ class SelectBase(Executable, FromClause):
         The criterion will be appended to any pre-existing GROUP BY criterion.
 
         This is an **in-place** mutation method; the
-        :meth:`~.SelectBase.group_by` method is preferred, as it provides standard
+        :meth:`~.GenerativeSelect.group_by` method is preferred, as it provides standard
         :term:`method chaining`.
 
         """
@@ -1565,12 +1595,8 @@ class SelectBase(Executable, FromClause):
                 clauses = list(self._group_by_clause) + list(clauses)
             self._group_by_clause = ClauseList(*clauses)
 
-    @property
-    def _from_objects(self):
-        return [self]
-
 
-class CompoundSelect(SelectBase):
+class CompoundSelect(GenerativeSelect):
     """Forms the basis of ``UNION``, ``UNION ALL``, and other
         SELECT-based set operations.
 
@@ -1622,7 +1648,7 @@ class CompoundSelect(SelectBase):
 
             self.selects.append(s.self_group(self))
 
-        SelectBase.__init__(self, **kwargs)
+        GenerativeSelect.__init__(self, **kwargs)
 
     @classmethod
     def _create_union(cls, *selects, **kwargs):
@@ -1852,7 +1878,7 @@ class HasPrefixes(object):
 
 
 
-class Select(HasPrefixes, SelectBase):
+class Select(HasPrefixes, GenerativeSelect):
     """Represents a ``SELECT`` statement.
 
     """
@@ -1956,7 +1982,7 @@ class Select(HasPrefixes, SelectBase):
           when ``True``, applies ``FOR UPDATE`` to the end of the
           resulting statement.
 
-          .. deprecated:: 0.9.0 - use :meth:`.SelectBase.with_for_update`
+          .. deprecated:: 0.9.0 - use :meth:`.GenerativeSelect.with_for_update`
              to specify the structure of the ``FOR UPDATE`` clause.
 
           ``for_update`` accepts various string values interpreted by
@@ -1971,7 +1997,7 @@ class Select(HasPrefixes, SelectBase):
 
          .. seealso::
 
-            :meth:`.SelectBase.with_for_update` - improved API for
+            :meth:`.GenerativeSelect.with_for_update` - improved API for
             specifying the ``FOR UPDATE`` clause.
 
         :param group_by:
@@ -2007,7 +2033,7 @@ class Select(HasPrefixes, SelectBase):
           collection of the resulting :class:`.Select` object will use these
           names as well for targeting column members.
 
-          use_labels is also available via the :meth:`~.SelectBase.apply_labels`
+          use_labels is also available via the :meth:`~.GenerativeSelect.apply_labels`
           generative method.
 
         """
@@ -2057,7 +2083,7 @@ class Select(HasPrefixes, SelectBase):
         if prefixes:
             self._setup_prefixes(prefixes)
 
-        SelectBase.__init__(self, **kwargs)
+        GenerativeSelect.__init__(self, **kwargs)
 
     @property
     def _froms(self):
@@ -2912,6 +2938,47 @@ class Exists(UnaryExpression):
         return e
 
 
+class TextAsFrom(SelectBase):
+    """Wrap a :class:`.TextClause` construct within a :class:`.SelectBase`
+    interface.
+
+    This allows the :class:`.Text` object to gain a ``.c`` collection and
+    other FROM-like capabilities such as :meth:`.FromClause.alias`,
+    :meth:`.SelectBase.cte`, etc.
+
+    The :class:`.TextAsFrom` construct is produced via the
+    :meth:`.TextClause.columns` method - see that method for details.
+
+    .. versionadded:: 0.9.0b2
+
+    .. seealso::
+
+        :func:`.text`
+
+        :meth:`.TextClause.columns`
+
+    """
+    __visit_name__ = "text_as_from"
+
+    def __init__(self, text, columns):
+        self.element = text
+        self.column_args = columns
+
+    @property
+    def _bind(self):
+        return self.element._bind
+
+    def _populate_column_collection(self):
+        for c in self.column_args:
+            c._make_proxy(self)
+
+    def _copy_internals(self, clone=_clone, **kw):
+        self._reset_exported()
+        self.element = clone(self.element, **kw)
+
+    def _scalar_type(self):
+        return self.column_args[0].type
+
 class AnnotatedFromClause(Annotated):
     def __init__(self, element, values):
         # force FromClause to generate their internal