need autoflush w pre-attached object.
[ticket:2464]
+ - [feature] ORM entities can be passed
+ to select() as well as the select_from(),
+ correlate(), and correlate_except()
+ methods, where they will be unwrapped
+ into selectables. [ticket:2245]
+
- [feature] The registry of classes
in declarative_base() is now a
WeakValueDictionary. So subclasses of
)
-class QueryableAttribute(interfaces.PropComparator):
+class QueryableAttribute(interfaces._InspectionAttr, interfaces.PropComparator):
"""Base class for class-bound attributes. """
+ is_attribute = True
+
def __init__(self, class_, key, impl=None,
comparator=None, parententity=None,
of_type=None):
# TODO: conditionally attach this method based on clause_element ?
return self
+ @property
+ def expression(self):
+ return self.comparator.__clause_element__()
+
def __clause_element__(self):
return self.comparator.__clause_element__()
def property(self):
return self.comparator.property
-
-@inspection._inspects(QueryableAttribute)
-def _get_prop(source):
- return source.property
+inspection._self_inspects(QueryableAttribute)
class InstrumentedAttribute(QueryableAttribute):
"""Class bound instrumented attribute which adds descriptor methods."""
is_instance = False
is_mapper = False
is_property = False
+ is_attribute = False
+ is_clause_element = False
class MapperProperty(_InspectionAttr):
"""Manage the relationship of a ``Mapper`` to a single class
else:
self.strategy_class = strategies.ColumnLoader
+
@property
def expression(self):
"""Return the primary column or expression for this ColumnProperty.
)
from .util import (
AliasedClass, ORMAdapter, _entity_descriptor, PathRegistry,
- _is_aliased_class, _is_mapped_class, _orm_columns, _orm_selectable,
+ _is_aliased_class, _is_mapped_class, _orm_columns,
join as orm_join,with_parent, aliased
)
-from .. import sql, util, log, exc as sa_exc, inspect
+from .. import sql, util, log, exc as sa_exc, inspect, inspection
+from ..sql.expression import _interpret_as_from
from ..sql import (
util as sql_util,
expression, visitors
return self.enable_eagerloads(False).statement.as_scalar()
+ @property
+ def selectable(self):
+ return self.__clause_element__()
def __clause_element__(self):
return self.enable_eagerloads(False).with_labels().statement
"""
self._correlate = self._correlate.union(
- _orm_selectable(s)
+ _interpret_as_from(s)
+ if s is not None else None
for s in args)
@_generative()
statement.append_order_by(*context.eager_order_by)
return statement
-
def _adjust_for_single_inheritance(self, context):
"""Apply single-table-inheritance filtering.
def __str__(self):
return str(self._compile_context().statement)
+inspection._self_inspects(Query)
class _QueryEntity(object):
"""represent an entity column returned within a Query result."""
else:
return [entity]
-def _orm_selectable(entity):
- insp = inspection.inspect(entity, False)
- if hasattr(insp, 'selectable'):
- return insp.selectable
- else:
- return entity
-
def has_identity(object):
state = attributes.instance_state(object)
return state.has_identity
"SQL expression object or string expected."
)
+def _interpret_as_from(element):
+ insp = inspection.inspect(element, raiseerr=False)
+ if insp is None:
+ if isinstance(element, (util.NoneType, bool)):
+ return _const_expr(element)
+ elif isinstance(element, basestring):
+ return TextClause(unicode(element))
+ elif hasattr(insp, "selectable"):
+ return insp.selectable
+ else:
+ raise exc.ArgumentError("FROM expression expected")
+
+
def _const_expr(element):
if element is None:
return null()
return element
def _literal_as_column(element):
- if isinstance(element, Visitable):
- return element
- elif hasattr(element, '__clause_element__'):
- return element.__clause_element__()
- else:
- return literal_column(str(element))
+ insp = inspection.inspect(element, raiseerr=False)
+ if insp is not None:
+ if hasattr(insp, "expression"):
+ return insp.expression
+ elif hasattr(insp, "selectable"):
+ return insp.selectable
+ elif insp.is_clause_element:
+ return insp
+ return literal_column(str(element))
def _literal_as_binds(element, name=None, type_=None):
if hasattr(element, '__clause_element__'):
bind = None
_is_clone_of = None
is_selectable = False
+ is_clause_element = True
def _clone(self):
"""Create a shallow copy of this ClauseElement.
_key_label = None
_alt_names = ()
+ @property
+ def expression(self):
+ """Return a column expression.
+
+ Part of the inspection interface; returns self.
+
+ """
+ return self
+
@property
def _select_iterable(self):
return (self, )
def _select_iterable(self):
return (self,)
+ @property
+ def selectable(self):
+ return self
+
_hide_froms = []
def __init__(
if fromclauses and fromclauses[0] is None:
self._correlate = ()
else:
- self._correlate = set(self._correlate).union(fromclauses)
+ self._correlate = set(self._correlate).union(
+ _interpret_as_from(f) for f in fromclauses)
@_generative
def correlate_except(self, *fromclauses):
if fromclauses and fromclauses[0] is None:
self._correlate_except = ()
else:
- self._correlate_except = set(self._correlate_except
- ).union(fromclauses)
+ self._correlate_except = set(self._correlate_except).union(
+ _interpret_as_from(f) for f in fromclauses)
def append_correlation(self, fromclause):
"""append the given correlation expression to this select()
construct."""
self._should_correlate = False
- self._correlate = set(self._correlate).union([fromclause])
+ self._correlate = set(self._correlate).union(
+ _interpret_as_from(f) for f in fromclause)
def append_column(self, column):
"""append the given column expression to the columns clause of this
"""
self._reset_exported()
- fromclause = _literal_as_text(fromclause)
+ fromclause = _interpret_as_from(fromclause)
self._from_obj = self._from_obj.union([fromclause])
def _populate_column_collection(self):
configure_mappers()
-
class QueryCorrelatesLikeSelect(QueryTest, AssertsCompiledSQL):
query_correlated = "SELECT users.name AS users_name, " \
def test_insp_prop(self):
User = self.classes.User
prop = inspect(User.addresses)
- is_(prop, User.addresses.property)
+ is_(prop, User.addresses)
+
+ def test_insp_aliased_prop(self):
+ User = self.classes.User
+ ua = aliased(User)
+ prop = inspect(ua.addresses)
+ is_(prop, ua.addresses)
def test_rel_accessors(self):
User = self.classes.User
Address = self.classes.Address
prop = inspect(User.addresses)
- is_(prop.parent, class_mapper(User))
- is_(prop.mapper, class_mapper(Address))
+ is_(prop.property.parent, class_mapper(User))
+ is_(prop.property.mapper, class_mapper(Address))
assert not hasattr(prop, 'columns')
- assert not hasattr(prop, 'expression')
+ assert hasattr(prop, 'expression')
def test_instance_state(self):
asserted
)
+class RawSelectTest(QueryTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_select_from_entity(self):
+ User = self.classes.User
+
+ self.assert_compile(
+ select(['*']).select_from(User),
+ "SELECT * FROM users"
+ )
+
+ def test_select_from_aliased_entity(self):
+ User = self.classes.User
+ ua = aliased(User, name="ua")
+ self.assert_compile(
+ select(['*']).select_from(ua),
+ "SELECT * FROM users AS ua"
+ )
+
+ def test_correlate_entity(self):
+ User = self.classes.User
+ Address = self.classes.Address
+
+ self.assert_compile(
+ select([User]).where(User.id == Address.user_id).
+ correlate(Address),
+ "SELECT users.id, users.name FROM users "
+ "WHERE users.id = addresses.user_id"
+ )
+
+ def test_correlate_aliased_entity(self):
+ User = self.classes.User
+ Address = self.classes.Address
+ aa = aliased(Address, name="aa")
+
+ self.assert_compile(
+ select([User]).where(User.id == aa.user_id).
+ correlate(aa),
+ "SELECT users.id, users.name FROM users "
+ "WHERE users.id = aa.user_id"
+ )
+
+ def test_columns_clause_entity(self):
+ User = self.classes.User
+
+ self.assert_compile(
+ select([User]),
+ "SELECT users.id, users.name FROM users"
+ )
+
+ def test_columns_clause_columns(self):
+ User = self.classes.User
+
+ self.assert_compile(
+ select([User.id, User.name]),
+ "SELECT users.id, users.name FROM users"
+ )
+
+ def test_columns_clause_aliased_columns(self):
+ User = self.classes.User
+ ua = aliased(User, name='ua')
+ self.assert_compile(
+ select([ua.id, ua.name]),
+ "SELECT ua.id, ua.name FROM users AS ua"
+ )
+
+ def test_columns_clause_aliased_entity(self):
+ User = self.classes.User
+ ua = aliased(User, name='ua')
+ self.assert_compile(
+ select([ua]),
+ "SELECT ua.id, ua.name FROM users AS ua"
+ )
+
class GetTest(QueryTest):
def test_get(self):
User = self.classes.User
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
