from .. import exc
from .. import util
from ..util import collections_abc
+from ..util import immutabledict
def connection_memoize(key):
def _distill_params_20(params):
+ # TODO: this has to be in C
if params is None:
return _no_tuple, _no_kw, []
- elif isinstance(params, collections_abc.MutableSequence): # list
+ elif isinstance(params, list):
+ # collections_abc.MutableSequence): # avoid abc.__instancecheck__
if params and not isinstance(
params[0], (collections_abc.Mapping, tuple)
):
return tuple(params), _no_kw, params
elif isinstance(
params,
- (collections_abc.Sequence, collections_abc.Mapping), # tuple or dict
+ (tuple, dict, immutabledict),
+ # avoid abc.__instancecheck__
+ # (collections_abc.Sequence, collections_abc.Mapping),
):
return _no_tuple, params, [params]
else:
class _ColumnEntity(_QueryEntity):
- __slots__ = ()
+ __slots__ = ("_fetch_column", "_row_processor")
@classmethod
def _for_columns(cls, compile_state, columns, parent_bundle=None):
def use_id_for_hash(self):
return not self.column.type.hashable
+ def row_processor(self, context, result):
+ compile_state = context.compile_state
+
+ # the resulting callable is entirely cacheable so just return
+ # it if we already made one
+ if self._row_processor is not None:
+ return self._row_processor
+
+ # retrieve the column that would have been set up in
+ # setup_compile_state, to avoid doing redundant work
+ if self._fetch_column is not None:
+ column = self._fetch_column
+ else:
+ # fetch_column will be None when we are doing a from_statement
+ # and setup_compile_state may not have been called.
+ column = self.column
+
+ # previously, the RawColumnEntity didn't look for from_obj_alias
+ # however I can't think of a case where we would be here and
+ # we'd want to ignore it if this is the from_statement use case.
+ # it's not really a use case to have raw columns + from_statement
+ if compile_state._from_obj_alias:
+ column = compile_state._from_obj_alias.columns[column]
+
+ if column._annotations:
+ # annotated columns perform more slowly in compiler and
+ # result due to the __eq__() method, so use deannotated
+ column = column._deannotate()
+
+ if compile_state.compound_eager_adapter:
+ column = compile_state.compound_eager_adapter.columns[column]
+
+ getter = result._getter(column)
+
+ ret = getter, self._label_name, self._extra_entities
+ self._row_processor = ret
+ return ret
+
class _RawColumnEntity(_ColumnEntity):
entity_zero = None
self.column._from_objects[0] if self.column._from_objects else None
)
self._extra_entities = (self.expr, self.column)
+ self._fetch_column = self._row_processor = None
def corresponds_to(self, entity):
return False
- def row_processor(self, context, result):
- if ("fetch_column", self) in context.attributes:
- column = context.attributes[("fetch_column", self)]
- else:
- column = self.column
-
- if column._annotations:
- # annotated columns perform more slowly in compiler and
- # result due to the __eq__() method, so use deannotated
- column = column._deannotate()
-
- compile_state = context.compile_state
- if compile_state.compound_eager_adapter:
- column = compile_state.compound_eager_adapter.columns[column]
-
- getter = result._getter(column)
- return getter, self._label_name, self._extra_entities
-
def setup_compile_state(self, compile_state):
current_adapter = compile_state._get_current_adapter()
if current_adapter:
column = column._deannotate()
compile_state.primary_columns.append(column)
- compile_state.attributes[("fetch_column", self)] = column
+ self._fetch_column = column
class _ORMColumnEntity(_ColumnEntity):
compile_state._has_orm_entities = True
self.column = column
+ self._fetch_column = self._row_processor = None
self._extra_entities = (self.expr, self.column)
self.entity_zero
) and entity.common_parent(self.entity_zero)
- def row_processor(self, context, result):
- compile_state = context.compile_state
-
- if ("fetch_column", self) in context.attributes:
- column = context.attributes[("fetch_column", self)]
- else:
- column = self.column
- if compile_state._from_obj_alias:
- column = compile_state._from_obj_alias.columns[column]
-
- if column._annotations:
- # annotated columns perform more slowly in compiler and
- # result due to the __eq__() method, so use deannotated
- column = column._deannotate()
-
- if compile_state.compound_eager_adapter:
- column = compile_state.compound_eager_adapter.columns[column]
-
- getter = result._getter(column)
- return getter, self._label_name, self._extra_entities
-
def setup_compile_state(self, compile_state):
current_adapter = compile_state._get_current_adapter()
if current_adapter:
compile_state._fallback_from_clauses.append(ezero.selectable)
compile_state.primary_columns.append(column)
-
- compile_state.attributes[("fetch_column", self)] = column
+ self._fetch_column = column
self.assert_sql_count(testing.db, go, 1)
+ def test_textual_select_orm_columns(self):
+ # test that columns using column._label match, as well as that
+ # ordering doesn't matter.
+ User = self.classes.User
+ Address = self.classes.Address
+ users = self.tables.users
+ addresses = self.tables.addresses
+
+ s = create_session()
+ q = s.query(User.name, User.id, Address.id).from_statement(
+ text(
+ "select users.name AS users_name, users.id AS users_id, "
+ "addresses.id AS addresses_id FROM users JOIN addresses "
+ "ON users.id = addresses.user_id WHERE users.id=8 "
+ "ORDER BY addresses.id"
+ ).columns(users.c.name, users.c.id, addresses.c.id)
+ )
+
+ eq_(q.all(), [("ed", 8, 2), ("ed", 8, 3), ("ed", 8, 4)])
+
@testing.combinations(
(
False,
[User(id=7), User(id=8), User(id=9), User(id=10)],
)
+ def test_columns_via_textasfrom_from_statement(self):
+ User = self.classes.User
+ s = create_session()
+
+ eq_(
+ s.query(User.id, User.name)
+ .from_statement(
+ text("select * from users order by id").columns(
+ id=Integer, name=String
+ )
+ )
+ .all(),
+ [(7, "jack"), (8, "ed"), (9, "fred"), (10, "chuck")],
+ )
+
def test_via_textasfrom_use_mapped_columns(self):
User = self.classes.User
s = create_session()