summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--doc/build/changelog/changelog_10.rst27
-rw-r--r--doc/build/changelog/migration_10.rst82
-rw-r--r--lib/sqlalchemy/orm/query.py5
-rw-r--r--lib/sqlalchemy/orm/strategies.py3
-rw-r--r--lib/sqlalchemy/orm/util.py32
-rw-r--r--lib/sqlalchemy/sql/compiler.py8
-rw-r--r--lib/sqlalchemy/sql/elements.py103
-rw-r--r--lib/sqlalchemy/sql/selectable.py19
-rw-r--r--lib/sqlalchemy/sql/util.py160
-rw-r--r--test/orm/test_eager_relations.py51
-rw-r--r--test/orm/test_query.py230
-rw-r--r--test/sql/test_generative.py310
-rw-r--r--test/sql/test_selectable.py7
-rw-r--r--test/sql/test_text.py64
14 files changed, 969 insertions, 132 deletions
diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 55a275659..cb7a9088d 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -22,6 +22,33 @@
on compatibility concerns, see :doc:`/changelog/migration_10`.
.. change::
+ :tags: bug, orm
+ :tickets: 3148, 3188
+
+ A major rework to the behavior of expression labels, most
+ specifically when used with ColumnProperty constructs with
+ custom SQL expressions and in conjunction with the "order by
+ labels" logic first introduced in 0.9. Fixes include that an
+ ``order_by(Entity.some_col_prop)`` will now make use of "order by
+ label" rules even if Entity has been subject to aliasing,
+ either via inheritance rendering or via the use of the
+ ``aliased()`` construct; rendering of the same column property
+ multiple times with aliasing (e.g. ``query(Entity.some_prop,
+ entity_alias.some_prop)``) will label each occurrence of the
+ entity with a distinct label, and additionally "order by
+ label" rules will work for both (e.g.
+ ``order_by(Entity.some_prop, entity_alias.some_prop)``).
+ Additional issues that could prevent the "order by label"
+ logic from working in 0.9, most notably that the state of a
+ Label could change such that "order by label" would stop
+ working depending on how things were called, has been fixed.
+
+ .. seealso::
+
+ :ref:`bug_3188`
+
+
+ .. change::
:tags: bug, mysql
:tickets: 3186
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index 2b9e5f0fd..6a48b31fa 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -8,7 +8,7 @@ What's New in SQLAlchemy 1.0?
undergoing maintenance releases as of May, 2014,
and SQLAlchemy version 1.0, as of yet unreleased.
- Document last updated: September 1, 2014
+ Document last updated: September 7, 2014
Introduction
============
@@ -307,6 +307,86 @@ Renders::
:ticket:`3177`
+
+.. _bug_3188:
+
+ColumnProperty constructs work a lot better with aliases, order_by
+-------------------------------------------------------------------
+
+A variety of issues regarding :func:`.column_property` have been fixed,
+most specifically with regards to the :func:`.aliased` construct as well
+as the "order by label" logic introduced in 0.9 (see :ref:`migration_1068`).
+
+Given a mapping like the following::
+
+ class A(Base):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+
+ class B(Base):
+ __tablename__ = 'b'
+
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey('a.id'))
+
+
+ A.b = column_property(
+ select([func.max(B.id)]).where(B.a_id == A.id).correlate(A)
+ )
+
+A simple scenario that included "A.b" twice would fail to render
+correctly::
+
+ print sess.query(A, a1).order_by(a1.b)
+
+This would order by the wrong column::
+
+ SELECT a.id AS a_id, (SELECT max(b.id) AS max_1 FROM b
+ WHERE b.a_id = a.id) AS anon_1, a_1.id AS a_1_id,
+ (SELECT max(b.id) AS max_2
+ FROM b WHERE b.a_id = a_1.id) AS anon_2
+ FROM a, a AS a_1 ORDER BY anon_1
+
+New output::
+
+ SELECT a.id AS a_id, (SELECT max(b.id) AS max_1
+ FROM b WHERE b.a_id = a.id) AS anon_1, a_1.id AS a_1_id,
+ (SELECT max(b.id) AS max_2
+ FROM b WHERE b.a_id = a_1.id) AS anon_2
+ FROM a, a AS a_1 ORDER BY anon_2
+
+There were also many scenarios where the "order by" logic would fail
+to order by label, for example if the mapping were "polymorphic"::
+
+ class A(Base):
+ __tablename__ = 'a'
+
+ id = Column(Integer, primary_key=True)
+ type = Column(String)
+
+ __mapper_args__ = {'polymorphic_on': type, 'with_polymorphic': '*'}
+
+The order_by would fail to use the label, as it would be anonymized due
+to the polymorphic loading::
+
+ SELECT a.id AS a_id, a.type AS a_type, (SELECT max(b.id) AS max_1
+ FROM b WHERE b.a_id = a.id) AS anon_1
+ FROM a ORDER BY (SELECT max(b.id) AS max_2
+ FROM b WHERE b.a_id = a.id)
+
+Now that the order by label tracks the anonymized label, this now works::
+
+ SELECT a.id AS a_id, a.type AS a_type, (SELECT max(b.id) AS max_1
+ FROM b WHERE b.a_id = a.id) AS anon_1
+ FROM a ORDER BY anon_1
+
+Included in these fixes are a variety of heisenbugs that could corrupt
+the state of an ``aliased()`` construct such that the labeling logic
+would again fail; these have also been fixed.
+
+:ticket:`3148` :ticket:`3188`
+
.. _behavioral_changes_orm_10:
Behavioral Changes - ORM
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index ba557ef79..60948293b 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -137,10 +137,7 @@ class Query(object):
)
aliased_adapter = None
elif ext_info.is_aliased_class:
- aliased_adapter = sql_util.ColumnAdapter(
- ext_info.selectable,
- ext_info.mapper._equivalent_columns
- )
+ aliased_adapter = ext_info._adapter
else:
aliased_adapter = None
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 2159d9135..cdb501c14 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -1242,7 +1242,8 @@ class JoinedLoader(AbstractRelationshipLoader):
clauses = orm_util.ORMAdapter(
to_adapt,
equivalents=self.mapper._equivalent_columns,
- adapt_required=True)
+ adapt_required=True, allow_label_resolve=False,
+ anonymize_labels=True)
assert clauses.aliased_class is not None
if self.parent_property.direction != interfaces.MANYTOONE:
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index ea7bfc294..1bb6b571e 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -270,15 +270,14 @@ first()
class ORMAdapter(sql_util.ColumnAdapter):
- """Extends ColumnAdapter to accept ORM entities.
-
- The selectable is extracted from the given entity,
- and the AliasedClass if any is referenced.
+ """ColumnAdapter subclass which excludes adaptation of entities from
+ non-matching mappers.
"""
def __init__(self, entity, equivalents=None, adapt_required=False,
- chain_to=None):
+ chain_to=None, allow_label_resolve=True,
+ anonymize_labels=False):
info = inspection.inspect(entity)
self.mapper = info.mapper
@@ -288,16 +287,18 @@ class ORMAdapter(sql_util.ColumnAdapter):
self.aliased_class = entity
else:
self.aliased_class = None
- sql_util.ColumnAdapter.__init__(self, selectable,
- equivalents, chain_to,
- adapt_required=adapt_required)
- def replace(self, elem):
+ sql_util.ColumnAdapter.__init__(
+ self, selectable, equivalents, chain_to,
+ adapt_required=adapt_required,
+ allow_label_resolve=allow_label_resolve,
+ anonymize_labels=anonymize_labels,
+ include_fn=self._include_fn
+ )
+
+ def _include_fn(self, elem):
entity = elem._annotations.get('parentmapper', None)
- if not entity or entity.isa(self.mapper):
- return sql_util.ColumnAdapter.replace(self, elem)
- else:
- return None
+ return not entity or entity.isa(self.mapper)
class AliasedClass(object):
@@ -354,6 +355,7 @@ class AliasedClass(object):
if alias is None:
alias = mapper._with_polymorphic_selectable.alias(
name=name, flat=flat)
+
self._aliased_insp = AliasedInsp(
self,
mapper,
@@ -460,9 +462,9 @@ class AliasedInsp(InspectionAttr):
self._base_alias = _base_alias or self
self._use_mapper_path = _use_mapper_path
- self._adapter = sql_util.ClauseAdapter(
+ self._adapter = sql_util.ColumnAdapter(
selectable, equivalents=mapper._equivalent_columns,
- adapt_on_names=adapt_on_names)
+ adapt_on_names=adapt_on_names, anonymize_labels=True)
self._adapt_on_names = adapt_on_names
self._target = mapper.class_
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index af0fff826..72dd11eaf 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -512,7 +512,7 @@ class SQLCompiler(Compiled):
selectable = self.stack[-1]['selectable']
try:
- col = selectable._inner_column_dict[element.text]
+ col = selectable._label_resolve_dict[element.text]
except KeyError:
# treat it like text()
util.warn_limited(
@@ -701,9 +701,7 @@ class SQLCompiler(Compiled):
# here; we can only add a label in the ORDER BY for an individual
# label expression in the columns clause.
- raw_col = set(l._order_by_label_element.name
- for l in order_by_select._raw_columns
- if l._order_by_label_element is not None)
+ raw_col = set(order_by_select._label_resolve_dict.keys())
return ", ".join(
s for s in
@@ -712,7 +710,7 @@ class SQLCompiler(Compiled):
self,
render_label_as_label=c._order_by_label_element if
c._order_by_label_element is not None and
- c._order_by_label_element.name in raw_col
+ c._order_by_label_element._label in raw_col
else None,
**kw)
for c in clauselist.clauses)
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 984cfe0ee..cf8de936d 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -625,8 +625,73 @@ class ColumnElement(operators.ColumnOperators, ClauseElement):
__visit_name__ = 'column'
primary_key = False
foreign_keys = []
- _label = _columns_clause_label = None
- _key_label = key = None
+
+ _label = None
+ """The named label that can be used to target
+ this column in a result set.
+
+ This label is almost always the label used when
+ rendering <expr> AS <label> in a SELECT statement. It also
+ refers to a name that this column expression can be located from
+ in a result set.
+
+ For a regular Column bound to a Table, this is typically the label
+ <tablename>_<columnname>. For other constructs, different rules
+ may apply, such as anonymized labels and others.
+
+ """
+
+ key = None
+ """the 'key' that in some circumstances refers to this object in a
+ Python namespace.
+
+ This typically refers to the "key" of the column as present in the
+ ``.c`` collection of a selectable, e.g. sometable.c["somekey"] would
+ return a Column with a .key of "somekey".
+
+ """
+
+ _key_label = None
+ """A label-based version of 'key' that in some circumstances refers
+ to this object in a Python namespace.
+
+
+ _key_label comes into play when a select() statement is constructed with
+ apply_labels(); in this case, all Column objects in the ``.c`` collection
+ are rendered as <tablename>_<columnname> in SQL; this is essentially the
+ value of ._label. But to locate those columns in the ``.c`` collection,
+ the name is along the lines of <tablename>_<key>; that's the typical
+ value of .key_label.
+
+ """
+
+ _render_label_in_columns_clause = True
+ """A flag used by select._columns_plus_names that helps to determine
+ we are actually going to render in terms of "SELECT <col> AS <label>".
+ This flag can be returned as False for some Column objects that want
+ to be rendered as simple "SELECT <col>"; typically columns that don't have
+ any parent table and are named the same as what the label would be
+ in any case.
+
+ """
+
+ _resolve_label = None
+ """The name that should be used to identify this ColumnElement in a
+ select() object when "label resolution" logic is used; this refers
+ to using a string name in an expression like order_by() or group_by()
+ that wishes to target a labeled expression in the columns clause.
+
+ The name is distinct from that of .name or ._label to account for the case
+ where anonymizing logic may be used to change the name that's actually
+ rendered at compile time; this attribute should hold onto the original
+ name that was user-assigned when producing a .label() construct.
+
+ """
+
+ _allow_label_resolve = True
+ """A flag that can be flipped to prevent a column from being resolvable
+ by string label name."""
+
_alt_names = ()
def self_group(self, against=None):
@@ -1183,7 +1248,9 @@ class TextClause(Executable, ClauseElement):
# help in those cases where text() is
# interpreted in a column expression situation
- key = _label = _columns_clause_label = None
+ key = _label = _resolve_label = None
+
+ _allow_label_resolve = False
def __init__(
self,
@@ -2521,7 +2588,7 @@ class UnaryExpression(ColumnElement):
return UnaryExpression(
expr, operator=operators.distinct_op, type_=expr.type)
- @util.memoized_property
+ @property
def _order_by_label_element(self):
if self.modifier in (operators.desc_op, operators.asc_op):
return self.element._order_by_label_element
@@ -2821,16 +2888,20 @@ class Label(ColumnElement):
:param obj: a :class:`.ColumnElement`.
"""
+
+ if isinstance(element, Label):
+ self._resolve_label = element._label
+
while isinstance(element, Label):
element = element.element
+
if name:
self.name = name
else:
self.name = _anonymous_label(
'%%(%d %s)s' % (id(self), getattr(element, 'name', 'anon'))
)
- self.key = self._label = self._key_label = \
- self._columns_clause_label = self.name
+ self.key = self._label = self._key_label = self.name
self._element = element
self._type = type_
self._proxies = [element]
@@ -2839,6 +2910,10 @@ class Label(ColumnElement):
return self.__class__, (self.name, self._element, self._type)
@util.memoized_property
+ def _allow_label_resolve(self):
+ return self.element._allow_label_resolve
+
+ @property
def _order_by_label_element(self):
return self
@@ -2872,8 +2947,15 @@ class Label(ColumnElement):
def get_children(self, **kwargs):
return self.element,
- def _copy_internals(self, clone=_clone, **kw):
+ def _copy_internals(self, clone=_clone, anonymize_labels=False, **kw):
self.element = clone(self.element, **kw)
+ self.__dict__.pop('_allow_label_resolve', None)
+ if anonymize_labels:
+ self.name = _anonymous_label(
+ '%%(%d %s)s' % (
+ id(self), getattr(self.element, 'name', 'anon'))
+ )
+ self.key = self._label = self._key_label = self.name
@property
def _from_objects(self):
@@ -3066,11 +3148,8 @@ class ColumnClause(Immutable, ColumnElement):
return self._gen_label(self.name)
@_memoized_property
- def _columns_clause_label(self):
- if self.table is None:
- return None
- else:
- return self._label
+ def _render_label_in_columns_clause(self):
+ return self.table is not None
def _gen_label(self, name):
t = self.table
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index a49493995..57b16f45f 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1814,7 +1814,7 @@ class GenerativeSelect(SelectBase):
*clauses, _literal_as_text=_literal_as_label_reference)
@property
- def _inner_column_dict(self):
+ def _label_resolve_dict(self):
raise NotImplementedError()
def _copy_internals(self, clone=_clone, **kw):
@@ -1884,7 +1884,7 @@ class CompoundSelect(GenerativeSelect):
GenerativeSelect.__init__(self, **kwargs)
@property
- def _inner_column_dict(self):
+ def _label_resolve_dict(self):
return dict(
(c.key, c) for c in self.c
)
@@ -2498,11 +2498,14 @@ class Select(HasPrefixes, GenerativeSelect):
return _select_iterables(self._raw_columns)
@_memoized_property
- def _inner_column_dict(self):
+ def _label_resolve_dict(self):
d = dict(
- (c._label or c.key, c)
- for c in _select_iterables(self._raw_columns))
- d.update((c.key, c) for c in _select_iterables(self.froms))
+ (c._resolve_label or c._label or c.key, c)
+ for c in _select_iterables(self._raw_columns)
+ if c._allow_label_resolve)
+ d.update(
+ (c.key, c) for c in
+ _select_iterables(self.froms) if c._allow_label_resolve)
return d
@@ -2974,10 +2977,10 @@ class Select(HasPrefixes, GenerativeSelect):
names = set()
def name_for_col(c):
- if c._columns_clause_label is None:
+ if c._label is None or not c._render_label_in_columns_clause:
return (None, c)
- name = c._columns_clause_label
+ name = c._label
if name in names:
name = c.anon_label
else:
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 8bbae8b93..d6f3b5915 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -428,35 +428,6 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None,
return pairs
-class AliasedRow(object):
- """Wrap a RowProxy with a translation map.
-
- This object allows a set of keys to be translated
- to those present in a RowProxy.
-
- """
-
- def __init__(self, row, map):
- # AliasedRow objects don't nest, so un-nest
- # if another AliasedRow was passed
- if isinstance(row, AliasedRow):
- self.row = row.row
- else:
- self.row = row
- self.map = map
-
- def __contains__(self, key):
- return self.map[key] in self.row
-
- def has_key(self, key):
- return key in self
-
- def __getitem__(self, key):
- return self.row[self.map[key]]
-
- def keys(self):
- return self.row.keys()
-
class ClauseAdapter(visitors.ReplacingCloningVisitor):
"""Clones and modifies clauses based on column correspondence.
@@ -486,21 +457,14 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
"""
def __init__(self, selectable, equivalents=None,
- include=None, exclude=None,
include_fn=None, exclude_fn=None,
- adapt_on_names=False):
- self.__traverse_options__ = {'stop_on': [selectable]}
+ adapt_on_names=False, anonymize_labels=False):
+ self.__traverse_options__ = {
+ 'stop_on': [selectable],
+ 'anonymize_labels': anonymize_labels}
self.selectable = selectable
- if include:
- assert not include_fn
- self.include_fn = lambda e: e in include
- else:
- self.include_fn = include_fn
- if exclude:
- assert not exclude_fn
- self.exclude_fn = lambda e: e in exclude
- else:
- self.exclude_fn = exclude_fn
+ self.include_fn = include_fn
+ self.exclude_fn = exclude_fn
self.equivalents = util.column_dict(equivalents or {})
self.adapt_on_names = adapt_on_names
@@ -520,10 +484,8 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
newcol = self.selectable.c.get(col.name)
return newcol
- magic_flag = False
-
def replace(self, col):
- if not self.magic_flag and isinstance(col, FromClause) and \
+ if isinstance(col, FromClause) and \
self.selectable.is_derived_from(col):
return self.selectable
elif not isinstance(col, ColumnElement):
@@ -539,62 +501,102 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
class ColumnAdapter(ClauseAdapter):
"""Extends ClauseAdapter with extra utility functions.
- Provides the ability to "wrap" this ClauseAdapter
- around another, a columns dictionary which returns
- adapted elements given an original, and an
- adapted_row() factory.
+ Key aspects of ColumnAdapter include:
+
+ * Expressions that are adapted are stored in a persistent
+ .columns collection; so that an expression E adapted into
+ an expression E1, will return the same object E1 when adapted
+ a second time. This is important in particular for things like
+ Label objects that are anonymized, so that the ColumnAdapter can
+ be used to present a consistent "adapted" view of things.
+
+ * Exclusion of items from the persistent collection based on
+ include/exclude rules, but also independent of hash identity.
+ This because "annotated" items all have the same hash identity as their
+ parent.
+
+ * "wrapping" capability is added, so that the replacement of an expression
+ E can proceed through a series of adapters. This differs from the
+ visitor's "chaining" feature in that the resulting object is passed
+ through all replacing functions unconditionally, rather than stopping
+ at the first one that returns non-None.
+
+ * An adapt_required option, used by eager loading to indicate that
+ We don't trust a result row column that is not translated.
+ This is to prevent a column from being interpreted as that
+ of the child row in a self-referential scenario, see
+ inheritance/test_basic.py->EagerTargetingTest.test_adapt_stringency
"""
def __init__(self, selectable, equivalents=None,
- chain_to=None, include=None,
- exclude=None, adapt_required=False):
+ chain_to=None, adapt_required=False,
+ include_fn=None, exclude_fn=None,
+ adapt_on_names=False,
+ allow_label_resolve=True,
+ anonymize_labels=False):
ClauseAdapter.__init__(self, selectable, equivalents,
- include, exclude)
+ include_fn=include_fn, exclude_fn=exclude_fn,
+ adapt_on_names=adapt_on_names,
+ anonymize_labels=anonymize_labels)
+
if chain_to:
self.chain(chain_to)
self.columns = util.populate_column_dict(self._locate_col)
+ if self.include_fn or self.exclude_fn:
+ self.columns = self._IncludeExcludeMapping(self, self.columns)
self.adapt_required = adapt_required
+ self.allow_label_resolve = allow_label_resolve
+ self._wrap = None
+
+ class _IncludeExcludeMapping(object):
+ def __init__(self, parent, columns):
+ self.parent = parent
+ self.columns = columns
+
+ def __getitem__(self, key):
+ if (
+ self.parent.include_fn and not self.parent.include_fn(key)
+ ) or (
+ self.parent.exclude_fn and self.parent.exclude_fn(key)
+ ):
+ if self.parent._wrap:
+ return self.parent._wrap.columns[key]
+ else:
+ return key
+ return self.columns[key]
def wrap(self, adapter):
ac = self.__class__.__new__(self.__class__)
- ac.__dict__ = self.__dict__.copy()
- ac._locate_col = ac._wrap(ac._locate_col, adapter._locate_col)
- ac.adapt_clause = ac._wrap(ac.adapt_clause, adapter.adapt_clause)
- ac.adapt_list = ac._wrap(ac.adapt_list, adapter.adapt_list)
+ ac.__dict__.update(self.__dict__)
+ ac._wrap = adapter
ac.columns = util.populate_column_dict(ac._locate_col)
+ if ac.include_fn or ac.exclude_fn:
+ ac.columns = self._IncludeExcludeMapping(ac, ac.columns)
+
return ac
- adapt_clause = ClauseAdapter.traverse
- adapt_list = ClauseAdapter.copy_and_process
+ def traverse(self, obj):
+ return self.columns[obj]
- def _wrap(self, local, wrapped):
- def locate(col):
- col = local(col)
- return wrapped(col)
- return locate
+ adapt_clause = traverse
+ adapt_list = ClauseAdapter.copy_and_process
def _locate_col(self, col):
- c = self._corresponding_column(col, True)
- if c is None:
- c = self.adapt_clause(col)
-
- # anonymize labels in case they have a hardcoded name
- if isinstance(c, Label):
- c = c.label(None)
-
- # adapt_required used by eager loading to indicate that
- # we don't trust a result row column that is not translated.
- # this is to prevent a column from being interpreted as that
- # of the child row in a self-referential scenario, see
- # inheritance/test_basic.py->EagerTargetingTest.test_adapt_stringency
+
+ c = ClauseAdapter.traverse(self, col)
+
+ if self._wrap:
+ c2 = self._wrap._locate_col(c)
+ if c2 is not None:
+ c = c2
+
if self.adapt_required and c is col:
return None
- return c
+ c._allow_label_resolve = self.allow_label_resolve
- def adapted_row(self, row):
- return AliasedRow(row, self.columns)
+ return c
def __getstate__(self):
d = self.__dict__.copy()
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index b0c203bf1..214b592b5 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -14,7 +14,7 @@ from sqlalchemy.orm import mapper, relationship, create_session, \
from sqlalchemy.sql import operators
from sqlalchemy.testing import assert_raises, assert_raises_message
from sqlalchemy.testing.assertsql import CompiledSQL
-from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import fixtures, expect_warnings
from test.orm import _fixtures
from sqlalchemy.util import OrderedDict as odict
import datetime
@@ -210,6 +210,55 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
User(id=10, addresses=[])
], sess.query(User).order_by(User.id).all())
+ def test_no_ad_hoc_orderby(self):
+ """part of #2992; make sure string label references can't
+ access an eager loader, else an eager load can corrupt the query.
+
+ """
+ Address, addresses, users, User = (self.classes.Address,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
+ mapper(Address, addresses)
+ mapper(User, users, properties=dict(
+ addresses=relationship(
+ Address),
+ ))
+
+ sess = create_session()
+ q = sess.query(User).\
+ join("addresses").\
+ options(joinedload("addresses")).\
+ order_by("email_address")
+
+ self.assert_compile(
+ q,
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "addresses_1.id AS addresses_1_id, addresses_1.user_id AS "
+ "addresses_1_user_id, addresses_1.email_address AS "
+ "addresses_1_email_address FROM users JOIN addresses "
+ "ON users.id = addresses.user_id LEFT OUTER JOIN addresses "
+ "AS addresses_1 ON users.id = addresses_1.user_id "
+ "ORDER BY addresses.email_address"
+ )
+
+ q = sess.query(User).options(joinedload("addresses")).\
+ order_by("email_address")
+
+ with expect_warnings("Can't resolve label reference 'email_address'"):
+ self.assert_compile(
+ q,
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "addresses_1.id AS addresses_1_id, addresses_1.user_id AS "
+ "addresses_1_user_id, addresses_1.email_address AS "
+ "addresses_1_email_address FROM users LEFT OUTER JOIN "
+ "addresses AS addresses_1 ON users.id = addresses_1.user_id "
+ "ORDER BY email_address"
+ )
+
+
+
def test_deferred_fk_col(self):
users, Dingaling, User, dingalings, Address, addresses = (
self.tables.users,
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index f0470e172..3f6813138 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -8,13 +8,13 @@ from sqlalchemy.engine import default
from sqlalchemy.orm import (
attributes, mapper, relationship, create_session, synonym, Session,
aliased, column_property, joinedload_all, joinedload, Query, Bundle,
- subqueryload, backref, lazyload)
+ subqueryload, backref, lazyload, defer)
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.schema import Table, Column
import sqlalchemy as sa
from sqlalchemy.testing.assertions import (
eq_, assert_raises, assert_raises_message, expect_warnings)
-from sqlalchemy.testing import fixtures, AssertsCompiledSQL
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL, assert_warnings
from test.orm import _fixtures
from sqlalchemy.orm.util import join, with_parent
@@ -1232,6 +1232,232 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
Address(email_address='jack@bean.com', user_id=7, id=1))])
+class ColumnPropertyTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+ run_setup_mappers = 'each'
+
+ def _fixture(self, label=True):
+ User, Address = self.classes("User", "Address")
+ users, addresses = self.tables("users", "addresses")
+ stmt = select([func.max(addresses.c.email_address)]).\
+ where(addresses.c.user_id == users.c.id).\
+ correlate(users)
+ if label:
+ stmt = stmt.label("email_ad")
+
+ mapper(User, users, properties={
+ "ead": column_property(stmt)
+ })
+ mapper(Address, addresses)
+
+ def test_order_by_column_prop_string(self):
+ User, Address = self.classes("User", "Address")
+ self._fixture(label=True)
+
+ s = Session()
+ q = s.query(User).order_by("email_ad")
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses "
+ "WHERE addresses.user_id = users.id) AS email_ad, "
+ "users.id AS users_id, users.name AS users_name "
+ "FROM users ORDER BY email_ad"
+ )
+
+ def test_order_by_column_prop_aliased_string(self):
+ User, Address = self.classes("User", "Address")
+ self._fixture(label=True)
+
+ s = Session()
+ ua = aliased(User)
+ q = s.query(ua).order_by("email_ad")
+
+ def go():
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users_1.id) "
+ "AS anon_1, users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name FROM users AS users_1 "
+ "ORDER BY email_ad"
+ )
+ assert_warnings(
+ go,
+ ["Can't resolve label reference 'email_ad'"], regex=True)
+
+ def test_order_by_column_labeled_prop_attr_aliased_one(self):
+ User = self.classes.User
+ self._fixture(label=True)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(ua).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users_1.id) AS anon_1, "
+ "users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users AS users_1 ORDER BY anon_1"
+ )
+
+ def test_order_by_column_labeled_prop_attr_aliased_two(self):
+ User = self.classes.User
+ self._fixture(label=True)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(ua.ead).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, "
+ "users AS users_1 WHERE addresses.user_id = users_1.id) "
+ "AS anon_1 ORDER BY anon_1"
+ )
+
+ # we're also testing that the state of "ua" is OK after the
+ # previous call, so the batching into one test is intentional
+ q = s.query(ua).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users_1.id) AS anon_1, "
+ "users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users AS users_1 ORDER BY anon_1"
+ )
+
+ def test_order_by_column_labeled_prop_attr_aliased_three(self):
+ User = self.classes.User
+ self._fixture(label=True)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(User.ead, ua.ead).order_by(User.ead, ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, users WHERE addresses.user_id = users.id) "
+ "AS email_ad, (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, users AS users_1 WHERE addresses.user_id = "
+ "users_1.id) AS anon_1 ORDER BY email_ad, anon_1"
+ )
+
+ q = s.query(User, ua).order_by(User.ead, ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users.id) AS "
+ "email_ad, users.id AS users_id, users.name AS users_name, "
+ "(SELECT max(addresses.email_address) AS max_1 FROM addresses "
+ "WHERE addresses.user_id = users_1.id) AS anon_1, users_1.id "
+ "AS users_1_id, users_1.name AS users_1_name FROM users, "
+ "users AS users_1 ORDER BY email_ad, anon_1"
+ )
+
+ def test_order_by_column_unlabeled_prop_attr_aliased_one(self):
+ User = self.classes.User
+ self._fixture(label=False)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(ua).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users_1.id) AS anon_1, "
+ "users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users AS users_1 ORDER BY anon_1"
+ )
+
+ def test_order_by_column_unlabeled_prop_attr_aliased_two(self):
+ User = self.classes.User
+ self._fixture(label=False)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(ua.ead).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, "
+ "users AS users_1 WHERE addresses.user_id = users_1.id) "
+ "AS anon_1 ORDER BY anon_1"
+ )
+
+ # we're also testing that the state of "ua" is OK after the
+ # previous call, so the batching into one test is intentional
+ q = s.query(ua).order_by(ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users_1.id) AS anon_1, "
+ "users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users AS users_1 ORDER BY anon_1"
+ )
+
+ def test_order_by_column_unlabeled_prop_attr_aliased_three(self):
+ User = self.classes.User
+ self._fixture(label=False)
+
+ ua = aliased(User)
+ s = Session()
+ q = s.query(User.ead, ua.ead).order_by(User.ead, ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, users WHERE addresses.user_id = users.id) "
+ "AS anon_1, (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses, users AS users_1 "
+ "WHERE addresses.user_id = users_1.id) AS anon_2 "
+ "ORDER BY anon_1, anon_2"
+ )
+
+ q = s.query(User, ua).order_by(User.ead, ua.ead)
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses WHERE addresses.user_id = users.id) AS "
+ "anon_1, users.id AS users_id, users.name AS users_name, "
+ "(SELECT max(addresses.email_address) AS max_1 FROM addresses "
+ "WHERE addresses.user_id = users_1.id) AS anon_2, users_1.id "
+ "AS users_1_id, users_1.name AS users_1_name FROM users, "
+ "users AS users_1 ORDER BY anon_1, anon_2"
+ )
+
+ def test_order_by_column_prop_attr(self):
+ User, Address = self.classes("User", "Address")
+ self._fixture(label=True)
+
+ s = Session()
+ q = s.query(User).order_by(User.ead)
+ # this one is a bit of a surprise; this is compiler
+ # label-order-by logic kicking in, but won't work in more
+ # complex cases.
+ self.assert_compile(
+ q,
+ "SELECT (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses "
+ "WHERE addresses.user_id = users.id) AS email_ad, "
+ "users.id AS users_id, users.name AS users_name "
+ "FROM users ORDER BY email_ad"
+ )
+
+ def test_order_by_column_prop_attr_non_present(self):
+ User, Address = self.classes("User", "Address")
+ self._fixture(label=True)
+
+ s = Session()
+ q = s.query(User).options(defer(User.ead)).order_by(User.ead)
+ self.assert_compile(
+ q,
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users ORDER BY (SELECT max(addresses.email_address) AS max_1 "
+ "FROM addresses "
+ "WHERE addresses.user_id = users.id)"
+ )
+
+
# more slice tests are available in test/orm/generative.py
class SliceTest(QueryTest):
def test_first(self):
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index 2e3c4b1e8..013ba8082 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -10,7 +10,7 @@ from sqlalchemy.sql.visitors import ClauseVisitor, CloningVisitor, \
cloned_traverse, ReplacingCloningVisitor
from sqlalchemy import exc
from sqlalchemy.sql import util as sql_util
-from sqlalchemy.testing import eq_, is_, assert_raises, assert_raises_message
+from sqlalchemy.testing import eq_, is_, is_not_, assert_raises, assert_raises_message
A = B = t1 = t2 = t3 = table1 = table2 = table3 = table4 = None
@@ -696,6 +696,244 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
"AS anon_1 WHERE table1.col1 = anon_1.col1)")
+class ColumnAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ @classmethod
+ def setup_class(cls):
+ global t1, t2
+ t1 = table("table1",
+ column("col1"),
+ column("col2"),
+ column("col3"),
+ column("col4")
+ )
+ t2 = table("table2",
+ column("col1"),
+ column("col2"),
+ column("col3"),
+ )
+
+ def test_traverse_memoizes_w_columns(self):
+ t1a = t1.alias()
+ adapter = sql_util.ColumnAdapter(t1a, anonymize_labels=True)
+
+ expr = select([t1a.c.col1]).label('x')
+ expr_adapted = adapter.traverse(expr)
+ is_not_(expr, expr_adapted)
+ is_(
+ adapter.columns[expr],
+ expr_adapted
+ )
+
+ def test_traverse_memoizes_w_itself(self):
+ t1a = t1.alias()
+ adapter = sql_util.ColumnAdapter(t1a, anonymize_labels=True)
+
+ expr = select([t1a.c.col1]).label('x')
+ expr_adapted = adapter.traverse(expr)
+ is_not_(expr, expr_adapted)
+ is_(
+ adapter.traverse(expr),
+ expr_adapted
+ )
+
+ def test_columns_memoizes_w_itself(self):
+ t1a = t1.alias()
+ adapter = sql_util.ColumnAdapter(t1a, anonymize_labels=True)
+
+ expr = select([t1a.c.col1]).label('x')
+ expr_adapted = adapter.columns[expr]
+ is_not_(expr, expr_adapted)
+ is_(
+ adapter.columns[expr],
+ expr_adapted
+ )
+
+ def test_wrapping_fallthrough(self):
+ t1a = t1.alias(name="t1a")
+ t2a = t2.alias(name="t2a")
+ a1 = sql_util.ColumnAdapter(t1a)
+
+ s1 = select([t1a.c.col1, t2a.c.col1]).apply_labels().alias()
+ a2 = sql_util.ColumnAdapter(s1)
+ a3 = a2.wrap(a1)
+ a4 = a1.wrap(a2)
+ a5 = a1.chain(a2)
+
+ # t1.c.col1 -> s1.c.t1a_col1
+
+ # adapted by a2
+ is_(
+ a3.columns[t1.c.col1], s1.c.t1a_col1
+ )
+ is_(
+ a4.columns[t1.c.col1], s1.c.t1a_col1
+ )
+
+ # chaining can't fall through because a1 grabs it
+ # first
+ is_(
+ a5.columns[t1.c.col1], t1a.c.col1
+ )
+
+ # t2.c.col1 -> s1.c.t2a_col1
+
+ # adapted by a2
+ is_(
+ a3.columns[t2.c.col1], s1.c.t2a_col1
+ )
+ is_(
+ a4.columns[t2.c.col1], s1.c.t2a_col1
+ )
+ # chaining, t2 hits s1
+ is_(
+ a5.columns[t2.c.col1], s1.c.t2a_col1
+ )
+
+ # t1.c.col2 -> t1a.c.col2
+
+ # fallthrough to a1
+ is_(
+ a3.columns[t1.c.col2], t1a.c.col2
+ )
+ is_(
+ a4.columns[t1.c.col2], t1a.c.col2
+ )
+
+ # chaining hits a1
+ is_(
+ a5.columns[t1.c.col2], t1a.c.col2
+ )
+
+ # t2.c.col2 -> t2.c.col2
+
+ # fallthrough to no adaption
+ is_(
+ a3.columns[t2.c.col2], t2.c.col2
+ )
+ is_(
+ a4.columns[t2.c.col2], t2.c.col2
+ )
+
+ def test_wrapping_ordering(self):
+ """illustrate an example where order of wrappers matters.
+
+ This test illustrates both the ordering being significant
+ as well as a scenario where multiple translations are needed
+ (e.g. wrapping vs. chaining).
+
+ """
+
+ stmt = select([t1.c.col1, t2.c.col1]).apply_labels()
+
+ sa = stmt.alias()
+ stmt2 = select([t2, sa])
+
+ a1 = sql_util.ColumnAdapter(stmt)
+ a2 = sql_util.ColumnAdapter(stmt2)
+
+ a2_to_a1 = a2.wrap(a1)
+ a1_to_a2 = a1.wrap(a2)
+
+ # when stmt2 and stmt represent the same column
+ # in different contexts, order of wrapping matters
+
+ # t2.c.col1 via a2 is stmt2.c.col1; then ignored by a1
+ is_(
+ a2_to_a1.columns[t2.c.col1], stmt2.c.col1
+ )
+ # t2.c.col1 via a1 is stmt.c.table2_col1; a2 then
+ # sends this to stmt2.c.table2_col1
+ is_(
+ a1_to_a2.columns[t2.c.col1], stmt2.c.table2_col1
+ )
+
+ # for mutually exclusive columns, order doesn't matter
+ is_(
+ a2_to_a1.columns[t1.c.col1], stmt2.c.table1_col1
+ )
+ is_(
+ a1_to_a2.columns[t1.c.col1], stmt2.c.table1_col1
+ )
+ is_(
+ a2_to_a1.columns[t2.c.col2], stmt2.c.col2
+ )
+
+
+ def test_wrapping_multiple(self):
+ """illustrate that wrapping runs both adapters"""
+
+ t1a = t1.alias(name="t1a")
+ t2a = t2.alias(name="t2a")
+ a1 = sql_util.ColumnAdapter(t1a)
+ a2 = sql_util.ColumnAdapter(t2a)
+ a3 = a2.wrap(a1)
+
+ stmt = select([t1.c.col1, t2.c.col2])
+
+ self.assert_compile(
+ a3.traverse(stmt),
+ "SELECT t1a.col1, t2a.col2 FROM table1 AS t1a, table2 AS t2a"
+ )
+
+ # chaining does too because these adapters don't share any
+ # columns
+ a4 = a2.chain(a1)
+ self.assert_compile(
+ a4.traverse(stmt),
+ "SELECT t1a.col1, t2a.col2 FROM table1 AS t1a, table2 AS t2a"
+ )
+
+ def test_wrapping_inclusions(self):
+ """test wrapping and inclusion rules together,
+ taking into account multiple objects with equivalent hash identity."""
+
+ t1a = t1.alias(name="t1a")
+ t2a = t2.alias(name="t2a")
+ a1 = sql_util.ColumnAdapter(
+ t1a,
+ include_fn=lambda col: "a1" in col._annotations)
+
+ s1 = select([t1a, t2a]).apply_labels().alias()
+ a2 = sql_util.ColumnAdapter(
+ s1,
+ include_fn=lambda col: "a2" in col._annotations)
+ a3 = a2.wrap(a1)
+
+ c1a1 = t1.c.col1._annotate(dict(a1=True))
+ c1a2 = t1.c.col1._annotate(dict(a2=True))
+ c1aa = t1.c.col1._annotate(dict(a1=True, a2=True))
+
+ c2a1 = t2.c.col1._annotate(dict(a1=True))
+ c2a2 = t2.c.col1._annotate(dict(a2=True))
+ c2aa = t2.c.col1._annotate(dict(a1=True, a2=True))
+
+ is_(
+ a3.columns[c1a1], t1a.c.col1
+ )
+ is_(
+ a3.columns[c1a2], s1.c.t1a_col1
+ )
+ is_(
+ a3.columns[c1aa], s1.c.t1a_col1
+ )
+
+ # not covered by a1, accepted by a2
+ is_(
+ a3.columns[c2aa], s1.c.t2a_col1
+ )
+
+ # not covered by a1, accepted by a2
+ is_(
+ a3.columns[c2a2], s1.c.t2a_col1
+ )
+ # not covered by a1, rejected by a2
+ is_(
+ a3.columns[c2a1], c2a1
+ )
+
+
class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = 'default'
@@ -1022,7 +1260,7 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
assert str(e) == "a.id = a.xxx_id"
b = a.alias()
- e = sql_util.ClauseAdapter(b, include=set([a.c.id]),
+ e = sql_util.ClauseAdapter(b, include_fn=lambda x: x in set([a.c.id]),
equivalents={a.c.id: set([a.c.id])}
).traverse(e)
@@ -1207,6 +1445,74 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
"WHERE c.bid = anon_1.b_aid"
)
+ t1 = table("table1",
+ column("col1"),
+ column("col2"),
+ column("col3"),
+ )
+ t2 = table("table2",
+ column("col1"),
+ column("col2"),
+ column("col3"),
+ )
+
+ def test_label_anonymize_one(self):
+ t1a = t1.alias()
+ adapter = sql_util.ClauseAdapter(t1a, anonymize_labels=True)
+
+ expr = select([t1.c.col2]).where(t1.c.col3 == 5).label('expr')
+ expr_adapted = adapter.traverse(expr)
+
+ stmt = select([expr, expr_adapted]).order_by(expr, expr_adapted)
+ self.assert_compile(
+ stmt,
+ "SELECT "
+ "(SELECT table1.col2 FROM table1 WHERE table1.col3 = :col3_1) "
+ "AS expr, "
+ "(SELECT table1_1.col2 FROM table1 AS table1_1 "
+ "WHERE table1_1.col3 = :col3_2) AS anon_1 "
+ "ORDER BY expr, anon_1"
+ )
+
+ def test_label_anonymize_two(self):
+ t1a = t1.alias()
+ adapter = sql_util.ClauseAdapter(t1a, anonymize_labels=True)
+
+ expr = select([t1.c.col2]).where(t1.c.col3 == 5).label(None)
+ expr_adapted = adapter.traverse(expr)
+
+ stmt = select([expr, expr_adapted]).order_by(expr, expr_adapted)
+ self.assert_compile(
+ stmt,
+ "SELECT "
+ "(SELECT table1.col2 FROM table1 WHERE table1.col3 = :col3_1) "
+ "AS anon_1, "
+ "(SELECT table1_1.col2 FROM table1 AS table1_1 "
+ "WHERE table1_1.col3 = :col3_2) AS anon_2 "
+ "ORDER BY anon_1, anon_2"
+ )
+
+ def test_label_anonymize_three(self):
+ t1a = t1.alias()
+ adapter = sql_util.ColumnAdapter(
+ t1a, anonymize_labels=True,
+ allow_label_resolve=False)
+
+ expr = select([t1.c.col2]).where(t1.c.col3 == 5).label(None)
+ l1 = expr
+ is_(l1._order_by_label_element, l1)
+ eq_(l1._allow_label_resolve, True)
+
+ expr_adapted = adapter.traverse(expr)
+ l2 = expr_adapted
+ is_(l2._order_by_label_element, l2)
+ eq_(l2._allow_label_resolve, False)
+
+ l3 = adapter.traverse(expr)
+ is_(l3._order_by_label_element, l3)
+ eq_(l3._allow_label_resolve, False)
+
+
class SpliceJoinsTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = 'default'
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index c5736b26f..a3b2b0e93 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1724,6 +1724,13 @@ class AnnotationsTest(fixtures.TestBase):
b5 = visitors.cloned_traverse(b3, {}, {'binary': visit_binary})
assert str(b5) == ":bar = table1.col2"
+ def test_label_accessors(self):
+ t1 = table('t1', column('c1'))
+ l1 = t1.c.c1.label(None)
+ is_(l1._order_by_label_element, l1)
+ l1a = l1._annotate({"foo": "bar"})
+ is_(l1a._order_by_label_element, l1a)
+
def test_annotate_aliased(self):
t1 = table('t1', column('c1'))
s = select([(t1.c.c1 + 3).label('bat')])
diff --git a/test/sql/test_text.py b/test/sql/test_text.py
index e84a2907c..60d90196e 100644
--- a/test/sql/test_text.py
+++ b/test/sql/test_text.py
@@ -1,12 +1,12 @@
"""Test the TextClause and related constructs."""
from sqlalchemy.testing import fixtures, AssertsCompiledSQL, eq_, \
- assert_raises_message, expect_warnings
+ assert_raises_message, expect_warnings, assert_warnings
from sqlalchemy import text, select, Integer, String, Float, \
bindparam, and_, func, literal_column, exc, MetaData, Table, Column,\
asc, func, desc, union
from sqlalchemy.types import NullType
-from sqlalchemy.sql import table, column
+from sqlalchemy.sql import table, column, util as sql_util
from sqlalchemy import util
table1 = table('mytable',
@@ -679,3 +679,63 @@ class OrderByLabelResolutionTest(fixtures.TestBase, AssertsCompiledSQL):
desc("somelabel"),
"somelabel DESC"
)
+
+ def test_columnadapter_anonymized(self):
+ """test issue #3148
+
+ Testing the anonymization applied from the ColumnAdapter.columns
+ collection, typically as used in eager loading.
+
+ """
+ exprs = [
+ table1.c.myid,
+ table1.c.name.label('t1name'),
+ func.foo("hoho").label('x')]
+
+ ta = table1.alias()
+ adapter = sql_util.ColumnAdapter(ta, anonymize_labels=True)
+
+ s1 = select([adapter.columns[expr] for expr in exprs]).\
+ apply_labels().order_by("myid", "t1name", "x")
+
+ def go():
+ # the labels here are anonymized, so label naming
+ # can't catch these.
+ self.assert_compile(
+ s1,
+ "SELECT mytable_1.myid AS mytable_1_myid, "
+ "mytable_1.name AS name_1, foo(:foo_2) AS foo_1 "
+ "FROM mytable AS mytable_1 ORDER BY mytable_1.myid, t1name, x"
+ )
+
+ assert_warnings(
+ go,
+ ["Can't resolve label reference 't1name'",
+ "Can't resolve label reference 'x'"], regex=True)
+
+ def test_columnadapter_non_anonymized(self):
+ """test issue #3148
+
+ Testing the anonymization applied from the ColumnAdapter.columns
+ collection, typically as used in eager loading.
+
+ """
+ exprs = [
+ table1.c.myid,
+ table1.c.name.label('t1name'),
+ func.foo("hoho").label('x')]
+
+ ta = table1.alias()
+ adapter = sql_util.ColumnAdapter(ta)
+
+ s1 = select([adapter.columns[expr] for expr in exprs]).\
+ apply_labels().order_by("myid", "t1name", "x")
+
+ # labels are maintained
+ self.assert_compile(
+ s1,
+ "SELECT mytable_1.myid AS mytable_1_myid, "
+ "mytable_1.name AS t1name, foo(:foo_1) AS x "
+ "FROM mytable AS mytable_1 ORDER BY mytable_1.myid, t1name, x"
+ )
+
View Lorry Controller Status