.. changelog::
:version: 0.9.0b2
+ .. change::
+ :tags: feature, sql
+ :tickets: 2877
+
+ 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`.
+
+ .. seealso::
+
+ :ref:`feature_2877`
+
.. change::
:tags: feature, sql
:pullreq: github:42
A new API for specifying the ``FOR UPDATE`` clause of a ``SELECT``
- is added with the new :meth:`.SelectBase.with_for_update` method.
+ is added with the new :meth:`.GenerativeSelect.with_for_update` method.
This method supports a more straightforward system of setting
dialect-specific options compared to the ``for_update`` keyword
argument of :func:`.select`, and also includes support for the
A new API for specifying the ``FOR UPDATE`` clause of a ``SELECT``
is added with the new :meth:`.Query.with_for_update` method,
- to complement the new :meth:`.SelectBase.with_for_update` method.
+ to complement the new :meth:`.GenerativeSelect.with_for_update` method.
Pull request courtesy Mario Lassnig.
.. seealso::
:ticket:`1418`
+.. _feature_2877:
+
+New ``text()`` Capabilities
+---------------------------
+
+The :func:`.text` construct gains new methods:
+
+* :meth:`.TextClause.bindparams` allows bound parameter types and values
+ to be set flexibly::
+
+ # setup values
+ stmt = text("SELECT id, name FROM user "
+ "WHERE name=:name AND timestamp=:timestamp").\
+ bindparams(name="ed", timestamp=datetime(2012, 11, 10, 15, 12, 35))
+
+ # setup types and/or values
+ stmt = text("SELECT id, name FROM user "
+ "WHERE name=:name AND timestamp=:timestamp").\
+ bindparams(
+ bindparam("name", value="ed"),
+ bindparam("timestamp", type_=DateTime()
+ ).bindparam(timestamp=datetime(2012, 11, 10, 15, 12, 35))
+
+* :meth:`.TextClause.columns` supersedes the ``typemap`` option
+ of :func:`.text`, returning a new construct :class:`.TextAsFrom`::
+
+ # turn a text() into an alias(), with a .c. collection:
+ stmt = text("SELECT id, name FROM user").columns(id=Integer, name=String)
+ stmt = stmt.alias()
+
+ stmt = select([addresses]).select_from(
+ addresses.join(stmt), addresses.c.user_id == stmt.c.id)
+
+
+ # or into a cte():
+ stmt = text("SELECT id, name FROM user").columns(id=Integer, name=String)
+ stmt = stmt.cte("x")
+
+ stmt = select([addresses]).select_from(
+ addresses.join(stmt), addresses.c.user_id == stmt.c.id)
+
+:ticket:`2877`
+
.. _feature_722:
INSERT from SELECT
clause on ``SELECT`` statements made within Core and ORM, and support is added
for the ``FOR UPDATE OF`` SQL supported by Postgresql and Oracle.
-Using the core :meth:`.SelectBase.with_for_update`, options like ``FOR SHARE`` and
+Using the core :meth:`.GenerativeSelect.with_for_update`, options like ``FOR SHARE`` and
``NOWAIT`` can be specified individually, rather than linking to arbitrary
string codes::
.. autoclass:: Executable
:members:
-
.. autoclass:: FromClause
:members:
+.. autoclass:: GenerativeSelect
+ :members:
+ :inherited-members:
+
.. autoclass:: HasPrefixes
:members:
.. autoclass:: Selectable
:members:
-
.. autoclass:: SelectBase
:members:
-
.. autoclass:: TableClause
:members:
:inherited-members:
+
+.. autoclass:: TextAsFrom
+ :members:
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)
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'
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
from .annotation import Annotated
import itertools
from .base import Executable, PARSE_AUTOCOMMIT, Immutable, NO_ARG
+from .base import _generative, Generative
+
import re
import operator
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,
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.::
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.::
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.
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):
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):
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
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")
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
"""
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
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
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`.
"""
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`.
"""
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.
self.selects.append(s.self_group(self))
- SelectBase.__init__(self, **kwargs)
+ GenerativeSelect.__init__(self, **kwargs)
@classmethod
def _create_union(cls, *selects, **kwargs):
-class Select(HasPrefixes, SelectBase):
+class Select(HasPrefixes, GenerativeSelect):
"""Represents a ``SELECT`` statement.
"""
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
.. 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:
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.
"""
if prefixes:
self._setup_prefixes(prefixes)
- SelectBase.__init__(self, **kwargs)
+ GenerativeSelect.__init__(self, **kwargs)
@property
def _froms(self):
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
dialect=mysql.dialect()
)
- def test_text(self):
- self.assert_compile(
- text("select * from foo where lala = bar"),
- "select * from foo where lala = bar"
- )
-
- # test bytestring
- self.assert_compile(select(
- ["foobar(a)", "pk_foo_bar(syslaal)"],
- "a = 12",
- from_obj=["foobar left outer join lala on foobar.foo = lala.foo"]
- ),
- "SELECT foobar(a), pk_foo_bar(syslaal) FROM foobar "
- "left outer join lala on foobar.foo = lala.foo WHERE a = 12"
- )
-
- # test unicode
- self.assert_compile(select(
- ["foobar(a)", "pk_foo_bar(syslaal)"],
- "a = 12",
- from_obj=["foobar left outer join lala on foobar.foo = lala.foo"]
- ),
- "SELECT foobar(a), pk_foo_bar(syslaal) FROM foobar "
- "left outer join lala on foobar.foo = lala.foo WHERE a = 12"
- )
-
- # test building a select query programmatically with text
- s = select()
- s.append_column("column1")
- s.append_column("column2")
- s.append_whereclause("column1=12")
- s.append_whereclause("column2=19")
- s = s.order_by("column1")
- s.append_from("table1")
- self.assert_compile(s, "SELECT column1, column2 FROM table1 WHERE "
- "column1=12 AND column2=19 ORDER BY column1")
-
- self.assert_compile(
- select(["column1", "column2"],
- from_obj=table1).alias('somealias').select(),
- "SELECT somealias.column1, somealias.column2 FROM "
- "(SELECT column1, column2 FROM mytable) AS somealias"
- )
-
- # test that use_labels doesnt interfere with literal columns
- self.assert_compile(
- select(["column1", "column2", table1.c.myid], from_obj=table1,
- use_labels=True),
- "SELECT column1, column2, mytable.myid AS mytable_myid "
- "FROM mytable"
- )
-
- # test that use_labels doesnt interfere
- # with literal columns that have textual labels
- self.assert_compile(
- select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
- from_obj=table1, use_labels=True),
- "SELECT column1 AS foobar, column2 AS hoho, "
- "mytable.myid AS mytable_myid FROM mytable"
- )
-
- # test that "auto-labeling of subquery columns"
- # doesnt interfere with literal columns,
- # exported columns dont get quoted
- self.assert_compile(
- select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
- from_obj=[table1]).select(),
- "SELECT column1 AS foobar, column2 AS hoho, myid FROM "
- "(SELECT column1 AS foobar, column2 AS hoho, "
- "mytable.myid AS myid FROM mytable)"
- )
-
- self.assert_compile(
- select(['col1', 'col2'], from_obj='tablename').alias('myalias'),
- "SELECT col1, col2 FROM tablename"
- )
-
- def test_binds_in_text(self):
- self.assert_compile(
- text("select * from foo where lala=:bar and hoho=:whee",
- bindparams=[bindparam('bar', 4), bindparam('whee', 7)]),
- "select * from foo where lala=:bar and hoho=:whee",
- checkparams={'bar': 4, 'whee': 7},
- )
-
- self.assert_compile(
- text("select * from foo where clock='05:06:07'"),
- "select * from foo where clock='05:06:07'",
- checkparams={},
- params={},
- )
-
- dialect = postgresql.dialect()
- self.assert_compile(
- text("select * from foo where lala=:bar and hoho=:whee",
- bindparams=[bindparam('bar', 4), bindparam('whee', 7)]),
- "select * from foo where lala=%(bar)s and hoho=%(whee)s",
- checkparams={'bar': 4, 'whee': 7},
- dialect=dialect
- )
-
- # test escaping out text() params with a backslash
- self.assert_compile(
- text("select * from foo where clock='05:06:07' "
- "and mork='\:mindy'"),
- "select * from foo where clock='05:06:07' and mork=':mindy'",
- checkparams={},
- params={},
- dialect=dialect
- )
-
- dialect = sqlite.dialect()
- self.assert_compile(
- text("select * from foo where lala=:bar and hoho=:whee",
- bindparams=[bindparam('bar', 4), bindparam('whee', 7)]),
- "select * from foo where lala=? and hoho=?",
- checkparams={'bar': 4, 'whee': 7},
- dialect=dialect
- )
-
- self.assert_compile(select(
- [table1, table2.c.otherid, "sysdate()", "foo, bar, lala"],
- and_(
- "foo.id = foofoo(lala)",
- "datetime(foo) = Today",
- table1.c.myid == table2.c.otherid,
- )
- ),
- "SELECT mytable.myid, mytable.name, mytable.description, "
- "myothertable.otherid, sysdate(), foo, bar, lala "
- "FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND "
- "datetime(foo) = Today AND mytable.myid = myothertable.otherid")
-
- self.assert_compile(select(
- [alias(table1, 't'), "foo.f"],
- "foo.f = t.id",
- from_obj=["(select f from bar where lala=heyhey) foo"]
- ),
- "SELECT t.myid, t.name, t.description, foo.f FROM mytable AS t, "
- "(select f from bar where lala=heyhey) foo WHERE foo.f = t.id")
-
- # test Text embedded within select_from(), using binds
- generate_series = text(
- "generate_series(:x, :y, :z) as s(a)",
- bindparams=[bindparam('x', None),
- bindparam('y', None), bindparam('z', None)]
- )
-
- s = select([
- (func.current_date() +
- literal_column("s.a")).label("dates")
- ]).select_from(generate_series)
- self.assert_compile(
- s,
- "SELECT CURRENT_DATE + s.a AS dates FROM "
- "generate_series(:x, :y, :z) as s(a)",
- checkparams={'y': None, 'x': None, 'z': None}
- )
-
- self.assert_compile(
- s.params(x=5, y=6, z=7),
- "SELECT CURRENT_DATE + s.a AS dates FROM "
- "generate_series(:x, :y, :z) as s(a)",
- checkparams={'y': 6, 'x': 5, 'z': 7}
- )
-
@testing.emits_warning('.*empty sequence.*')
def test_render_binds_as_literal(self):
"""test a compiler that renders binds inline into
class Vis(CloningVisitor):
def visit_textclause(self, text):
text.text = text.text + " SOME MODIFIER=:lala"
- text.bindparams['lala'] = bindparam('lala')
+ text._bindparams['lala'] = bindparam('lala')
clause2 = Vis().traverse(clause)
assert c1 == str(clause)
assert str(clause2) == c1 + " SOME MODIFIER=:lala"
- assert list(clause.bindparams.keys()) == ['bar']
- assert set(clause2.bindparams.keys()) == set(['bar', 'lala'])
+ assert list(clause._bindparams.keys()) == ['bar']
+ assert set(clause2._bindparams.keys()) == set(['bar', 'lala'])
def test_select(self):
s2 = select([t1])
--- /dev/null
+"""Test the TextClause and related constructs."""
+
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL, eq_, assert_raises_message
+from sqlalchemy import text, select, Integer, String, Float, \
+ bindparam, and_, func, literal_column, exc
+from sqlalchemy.types import NullType
+from sqlalchemy.sql import table, column
+
+table1 = table('mytable',
+ column('myid', Integer),
+ column('name', String),
+ column('description', String),
+)
+
+table2 = table(
+ 'myothertable',
+ column('otherid', Integer),
+ column('othername', String),
+)
+
+class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_basic(self):
+ self.assert_compile(
+ text("select * from foo where lala = bar"),
+ "select * from foo where lala = bar"
+ )
+
+class SelectCompositionTest(fixtures.TestBase, AssertsCompiledSQL):
+ """test the usage of text() implicit within the select() construct
+ when strings are passed."""
+
+ __dialect__ = 'default'
+
+ def test_select_composition_one(self):
+ self.assert_compile(select(
+ ["foobar(a)", "pk_foo_bar(syslaal)"],
+ "a = 12",
+ from_obj=["foobar left outer join lala on foobar.foo = lala.foo"]
+ ),
+ "SELECT foobar(a), pk_foo_bar(syslaal) FROM foobar "
+ "left outer join lala on foobar.foo = lala.foo WHERE a = 12"
+ )
+
+ def test_select_composition_two(self):
+ s = select()
+ s.append_column("column1")
+ s.append_column("column2")
+ s.append_whereclause("column1=12")
+ s.append_whereclause("column2=19")
+ s = s.order_by("column1")
+ s.append_from("table1")
+ self.assert_compile(s, "SELECT column1, column2 FROM table1 WHERE "
+ "column1=12 AND column2=19 ORDER BY column1")
+
+ def test_select_composition_three(self):
+ self.assert_compile(
+ select(["column1", "column2"],
+ from_obj=table1).alias('somealias').select(),
+ "SELECT somealias.column1, somealias.column2 FROM "
+ "(SELECT column1, column2 FROM mytable) AS somealias"
+ )
+
+ def test_select_composition_four(self):
+ # test that use_labels doesnt interfere with literal columns
+ self.assert_compile(
+ select(["column1", "column2", table1.c.myid], from_obj=table1,
+ use_labels=True),
+ "SELECT column1, column2, mytable.myid AS mytable_myid "
+ "FROM mytable"
+ )
+
+ def test_select_composition_five(self):
+ # test that use_labels doesnt interfere
+ # with literal columns that have textual labels
+ self.assert_compile(
+ select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
+ from_obj=table1, use_labels=True),
+ "SELECT column1 AS foobar, column2 AS hoho, "
+ "mytable.myid AS mytable_myid FROM mytable"
+ )
+
+ def test_select_composition_six(self):
+ # test that "auto-labeling of subquery columns"
+ # doesnt interfere with literal columns,
+ # exported columns dont get quoted
+ self.assert_compile(
+ select(["column1 AS foobar", "column2 AS hoho", table1.c.myid],
+ from_obj=[table1]).select(),
+ "SELECT column1 AS foobar, column2 AS hoho, myid FROM "
+ "(SELECT column1 AS foobar, column2 AS hoho, "
+ "mytable.myid AS myid FROM mytable)"
+ )
+
+ def test_select_composition_seven(self):
+ self.assert_compile(
+ select(['col1', 'col2'], from_obj='tablename').alias('myalias'),
+ "SELECT col1, col2 FROM tablename"
+ )
+
+ def test_select_composition_eight(self):
+ self.assert_compile(select(
+ [table1.alias('t'), "foo.f"],
+ "foo.f = t.id",
+ from_obj=["(select f from bar where lala=heyhey) foo"]
+ ),
+ "SELECT t.myid, t.name, t.description, foo.f FROM mytable AS t, "
+ "(select f from bar where lala=heyhey) foo WHERE foo.f = t.id")
+
+ def test_select_bundle_columns(self):
+ self.assert_compile(select(
+ [table1, table2.c.otherid, "sysdate()", "foo, bar, lala"],
+ and_(
+ "foo.id = foofoo(lala)",
+ "datetime(foo) = Today",
+ table1.c.myid == table2.c.otherid,
+ )
+ ),
+ "SELECT mytable.myid, mytable.name, mytable.description, "
+ "myothertable.otherid, sysdate(), foo, bar, lala "
+ "FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND "
+ "datetime(foo) = Today AND mytable.myid = myothertable.otherid")
+
+class BindParamTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_legacy(self):
+ t = text("select * from foo where lala=:bar and hoho=:whee",
+ bindparams=[bindparam('bar', 4), bindparam('whee', 7)])
+
+ self.assert_compile(
+ t,
+ "select * from foo where lala=:bar and hoho=:whee",
+ checkparams={'bar': 4, 'whee': 7},
+ )
+
+ def test_positional(self):
+ t = text("select * from foo where lala=:bar and hoho=:whee")
+ t = t.bindparams(bindparam('bar', 4), bindparam('whee', 7))
+
+ self.assert_compile(
+ t,
+ "select * from foo where lala=:bar and hoho=:whee",
+ checkparams={'bar': 4, 'whee': 7},
+ )
+
+ def test_kw(self):
+ t = text("select * from foo where lala=:bar and hoho=:whee")
+ t = t.bindparams(bar=4, whee=7)
+
+ self.assert_compile(
+ t,
+ "select * from foo where lala=:bar and hoho=:whee",
+ checkparams={'bar': 4, 'whee': 7},
+ )
+
+ def test_positional_plus_kw(self):
+ t = text("select * from foo where lala=:bar and hoho=:whee")
+ t = t.bindparams(bindparam('bar', 4), whee=7)
+
+ self.assert_compile(
+ t,
+ "select * from foo where lala=:bar and hoho=:whee",
+ checkparams={'bar': 4, 'whee': 7},
+ )
+
+ def _assert_type_map(self, t, compare):
+ map_ = dict(
+ (b.key, b.type) for b in t._bindparams.values()
+ )
+ for k in compare:
+ assert compare[k]._type_affinity is map_[k]._type_affinity
+
+ def test_typing_construction(self):
+ t = text("select * from table :foo :bar :bat")
+
+ self._assert_type_map(t, {"foo": NullType(),
+ "bar": NullType(),
+ "bat": NullType()})
+
+ t = t.bindparams(bindparam('foo', type_=String))
+
+ self._assert_type_map(t, {"foo": String(),
+ "bar": NullType(),
+ "bat": NullType()})
+
+ t = t.bindparams(bindparam('bar', type_=Integer))
+
+ self._assert_type_map(t, {"foo": String(),
+ "bar": Integer(),
+ "bat": NullType()})
+
+ t = t.bindparams(bat=45.564)
+
+ self._assert_type_map(t, {"foo": String(),
+ "bar": Integer(),
+ "bat": Float()})
+
+
+ def test_binds_compiled_named(self):
+ self.assert_compile(
+ text("select * from foo where lala=:bar and hoho=:whee").
+ bindparams(bar=4, whee=7),
+ "select * from foo where lala=%(bar)s and hoho=%(whee)s",
+ checkparams={'bar': 4, 'whee': 7},
+ dialect="postgresql"
+ )
+
+ def test_binds_compiled_positional(self):
+ self.assert_compile(
+ text("select * from foo where lala=:bar and hoho=:whee").
+ bindparams(bar=4, whee=7),
+ "select * from foo where lala=? and hoho=?",
+ checkparams={'bar': 4, 'whee': 7},
+ dialect="sqlite"
+ )
+
+ def test_missing_bind_kw(self):
+ assert_raises_message(
+ exc.ArgumentError,
+ "This text\(\) construct doesn't define a bound parameter named 'bar'",
+ text(":foo").bindparams,
+ foo=5, bar=7
+ )
+
+ def test_missing_bind_posn(self):
+ assert_raises_message(
+ exc.ArgumentError,
+ "This text\(\) construct doesn't define a bound parameter named 'bar'",
+ text(":foo").bindparams,
+ bindparam('foo', value=5), bindparam('bar', value=7)
+ )
+
+ def test_escaping_colons(self):
+ # test escaping out text() params with a backslash
+ self.assert_compile(
+ text("select * from foo where clock='05:06:07' "
+ "and mork='\:mindy'"),
+ "select * from foo where clock='05:06:07' and mork=':mindy'",
+ checkparams={},
+ params={},
+ dialect="postgresql"
+ )
+
+
+ def test_text_in_select_nonfrom(self):
+
+ generate_series = text("generate_series(:x, :y, :z) as s(a)").\
+ bindparams(x=None, y=None, z=None)
+
+ s = select([
+ (func.current_date() + literal_column("s.a")).label("dates")
+ ]).select_from(generate_series)
+
+ self.assert_compile(
+ s,
+ "SELECT CURRENT_DATE + s.a AS dates FROM "
+ "generate_series(:x, :y, :z) as s(a)",
+ checkparams={'y': None, 'x': None, 'z': None}
+ )
+
+ self.assert_compile(
+ s.params(x=5, y=6, z=7),
+ "SELECT CURRENT_DATE + s.a AS dates FROM "
+ "generate_series(:x, :y, :z) as s(a)",
+ checkparams={'y': 6, 'x': 5, 'z': 7}
+ )
+
+class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_basic_toplevel_resultmap_positional(self):
+ t = text("select id, name from user").columns(
+ column('id', Integer),
+ column('name')
+ )
+
+ compiled = t.compile()
+ eq_(
+ compiled.result_map,
+ {
+ 'id': ('id', (t.c.id,), t.c.id.type),
+ 'name': ('name', (t.c.name,), t.c.name.type)
+ }
+ )
+
+ def test_basic_toplevel_resultmap(self):
+ t = text("select id, name from user").columns(id=Integer, name=String)
+
+ compiled = t.compile()
+ eq_(
+ compiled.result_map,
+ {
+ 'id': ('id', (t.c.id,), t.c.id.type),
+ 'name': ('name', (t.c.name,), t.c.name.type)
+ }
+ )
+
+ def test_basic_subquery_resultmap(self):
+ t = text("select id, name from user").columns(id=Integer, name=String)
+
+ stmt = select([table1.c.myid]).select_from(
+ table1.join(t, table1.c.myid == t.c.id))
+ compiled = stmt.compile()
+ eq_(
+ compiled.result_map,
+ {
+ "myid": ("myid",
+ (table1.c.myid, "myid", "myid"), table1.c.myid.type),
+ }
+ )
+
+ def test_cte(self):
+ t = text("select id, name from user").columns(id=Integer, name=String).cte('t')
+
+ s = select([table1]).where(table1.c.myid == t.c.id)
+ self.assert_compile(
+ s,
+ "WITH t AS (select id, name from user) "
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable, t WHERE mytable.myid = t.id"
+ )
+
+
+ def test_alias(self):
+ t = text("select id, name from user").columns(id=Integer, name=String).alias('t')
+
+ s = select([table1]).where(table1.c.myid == t.c.id)
+ self.assert_compile(
+ s,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable, (select id, name from user) AS t "
+ "WHERE mytable.myid = t.id"
+ )
+
+ def test_scalar_subquery(self):
+ t = text("select id from user").columns(id=Integer)
+ subq = t.as_scalar()
+
+ assert subq.type._type_affinity is Integer()._type_affinity
+
+ s = select([table1.c.myid, subq]).where(table1.c.myid == subq)
+ self.assert_compile(
+ s,
+ "SELECT mytable.myid, (select id from user) AS anon_1 "
+ "FROM mytable WHERE mytable.myid = (select id from user)"
+ )
\ No newline at end of file
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
