def returning_clause(self, stmt, returning_cols):
columns = [
- self._label_select_column(None, c, True, False, {})
+ self._label_returning_column(stmt, c)
for c in expression._select_iterables(returning_cols)
]
# necessarily used an expensive KeyError in order to match.
columns = [
- self._label_select_column(
- None,
+ self._label_returning_column(
+ stmt,
adapter.traverse(c),
- True,
- False,
{"result_map_targets": (c,)},
)
for c in expression._select_iterables(returning_cols)
def returning_clause(self, stmt, returning_cols):
columns = [
- self._label_select_column(None, c, True, False, {})
+ self._label_returning_column(stmt, c)
for c in expression._select_iterables(returning_cols)
]
result = map_.get(key if self.case_sensitive else key.lower())
elif isinstance(key, expression.ColumnElement):
if (
- key._label
- and (key._label if self.case_sensitive else key._label.lower())
+ key._tq_label
+ and (
+ key._tq_label
+ if self.case_sensitive
+ else key._tq_label.lower()
+ )
in map_
):
result = map_[
- key._label if self.case_sensitive else key._label.lower()
+ key._tq_label
+ if self.case_sensitive
+ else key._tq_label.lower()
]
elif (
hasattr(key, "name")
if isinstance(col, expression.Label):
# new in 1.4, get column property against expressions
# to be addressable in subqueries
- col.key = col._key_label = key
+ col.key = col._tq_key_label = key
self.columns.add(col, key)
for col in prop.columns + prop._orig_columns:
from . import elements
from . import functions
from . import operators
-from . import roles
from . import schema
from . import selectable
from . import sqltypes
_loose_column_name_matching = False
"""tell the result object that the SQL staement is textual, wants to match
- up to Column objects, and may be using the ._label in the SELECT rather
+ up to Column objects, and may be using the ._tq_label in the SELECT rather
than the base name.
"""
if add_to_result_map is not None:
targets = (column, name, column.key) + result_map_targets
- if column._label:
- targets += (column._label,)
+ if column._tq_label:
+ targets += (column._tq_label,)
add_to_result_map(name, orig_name, targets, column.type)
)
self._result_columns.append((keyname, name, objects, type_))
+ def _label_returning_column(self, stmt, column, column_clause_args=None):
+ """Render a column with necessary labels inside of a RETURNING clause.
+
+ This method is provided for individual dialects in place of calling
+ the _label_select_column method directly, so that the two use cases
+ of RETURNING vs. SELECT can be disambiguated going forward.
+
+ .. versionadded:: 1.4.21
+
+ """
+ return self._label_select_column(
+ None,
+ column,
+ True,
+ False,
+ {} if column_clause_args is None else column_clause_args,
+ )
+
def _label_select_column(
self,
select,
asfrom,
column_clause_args,
name=None,
+ proxy_name=None,
+ fallback_label_name=None,
within_columns_clause=True,
column_is_repeated=False,
need_column_expressions=False,
else:
add_to_result_map = None
- if not within_columns_clause:
- result_expr = col_expr
- elif isinstance(column, elements.Label):
+ # this method is used by some of the dialects for RETURNING,
+ # which has different inputs. _label_returning_column was added
+ # as the better target for this now however for 1.4 we will keep
+ # _label_select_column directly compatible with this use case.
+ # these assertions right now set up the current expected inputs
+ assert within_columns_clause, (
+ "_label_select_column is only relevant within "
+ "the columns clause of a SELECT or RETURNING"
+ )
+
+ if isinstance(column, elements.Label):
if col_expr is not column:
result_expr = _CompileLabel(
col_expr, column.name, alt_names=(column.element,)
else:
result_expr = col_expr
- elif select is not None and name:
- result_expr = _CompileLabel(
- col_expr, name, alt_names=(column._key_label,)
- )
- elif (
- asfrom
- and isinstance(column, elements.ColumnClause)
- and not column.is_literal
- and column.table is not None
- and not isinstance(column.table, selectable.Select)
- ):
- result_expr = _CompileLabel(
- col_expr,
- coercions.expect(roles.TruncatedLabelRole, column.name),
- alt_names=(column.key,),
- )
- elif (
- not isinstance(column, elements.TextClause)
- and (
- not isinstance(column, elements.UnaryExpression)
- or column.wraps_column_expression
- or asfrom
- )
- and (
- not hasattr(column, "name")
- or isinstance(column, functions.FunctionElement)
- )
- ):
- result_expr = _CompileLabel(
- col_expr,
- column._anon_name_label
- if not column_is_repeated
- else column._dedupe_label_anon_label,
- )
- elif col_expr is not column:
- # TODO: are we sure "column" has a .name and .key here ?
- # assert isinstance(column, elements.ColumnClause)
+ elif name:
+ # here, _columns_plus_names has determined there's an explicit
+ # label name we need to use. this is the default for
+ # tablenames_plus_columnnames as well as when columns are being
+ # deduplicated on name
+
+ assert (
+ proxy_name is not None
+ ), "proxy_name is required if 'name' is passed"
+
result_expr = _CompileLabel(
col_expr,
- coercions.expect(roles.TruncatedLabelRole, column.name),
- alt_names=(column.key,),
+ name,
+ alt_names=(
+ proxy_name,
+ # this is a hack to allow legacy result column lookups
+ # to work as they did before; this goes away in 2.0.
+ # TODO: this only seems to be tested indirectly
+ # via test/orm/test_deprecations.py. should be a
+ # resultset test for this
+ column._tq_label,
+ ),
)
else:
- result_expr = col_expr
+ # determine here whether this column should be rendered in
+ # a labelled context or not, as we were given no required label
+ # name from the caller. Here we apply heuristics based on the kind
+ # of SQL expression involved.
+
+ if col_expr is not column:
+ # type-specific expression wrapping the given column,
+ # so we render a label
+ render_with_label = True
+ elif isinstance(column, elements.ColumnClause):
+ # table-bound column, we render its name as a label if we are
+ # inside of a subquery only
+ render_with_label = (
+ asfrom
+ and not column.is_literal
+ and column.table is not None
+ )
+ elif isinstance(column, elements.TextClause):
+ render_with_label = False
+ elif isinstance(column, elements.UnaryExpression):
+ render_with_label = column.wraps_column_expression or asfrom
+ elif (
+ # general class of expressions that don't have a SQL-column
+ # addressible name. includes scalar selects, bind parameters,
+ # SQL functions, others
+ not isinstance(column, elements.NamedColumn)
+ # deeper check that indicates there's no natural "name" to
+ # this element, which accommodates for custom SQL constructs
+ # that might have a ".name" attribute (but aren't SQL
+ # functions) but are not implementing this more recently added
+ # base class. in theory the "NamedColumn" check should be
+ # enough, however here we seek to maintain legacy behaviors
+ # as well.
+ and column._non_anon_label is None
+ ):
+ render_with_label = True
+ else:
+ render_with_label = False
+
+ if render_with_label:
+ if not fallback_label_name:
+ # used by the RETURNING case right now. we generate it
+ # here as 3rd party dialects may be referring to
+ # _label_select_column method directly instead of the
+ # just-added _label_returning_column method
+ assert not column_is_repeated
+ fallback_label_name = column._anon_name_label
+
+ fallback_label_name = (
+ elements._truncated_label(fallback_label_name)
+ if not isinstance(
+ fallback_label_name, elements._truncated_label
+ )
+ else fallback_label_name
+ )
+
+ result_expr = _CompileLabel(
+ col_expr, fallback_label_name, alt_names=(proxy_name,)
+ )
+ else:
+ result_expr = col_expr
column_clause_args.update(
within_columns_clause=within_columns_clause,
asfrom,
column_clause_args,
name=name,
+ proxy_name=proxy_name,
+ fallback_label_name=fallback_label_name,
column_is_repeated=repeated,
need_column_expressions=need_column_expressions,
)
- for name, column, repeated in compile_state.columns_plus_names
+ for (
+ name,
+ proxy_name,
+ fallback_label_name,
+ column,
+ repeated,
+ ) in compile_state.columns_plus_names
]
if c is not None
]
name
for (
key,
+ proxy_name,
+ fallback_label_name,
name,
repeated,
) in compile_state.columns_plus_names
name
for (
key,
+ proxy_name,
+ fallback_label_name,
name,
repeated,
) in compile_state_wraps_for.columns_plus_names
foreign_keys = []
_proxies = ()
- _label = None
+ _tq_label = None
"""The named label that can be used to target
- this column in a result set.
+ this column in a result set in a "table qualified" context.
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.
+ rendering <expr> AS <label> in a SELECT statement when using
+ the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the legacy
+ ORM ``Query`` object uses as well.
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.
+ .. versionchanged:: 1.4.21 renamed from ``._label``
+
"""
key = None
"""
- _key_label = None
- """A label-based version of 'key' that in some circumstances refers
- to this object in a Python namespace.
+ @HasMemoized.memoized_attribute
+ def _tq_key_label(self):
+ """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.
+ _tq_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.
- """
+ .. versionchanged:: 1.4.21 renamed from ``._key_label``
+
+ """
+ return self._proxy_key
+
+ @property
+ def _key_label(self):
+ """legacy; renamed to _tq_key_label"""
+ return self._tq_key_label
+
+ @property
+ def _label(self):
+ """legacy; renamed to _tq_label"""
+ return self._tq_label
+
+ @property
+ def _non_anon_label(self):
+ """the 'name' that naturally applies this element when rendered in
+ SQL.
+
+ Concretely, this is the "name" of a column or a label in a
+ SELECT statement; ``<columnname>`` and ``<labelname>`` below::
+
+ SELECT <columnmame> FROM table
+
+ SELECT column AS <labelname> FROM table
+
+ Above, the two names noted will be what's present in the DBAPI
+ ``cursor.description`` as the names.
+
+ If this attribute returns ``None``, it means that the SQL element as
+ written does not have a 100% fully predictable "name" that would appear
+ in the ``cursor.description``. Examples include SQL functions, CAST
+ functions, etc. While such things do return names in
+ ``cursor.description``, they are only predictable on a
+ database-specific basis; e.g. an expression like ``MAX(table.col)`` may
+ appear as the string ``max`` on one database (like PostgreSQL) or may
+ appear as the whole expression ``max(table.col)`` on SQLite.
+
+ The default implementation looks for a ``.name`` attribute on the
+ object, as has been the precedent established in SQLAlchemy for many
+ years. An exception is made on the ``FunctionElement`` subclass
+ so that the return value is always ``None``.
+
+ .. versionadded:: 1.4.21
+
+
+
+ """
+ return getattr(self, "name", None)
_render_label_in_columns_clause = True
"""A flag used by select._columns_plus_names that helps to determine
and other.name == self.name
)
- @util.memoized_property
+ @HasMemoized.memoized_attribute
def _proxy_key(self):
if self._annotations and "proxy_key" in self._annotations:
return self._annotations["proxy_key"]
- elif self.key:
- return self.key
+
+ name = self.key
+ if not name:
+ # there's a bit of a seeming contradiction which is that the
+ # "_non_anon_label" of a column can in fact be an
+ # "_anonymous_label"; this is when it's on a column that is
+ # proxying for an anonymous expression in a subquery.
+ name = self._non_anon_label
+
+ if isinstance(name, _anonymous_label):
+ return None
else:
- return getattr(self, "name", "_no_label")
+ return name
- @util.memoized_property
+ @HasMemoized.memoized_attribute
def _expression_label(self):
"""a suggested label to use in the case that the column has no name,
which should be used if possible as the explicit 'AS <label>'
where this expression would normally have an anon label.
+ this is essentially mostly what _proxy_key does except it returns
+ None if the column has a normal name that can be used.
+
"""
if getattr(self, "name", None) is not None:
@util.memoized_property
def _dedupe_anon_label(self):
- label = getattr(self, "name", None) or "anon"
- return self._anon_label(label + "_")
+ """label to apply to a column that is anon labeled, but repeated
+ in the SELECT, so that we have to make an "extra anon" label that
+ disambiguates it from the previous appearance.
+
+ these labels come out like "foo_bar_id__1" and have double underscores
+ in them.
+
+ """
+ label = getattr(self, "name", None)
+
+ # current convention is that if the element doesn't have a
+ # ".name" (usually because it is not NamedColumn), we try to
+ # use a "table qualified" form for the "dedupe anon" label,
+ # based on the notion that a label like
+ # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
+ # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
+
+ if label is None:
+ return self._dedupe_anon_tq_label
+ else:
+ return self._anon_label(label + "_")
@util.memoized_property
- def _label_anon_label(self):
- return self._anon_label(getattr(self, "_label", None))
+ def _anon_tq_label(self):
+ return self._anon_label(getattr(self, "_tq_label", None))
@util.memoized_property
- def _label_anon_key_label(self):
- return self._anon_label(getattr(self, "_key_label", None))
+ def _anon_tq_key_label(self):
+ return self._anon_label(getattr(self, "_tq_key_label", None))
@util.memoized_property
- def _dedupe_label_anon_label(self):
- label = getattr(self, "_label", None) or "anon"
+ def _dedupe_anon_tq_label(self):
+ label = getattr(self, "_tq_label", None) or "anon"
return self._anon_label(label + "_")
raise NotImplementedError()
@property
- def _label(self):
+ def _tq_label(self):
wce = self.wrapped_column_expression
- if hasattr(wce, "_label"):
- return wce._label
+ if hasattr(wce, "_tq_label"):
+ return wce._tq_label
else:
return None
+ _label = _tq_label
+
+ @property
+ def _non_anon_label(self):
+ return None
+
@property
def _anon_name_label(self):
wce = self.wrapped_column_expression
- if hasattr(wce, "name"):
- return wce.name
- elif hasattr(wce, "_anon_name_label"):
- return wce._anon_name_label
+
+ # this logic tries to get the WrappedColumnExpression to render
+ # with "<expr> AS <name>", where "<name>" is the natural name
+ # within the expression itself. e.g. "CAST(table.foo) AS foo".
+ if not wce._is_text_clause:
+ nal = wce._non_anon_label
+ if nal:
+ return nal
+ elif hasattr(wce, "_anon_name_label"):
+ return wce._anon_name_label
+ return super(WrapsColumnExpression, self)._anon_name_label
+
+ @property
+ def _dedupe_anon_label(self):
+ wce = self.wrapped_column_expression
+ nal = wce._non_anon_label
+ if nal:
+ return self._anon_label(nal + "_")
else:
- return super(WrapsColumnExpression, self)._anon_name_label
+ return self._dedupe_anon_tq_label
class BindParameter(roles.InElementRole, ColumnElement):
return self.element._is_implicitly_boolean
@property
- def _key_label(self):
- return self._label
-
- @property
- def _label(self):
- return getattr(self.element, "_label", None) or self._anon_name_label
+ def _tq_label(self):
+ return (
+ getattr(self.element, "_tq_label", None) or self._anon_name_label
+ )
@property
def _proxies(self):
id(self), getattr(element, "name", "anon")
)
- self.key = self._label = self._key_label = self.name
+ self.key = self._tq_label = self._tq_key_label = self.name
self._element = element
self._type = type_
self._proxies = [element]
self.name = self._resolve_label = _anonymous_label.safe_construct(
id(self), getattr(self.element, "name", "anon")
)
- self.key = self._label = self._key_label = self.name
+ self.key = self._tq_label = self._tq_key_label = self.name
@property
def _from_objects(self):
return self.name.encode("ascii", "backslashreplace")
@HasMemoized.memoized_attribute
- def _key_label(self):
+ def _tq_key_label(self):
+ """table qualified label based on column key.
+
+ for table-bound columns this is <tablename>_<column key/proxy key>;
+
+ all other expressions it resolves to key/proxy key.
+
+ """
proxy_key = self._proxy_key
- if proxy_key != self.name:
- return self._gen_label(proxy_key)
+ if proxy_key and proxy_key != self.name:
+ return self._gen_tq_label(proxy_key)
else:
- return self._label
+ return self._tq_label
@HasMemoized.memoized_attribute
- def _label(self):
- return self._gen_label(self.name)
+ def _tq_label(self):
+ """table qualified label based on column name.
+
+ for table-bound columns this is <tablename>_<columnname>; all other
+ expressions it resolves to .name.
+
+ """
+ return self._gen_tq_label(self.name)
@HasMemoized.memoized_attribute
def _render_label_in_columns_clause(self):
return True
- def _gen_label(self, name, dedupe_on_key=True):
+ @HasMemoized.memoized_attribute
+ def _non_anon_label(self):
+ return self.name
+
+ def _gen_tq_label(self, name, dedupe_on_key=True):
return name
def _bind_param(self, operator, obj, type_=None, expanding=False):
@property
def _ddl_label(self):
- return self._gen_label(self.name, dedupe_on_key=False)
+ return self._gen_tq_label(self.name, dedupe_on_key=False)
def _compare_name_for_result(self, other):
if (
)
):
return (hasattr(other, "name") and self.name == other.name) or (
- hasattr(other, "_label") and self._label == other._label
+ hasattr(other, "_tq_label")
+ and self._tq_label == other._tq_label
)
else:
return other.proxy_set.intersection(self.proxy_set)
- def _gen_label(self, name, dedupe_on_key=True):
+ def _gen_tq_label(self, name, dedupe_on_key=True):
+ """generate table-qualified label
+
+ for a table-bound column this is <tablename>_<columnname>.
+
+ used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
+ as well as the .columns collection on a Join object.
+
+ """
t = self.table
if self.is_literal:
return None
class AnnotatedColumnElement(Annotated):
def __init__(self, element, values):
Annotated.__init__(self, element, values)
- for attr in ("comparator", "_proxy_key", "_key_label"):
+ for attr in (
+ "comparator",
+ "_proxy_key",
+ "_tq_key_label",
+ "_tq_label",
+ "_non_anon_label",
+ ):
self.__dict__.pop(attr, None)
for attr in ("name", "key", "table"):
if self.__dict__.get(attr, False) is None:
operator=operators.comma_op, group_contents=True, *args
).self_group()
+ _non_anon_label = None
+
+ @property
+ def _proxy_key(self):
+ return super(FunctionElement, self)._proxy_key or getattr(
+ self, "name", None
+ )
+
def _execute_on_connection(
self, connection, multiparams, params, execution_options
):
)
)
self._columns._populate_separate_keys(
- (col._key_label, col) for col in columns
+ (col._tq_key_label, col) for col in columns
)
self.foreign_keys.update(
itertools.chain(*[col.foreign_keys for col in columns])
@classmethod
def _column_naming_convention(cls, label_style):
- # note: these functions won't work for TextClause objects,
- # which should be omitted when iterating through
- # _raw_columns.
- if label_style is LABEL_STYLE_NONE:
+ table_qualified = label_style is LABEL_STYLE_TABLENAME_PLUS_COL
+ dedupe = label_style is not LABEL_STYLE_NONE
- def go(c, col_name=None):
- return c._proxy_key
+ pa = prefix_anon_map()
+ names = set()
- elif label_style is LABEL_STYLE_TABLENAME_PLUS_COL:
- names = set()
- pa = [] # late-constructed as needed, python 2 has no "nonlocal"
-
- def go(c, col_name=None):
- # we use key_label since this name is intended for targeting
- # within the ColumnCollection only, it's not related to SQL
- # rendering which always uses column name for SQL label names
-
- name = c._key_label
-
- if name in names:
- if not pa:
- pa.append(prefix_anon_map())
-
- name = c._label_anon_key_label % pa[0]
- else:
- names.add(name)
+ def go(c, col_name=None):
+ if c._is_text_clause:
+ return None
+ elif not dedupe:
+ name = c._proxy_key
+ if name is None:
+ name = "_no_label"
return name
- else:
- names = set()
- pa = [] # late-constructed as needed, python 2 has no "nonlocal"
+ name = c._tq_key_label if table_qualified else c._proxy_key
- def go(c, col_name=None):
- name = c._proxy_key
+ if name is None:
+ name = "_no_label"
if name in names:
- if not pa:
- pa.append(prefix_anon_map())
- name = c._anon_key_label % pa[0]
+ return c._anon_label(name) % pa
else:
names.add(name)
+ return name
+ elif name in names:
+ return (
+ c._anon_tq_key_label % pa
+ if table_qualified
+ else c._anon_key_label % pa
+ )
+ else:
+ names.add(name)
return name
return go
def _memoized_attr__label_resolve_dict(self):
with_cols = dict(
- (c._resolve_label or c._label or c.key, c)
+ (c._resolve_label or c._tq_label or c.key, c)
for c in self.statement._all_selected_columns
if c._allow_label_resolve
)
"""Generate column names as rendered in a SELECT statement by
the compiler.
- This is distinct from other name generators that are intended for
- population of .c collections and similar, which may have slightly
- different rules.
+ This is distinct from the _column_naming_convention generator that's
+ intended for population of .c collections and similar, which has
+ different rules. the collection returned here calls upon the
+ _column_naming_convention as well.
"""
cols = self._all_selected_columns
- # when use_labels is on:
- # in all cases == if we see the same label name, use _label_anon_label
- # for subsequent occurrences of that label
- #
- # anon_for_dupe_key == if we see the same column object multiple
- # times under a particular name, whether it's the _label name or the
- # anon label, apply _dedupe_label_anon_label to the subsequent
- # occurrences of it.
- if self._label_style is LABEL_STYLE_NONE:
- # don't generate any labels
- same_cols = set()
+ key_naming_convention = SelectState._column_naming_convention(
+ self._label_style
+ )
- return [
- (None, c, c in same_cols or same_cols.add(c)) for c in cols
- ]
- else:
- names = {}
+ names = {}
- use_tablename_labels = (
- self._label_style is LABEL_STYLE_TABLENAME_PLUS_COL
- )
+ result = []
+ result_append = result.append
- def name_for_col(c):
- if not c._render_label_in_columns_clause:
- return (None, c, False)
- elif use_tablename_labels:
- if c._label is None:
- repeated = c._anon_name_label in names
- names[c._anon_name_label] = c
- return (None, c, repeated)
- else:
- name = effective_name = c._label
- elif getattr(c, "name", None) is None:
- # this is a scalar_select(). need to improve this case
+ table_qualified = self._label_style is LABEL_STYLE_TABLENAME_PLUS_COL
+ label_style_none = self._label_style is LABEL_STYLE_NONE
+
+ for c in cols:
+ repeated = False
+
+ if not c._render_label_in_columns_clause:
+ effective_name = (
+ required_label_name
+ ) = fallback_label_name = None
+ elif label_style_none:
+ effective_name = required_label_name = None
+ fallback_label_name = c._non_anon_label or c._anon_name_label
+ else:
+ if table_qualified:
+ required_label_name = (
+ effective_name
+ ) = fallback_label_name = c._tq_label
+ else:
+ effective_name = fallback_label_name = c._non_anon_label
+ required_label_name = None
+
+ if effective_name is None:
+ # it seems like this could be _proxy_key and we would
+ # not need _expression_label but it isn't
+ # giving us a clue when to use anon_label instead
expr_label = c._expression_label
if expr_label is None:
repeated = c._anon_name_label in names
names[c._anon_name_label] = c
- return (None, c, repeated)
- else:
- name = effective_name = expr_label
- else:
- name = None
- effective_name = c.name
+ effective_name = required_label_name = None
- repeated = False
+ if repeated:
+ # here, "required_label_name" is sent as
+ # "None" and "fallback_label_name" is sent.
+ if table_qualified:
+ fallback_label_name = c._dedupe_anon_tq_label
+ else:
+ fallback_label_name = c._dedupe_anon_label
+ else:
+ fallback_label_name = c._anon_name_label
+ else:
+ required_label_name = (
+ effective_name
+ ) = fallback_label_name = expr_label
+ if effective_name is not None:
if effective_name in names:
# when looking to see if names[name] is the same column as
# c, use hash(), so that an annotated version of the column
# different column under the same name. apply
# disambiguating label
- if use_tablename_labels:
- name = c._label_anon_label
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._anon_tq_label
else:
- name = c._anon_name_label
+ required_label_name = (
+ fallback_label_name
+ ) = c._anon_name_label
- if anon_for_dupe_key and name in names:
- # here, c._label_anon_label is definitely unique to
+ if anon_for_dupe_key and required_label_name in names:
+ # here, c._anon_tq_label is definitely unique to
# that column identity (or annotated version), so
# this should always be true.
# this is also an infrequent codepath because
# you need two levels of duplication to be here
- assert hash(names[name]) == hash(c)
+ assert hash(names[required_label_name]) == hash(c)
# the column under the disambiguating label is
# already present. apply the "dedupe" label to
# subsequent occurrences of the column so that the
# original stays non-ambiguous
- if use_tablename_labels:
- name = c._dedupe_label_anon_label
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_tq_label
else:
- name = c._dedupe_anon_label
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_label
repeated = True
else:
- names[name] = c
+ names[required_label_name] = c
elif anon_for_dupe_key:
# same column under the same name. apply the "dedupe"
# label so that the original stays non-ambiguous
- if use_tablename_labels:
- name = c._dedupe_label_anon_label
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_tq_label
else:
- name = c._dedupe_anon_label
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_label
repeated = True
else:
names[effective_name] = c
- return name, c, repeated
- return [name_for_col(c) for c in cols]
+ result_append(
+ (
+ # string label name, if non-None, must be rendered as a
+ # label, i.e. "AS <name>"
+ required_label_name,
+ # proxy_key that is to be part of the result map for this
+ # col. this is also the key in a fromclause.c or
+ # select.selected_columns collection
+ key_naming_convention(c),
+ # name that can be used to render an "AS <name>" when
+ # we have to render a label even though
+ # required_label_name was not given
+ fallback_label_name,
+ # the ColumnElement itself
+ c,
+ # True if this is a duplicate of a previous column
+ # in the list of columns
+ repeated,
+ )
+ )
+
+ return result
def _generate_fromclause_column_proxies(self, subquery):
"""Generate column proxies to place in the exported ``.c``
collection of a subquery."""
- keys_seen = set()
- prox = []
-
- pa = None
-
- tablename_plus_col = (
- self._label_style is LABEL_STYLE_TABLENAME_PLUS_COL
- )
- disambiguate_only = self._label_style is LABEL_STYLE_DISAMBIGUATE_ONLY
-
- for name, c, repeated in self._generate_columns_plus_names(False):
- if c._is_text_clause:
- continue
- elif tablename_plus_col:
- key = c._key_label
- if key is not None and key in keys_seen:
- if pa is None:
- pa = prefix_anon_map()
- key = c._label_anon_key_label % pa
- keys_seen.add(key)
- elif disambiguate_only:
- key = c._proxy_key
- if key is not None and key in keys_seen:
- if pa is None:
- pa = prefix_anon_map()
- key = c._anon_key_label % pa
- keys_seen.add(key)
- else:
- key = c._proxy_key
- prox.append(
- c._make_proxy(
- subquery, key=key, name=name, name_is_truncatable=True
- )
+ prox = [
+ c._make_proxy(
+ subquery,
+ key=proxy_key,
+ name=required_label_name,
+ name_is_truncatable=True,
)
+ for (
+ required_label_name,
+ proxy_key,
+ fallback_label_name,
+ c,
+ repeated,
+ ) in (self._generate_columns_plus_names(False))
+ if not c._is_text_clause
+ ]
+
subquery._columns._populate_separate_keys(prox)
def _needs_parens_for_grouping(self):
{"c2": "some data", "c1": 5},
(5, "some data"),
),
- ("SELECT lower", {"lower_1": "Foo"}, ("Foo",)),
+ ("SELECT lower", {"lower_2": "Foo"}, ("Foo",)),
(
"INSERT INTO t1 (c1, c2)",
{"c2": "foo", "c1": 6},
self.assert_compile(
stmt,
"SELECT a.id AS a_id, a.firstname || :firstname_1 || "
- "a.lastname AS anon_1 FROM a",
+ "a.lastname AS name FROM a",
)
- # but no ORM translate...
eq_(stmt.subquery().c.keys(), ["a_id", "name"])
- # then it comes out like this, not really sure if this is useful
self.assert_compile(
select(stmt.subquery()),
- "SELECT anon_1.a_id, anon_1.anon_2 FROM (SELECT a.id AS a_id, "
- "a.firstname || :firstname_1 || a.lastname AS anon_2 FROM a) "
+ "SELECT anon_1.a_id, anon_1.name FROM (SELECT a.id AS a_id, "
+ "a.firstname || :firstname_1 || a.lastname AS name FROM a) "
"AS anon_1",
)
stmt = sess.query(A.id, A.name)
- # TABLENAME_PLUS_COL uses anon label right now, this is a little
- # awkward looking, but loading.py translates
self.assert_compile(
stmt,
"SELECT a.id AS a_id, a.firstname || "
- ":firstname_1 || a.lastname AS anon_1 FROM a",
+ ":firstname_1 || a.lastname AS name FROM a",
)
# for the subquery, we lose the "ORM-ness" from the subquery
pass
if False:
+ # this conditional creates the table each time which would
+ # eliminate cross-test memoization issues. if the tests
+ # are failing without this then there's a memoization issue.
+ # check AnnotatedColumn memoized keys
m = MetaData()
users = Table(
"users",
m,
Column("id", Integer, primary_key=True),
- Column("name", String, key="uname"),
+ Column(
+ "name",
+ String,
+ ),
)
- mapper(Foo, users, properties={"uname": users.c.uname})
+ mapper(Foo, users, properties={"uname": users.c.name})
else:
users = self.tables.users
mapper(Foo, users, properties={"uname": users.c.name})
# TEST: test.aaa_profiling.test_compiler.CompileTest.test_select_labels
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mariadb_mysqldb_dbapiunicode_cextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mariadb_mysqldb_dbapiunicode_nocextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mariadb_pymysql_dbapiunicode_cextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mariadb_pymysql_dbapiunicode_nocextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mssql_pyodbc_dbapiunicode_cextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mssql_pyodbc_dbapiunicode_nocextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mysql_mysqldb_dbapiunicode_cextensions 179
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mysql_mysqldb_dbapiunicode_nocextensions 179
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mysql_pymysql_dbapiunicode_cextensions 179
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_mysql_pymysql_dbapiunicode_nocextensions 179
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_oracle_cx_oracle_dbapiunicode_cextensions 170
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_oracle_cx_oracle_dbapiunicode_nocextensions 182
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_postgresql_psycopg2_dbapiunicode_cextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_postgresql_psycopg2_dbapiunicode_nocextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_sqlite_pysqlite_dbapiunicode_cextensions 197
-test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_2.7_sqlite_pysqlite_dbapiunicode_nocextensions 197
test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_3.9_mariadb_mysqldb_dbapiunicode_cextensions 212
test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_3.9_mariadb_mysqldb_dbapiunicode_nocextensions 212
test.aaa_profiling.test_compiler.CompileTest.test_select_labels x86_64_linux_cpython_3.9_mariadb_pymysql_dbapiunicode_cextensions 212
"foo_bar.id AS foo_bar_id__2 " # 6. 3rd foo_bar.id, same as 5
"FROM foo, foo_bar",
)
+ eq_(
+ stmt.selected_columns.keys(),
+ [
+ "foo_id",
+ "foo_bar_id",
+ "foo_bar_id_1",
+ "foo_bar_id_2",
+ "foo_id_1",
+ "foo_bar_id_2",
+ "foo_bar_id_1",
+ "foo_bar_id_1",
+ ],
+ )
# for the subquery, the labels created for repeated occurrences
# of the same column are not used. only the label applied to the
") AS anon_1",
)
+ def test_overlapping_labels_plus_dupes_separate_keys_use_labels(self):
+ """test a condition related to #6710.
+
+ prior to this issue CTE uses selected_columns to render the
+ "WITH RECURSIVE (colnames)" part. This test shows that this isn't
+ correct when keys are present.
+
+ """
+ m = MetaData()
+ foo = Table(
+ "foo",
+ m,
+ Column("id", Integer),
+ Column("bar_id", Integer, key="bb"),
+ )
+ foo_bar = Table("foo_bar", m, Column("id", Integer, key="bb"))
+
+ stmt = select(
+ foo.c.id,
+ foo.c.bb,
+ foo_bar.c.bb,
+ foo.c.bb,
+ foo.c.id,
+ foo.c.bb,
+ foo_bar.c.bb,
+ foo_bar.c.bb,
+ ).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+
+ # note these keys are not what renders in the SQL. These keys
+ # will be addressable in the result set but can't be used in
+ # rendering, such as for a CTE
+ eq_(
+ stmt.selected_columns.keys(),
+ [
+ "foo_id",
+ "foo_bb",
+ "foo_bar_bb",
+ "foo_bb_1",
+ "foo_id_1",
+ "foo_bb_1",
+ "foo_bar_bb_1",
+ "foo_bar_bb_1",
+ ],
+ )
+ eq_(
+ stmt.subquery().c.keys(),
+ [
+ "foo_id",
+ "foo_bb",
+ "foo_bar_bb",
+ "foo_bb_1",
+ "foo_id_1",
+ "foo_bb_1",
+ "foo_bar_bb_1",
+ "foo_bar_bb_1",
+ ],
+ )
+ self.assert_compile(
+ stmt,
+ "SELECT foo.id AS foo_id, "
+ "foo.bar_id AS foo_bar_id, " # 1. 1st foo.bar_id, as is
+ "foo_bar.id AS foo_bar_id_1, " # 2. 1st foo_bar.id, disamb from 1
+ "foo.bar_id AS foo_bar_id__1, " # 3. 2nd foo.bar_id, dedupe from 1
+ "foo.id AS foo_id__1, "
+ "foo.bar_id AS foo_bar_id__1, " # 4. 3rd foo.bar_id, same as 3
+ "foo_bar.id AS foo_bar_id__2, " # 5. 2nd foo_bar.id
+ "foo_bar.id AS foo_bar_id__2 " # 6. 3rd foo_bar.id, same as 5
+ "FROM foo, foo_bar",
+ )
+
def test_dupe_columns_use_labels(self):
t = table("t", column("a"), column("b"))
self.assert_compile(
# coverage on other dialects.
sel = select(tbl, cast(tbl.c.v1, Numeric)).compile(dialect=dialect)
- # TODO: another unusual result from disambiguate only
+ # TODO: another unusual result from disambiguate only:
+ # v1__1 vs v1_1 are due to the special meaning
+ # WrapsColumnExpression gives to the "_anon_name_label" attribute,
+ # where it tries to default to a label name that matches that of
+ # the column within.
+
if isinstance(dialect, type(mysql.dialect())):
eq_(
str(sel),
"SELECT casttest.id, casttest.v1, casttest.v2, "
"casttest.ts, "
- "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1 \n"
+ "CAST(casttest.v1 AS DECIMAL) AS v1__1 \n"
"FROM casttest",
)
else:
str(sel),
"SELECT casttest.id, casttest.v1, casttest.v2, "
"casttest.ts, CAST(casttest.v1 AS NUMERIC) AS "
- "casttest_v1__1 \nFROM casttest",
+ "v1__1 \nFROM casttest",
)
sel = (
.set_label_style(LABEL_STYLE_NONE)
.compile(dialect=dialect)
)
+ # label style none - dupes v1
if isinstance(dialect, type(mysql.dialect())):
eq_(
str(sel),
'"some wacky thing"',
"",
),
- (exprs[3], exprs[3].key, ":param_1", "anon_1"),
+ (
+ exprs[3],
+ "_no_label",
+ ":param_1",
+ "anon_1",
+ ),
):
if getattr(col, "table", None) is not None:
t = col.table
"bar": ("bar", (l1, "bar"), l1.type, 1),
"anon_1": (
tc._anon_name_label,
- (tc_anon_label, "anon_1", tc),
+ (tc_anon_label, "anon_1", tc, "_no_label"),
tc.type,
2,
),
expr(table1.c.name),
),
"SELECT some_table.name, some_table.name AS name__1, "
- "SOME_COL_THING(some_table.name) AS some_table_name__1, "
- "SOME_COL_THING(some_table.name) AS some_table_name__2 "
+ "SOME_COL_THING(some_table.name) AS name__2, "
+ "SOME_COL_THING(some_table.name) AS name__3 "
"FROM some_table",
)
table1.c.name,
),
"SELECT CAST(some_table.name AS INTEGER) AS name, "
- "CAST(some_table.name AS VARCHAR) AS some_table_name__1, "
+ "CAST(some_table.name AS VARCHAR) AS name__1, "
"some_table.name AS name_1 FROM some_table",
)
),
# ideally type_coerce wouldn't label at all...
"SELECT some_table.name AS name, "
- "some_table.name AS some_table_name__1, "
+ "some_table.name AS name__1, "
"some_table.name AS name_1 FROM some_table",
)
@compiles(not_named_max)
def visit_max(element, compiler, **kw):
# explicit add
- kw["add_to_result_map"](None, None, (element,), NULLTYPE)
+ if "add_to_result_map" in kw:
+ kw["add_to_result_map"](None, None, (element,), NULLTYPE)
return "max(a)"
# assert that there is no "AS max_" or any label of any kind.
def test_labels_anon_w_separate_key_subquery(self):
label = select(table1.c.col1).label(None)
- label.key = label._key_label = "bar"
+ label.key = label._tq_key_label = "bar"
s1 = select(label)
def test_labels_anon_generate_binds_subquery(self):
label = select(table1.c.col1).label(None)
- label.key = label._key_label = "bar"
+ label.key = label._tq_key_label = "bar"
s1 = select(label)
table1.c.col1 == 10,
func.count(table1.c.col1),
literal_column("x"),
- ).subquery()
+ )
+
+ # the reason we return "_no_label" is because we dont have a system
+ # right now that is guaranteed to use the identical label in
+ # selected_columns as will be used when we compile the statement, and
+ # this includes the creation of _result_map right now which gets loaded
+ # with lots of unprocessed anon symbols for these kinds of cases,
+ # and we don't have a fully comprehensive approach for this to always
+ # do the right thing; as it is *vastly* simpler for the user to please
+ # use a label(), "_no_label" is meant to encourage this rather than
+ # relying on a system that we don't fully have on this end.
+ eq_(s1.subquery().c.keys(), ["_no_label", "_no_label_1", "count", "x"])
+
+ self.assert_compile(
+ s1,
+ "SELECT table1.col1 = :col1_1 AS anon_1, "
+ "table1.col1 = :col1_2 AS anon_2, count(table1.col1) AS count_1, "
+ "x FROM table1",
+ )
+ eq_(
+ s1.selected_columns.keys(),
+ ["_no_label", "_no_label_1", "count", "x"],
+ )
- eq_(s1.c.keys(), ["_no_label", "_no_label_1", "count", "x"])
+ eq_(
+ select(s1.subquery()).selected_columns.keys(),
+ ["_no_label", "_no_label_1", "_no_label_2", "x"],
+ )
+
+ self.assert_compile(
+ select(s1.subquery()),
+ "SELECT anon_2.anon_1, anon_2.anon_3, anon_2.count_1, anon_2.x "
+ "FROM (SELECT table1.col1 = :col1_1 AS anon_1, "
+ "table1.col1 = :col1_2 AS anon_3, "
+ "count(table1.col1) AS count_1, x FROM table1) AS anon_2",
+ )
def test_union_alias_dupe_keys(self):
s1 = select(table1.c.col1, table1.c.col2, table2.c.col1)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
