.. changelog::
:version: 0.9.0b2
+ .. change::
+ :tags: bug, orm
+ :tickets: 2872
+
+ Some refinements to the :class:`.AliasedClass` construct with regards
+ to descriptors, like hybrids, synonyms, composites, user-defined
+ descriptors, etc. The attribute
+ adaptation which goes on has been made more robust, such that if a descriptor
+ returns another instrumented attribute, rather than a compound SQL
+ expression element, the operation will still proceed.
+ Addtionally, the "adapted" operator will retain its class; previously,
+ a change in class from ``InstrumentedAttribute`` to ``QueryableAttribute``
+ (a superclass) would interact with Python's operator system such that
+ an expression like ``aliased(MyClass.x) > MyClass.x`` would reverse itself
+ to read ``myclass.x < myclass_1.x``. The adapted attribute will also
+ refer to the new :class:`.AliasedClass` as its parent which was not
+ always the case before.
+
.. change::
:tags: feature, sql
:tickets: 2867
return self.comparator._query_clause_element()
+ def adapt_to_entity(self, adapt_to_entity):
+ assert not self._of_type
+ return self.__class__(adapt_to_entity.entity, self.key, impl=self.impl,
+ comparator=self.comparator.adapt_to_entity(adapt_to_entity),
+ parententity=adapt_to_entity)
+
def of_type(self, cls):
return QueryableAttribute(
self.class_,
return self._comparator
def adapt_to_entity(self, adapt_to_entity):
- return self.__class__(self.class_, self.key, self.descriptor,
+ return self.__class__(adapt_to_entity.entity, self.key, self.descriptor,
self._comparator,
adapt_to_entity)
else:
raise AttributeError(key)
- if isinstance(attr, attributes.QueryableAttribute):
- return _aliased_insp._adapt_prop(attr, key)
+ if isinstance(attr, PropComparator):
+ ret = attr.adapt_to_entity(_aliased_insp)
+ setattr(self, key, ret)
+ return ret
elif hasattr(attr, 'func_code'):
is_method = getattr(_aliased_insp._target, key, None)
if is_method and is_method.__self__ is not None:
ret = attr.__get__(None, self)
if isinstance(ret, PropComparator):
return ret.adapt_to_entity(_aliased_insp)
- return ret
+ else:
+ return ret
else:
return attr
'parentmapper': self.mapper}
)
- def _adapt_prop(self, existing, key):
- comparator = existing.comparator.adapt_to_entity(self)
- queryattr = attributes.QueryableAttribute(
- self.entity, key,
- impl=existing.impl,
- parententity=self,
- comparator=comparator)
- setattr(self.entity, key, queryattr)
- return queryattr
-
-
def _entity_for_mapper(self, mapper):
self_poly = self.with_polymorphic_mappers
if mapper in self_poly:
users, User = self.tables.users, self.classes.User
-
mapper(User, users)
sess = create_session()
ualias = aliased(User)
self.assert_compile(
- sess.query(User).join(sel, User.id>sel.c.id),
+ sess.query(User).join(sel, User.id > sel.c.id),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users JOIN (SELECT users.id AS id, users.name AS name FROM "
"users WHERE users.id IN (:id_1, :id_2)) AS anon_1 ON users.id > anon_1.id",
)
self.assert_compile(
- sess.query(ualias).select_entity_from(sel).filter(ualias.id>sel.c.id),
+ sess.query(ualias).select_entity_from(sel).filter(ualias.id > sel.c.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name FROM "
"users AS users_1, (SELECT users.id AS id, users.name AS name FROM "
"users WHERE users.id IN (:id_1, :id_2)) AS anon_1 WHERE users_1.id > anon_1.id",
)
self.assert_compile(
- sess.query(ualias).select_entity_from(sel).join(ualias, ualias.id>sel.c.id),
+ sess.query(ualias).select_entity_from(sel).join(ualias, ualias.id > sel.c.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
"FROM users WHERE users.id IN (:id_1, :id_2)) AS anon_1 "
)
self.assert_compile(
- sess.query(ualias).select_entity_from(sel).join(ualias, ualias.id>User.id),
+ sess.query(ualias).select_entity_from(sel).join(ualias, ualias.id > User.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name FROM "
"users WHERE users.id IN (:id_1, :id_2)) AS anon_1 "
- "JOIN users AS users_1 ON anon_1.id < users_1.id"
+ "JOIN users AS users_1 ON users_1.id > anon_1.id"
)
salias = aliased(User, sel)
self.assert_compile(
- sess.query(salias).join(ualias, ualias.id>salias.id),
+ sess.query(salias).join(ualias, ualias.id > salias.id),
"SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name FROM "
"(SELECT users.id AS id, users.name AS name FROM users WHERE users.id "
"IN (:id_1, :id_2)) AS anon_1 JOIN users AS users_1 ON users_1.id > anon_1.id",
)
-
- # this one uses an explicit join(left, right, onclause) so works
self.assert_compile(
- sess.query(ualias).select_entity_from(join(sel, ualias, ualias.id>sel.c.id)),
+ sess.query(ualias).select_entity_from(join(sel, ualias, ualias.id > sel.c.id)),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name FROM "
"(SELECT users.id AS id, users.name AS name FROM users WHERE users.id "
- "IN (:id_1, :id_2)) AS anon_1 JOIN users AS users_1 ON users_1.id > anon_1.id",
- use_default_dialect=True
+ "IN (:id_1, :id_2)) AS anon_1 JOIN users AS users_1 ON users_1.id > anon_1.id"
)
self.assert_compile(
sess.query(User).select_from(ua).join(User, ua.name > User.name),
"SELECT users.id AS users_id, users.name AS users_name "
- "FROM users AS users_1 JOIN users ON users.name < users_1.name"
+ "FROM users AS users_1 JOIN users ON users_1.name > users.name"
)
self.assert_compile(
sess.query(User.name).select_from(ua).join(User, ua.name > User.name),
"SELECT users.name AS users_name FROM users AS users_1 "
- "JOIN users ON users.name < users_1.name"
+ "JOIN users ON users_1.name > users.name"
)
self.assert_compile(
sess.query(ua.name).select_from(ua).join(User, ua.name > User.name),
"SELECT users_1.name AS users_1_name FROM users AS users_1 "
- "JOIN users ON users.name < users_1.name"
+ "JOIN users ON users_1.name > users.name"
)
self.assert_compile(
sess.query(ua).select_from(User).join(ua, ua.name > User.name),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
- "FROM users JOIN users AS users_1 ON users.name < users_1.name"
+ "FROM users JOIN users AS users_1 ON users_1.name > users.name"
+ )
+
+ self.assert_compile(
+ sess.query(ua).select_from(User).join(ua, User.name > ua.name),
+ "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users JOIN users AS users_1 ON users.name > users_1.name"
)
# this is tested in many other places here, just adding it
"FROM job AS job_1 LEFT OUTER JOIN subjob AS subjob_1 "
"ON job_1.id = subjob_1.id "
"WHERE data_container.id = job_1.container_id "
- "AND job.id > job_1.id)"
+ "AND job_1.id < job.id)"
)
def test_any_walias(self):
"WHERE EXISTS (SELECT 1 "
"FROM job AS job_1 "
"WHERE data_container.id = job_1.container_id "
- "AND job.id > job_1.id AND job_1.type = :type_1)"
+ "AND job_1.id < job.id AND job_1.type = :type_1)"
)
def test_join_wpoly(self):
where(uu.id == Address.user_id).\
correlate(uu).as_scalar()
]),
- # curious, "address.user_id = uu.id" is reversed here
+ # for a long time, "uu.id = address.user_id" was reversed;
+ # this was resolved as of #2872 and had to do with
+ # InstrumentedAttribute.__eq__() taking precedence over
+ # QueryableAttribute.__eq__()
"SELECT uu.name, addresses.id, "
"(SELECT count(addresses.id) AS count_1 "
- "FROM addresses WHERE addresses.user_id = uu.id) AS anon_1 "
+ "FROM addresses WHERE uu.id = addresses.user_id) AS anon_1 "
"FROM users AS uu, addresses"
)
"SELECT users.id AS users_id, users.name AS users_name, "
"users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users INNER JOIN users AS users_1 USE INDEX (col1_index,col2_index) "
- "ON users.id < users_1.id",
+ "ON users_1.id > users.id",
dialect=dialect
)
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Table
-from sqlalchemy.orm import aliased, with_polymorphic
-from sqlalchemy.orm import mapper, create_session
+from sqlalchemy.orm import aliased, with_polymorphic, synonym
+from sqlalchemy.orm import mapper, create_session, Session
from sqlalchemy.testing import fixtures
from test.orm import _fixtures
from sqlalchemy.testing import eq_, is_
from sqlalchemy.orm.path_registry import PathRegistry, RootRegistry
from sqlalchemy import inspect
+from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
+from sqlalchemy.testing import AssertsCompiledSQL
-class AliasedClassTest(fixtures.TestBase):
- def point_map(self, cls):
+class AliasedClassTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def _fixture(self, cls, properties={}):
table = Table('point', MetaData(),
Column('id', Integer(), primary_key=True),
Column('x', Integer),
Column('y', Integer))
- mapper(cls, table)
+ mapper(cls, table, properties=properties)
return table
def test_simple(self):
class Point(object):
pass
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert Point.id.__clause_element__().table is table
assert alias.id.__clause_element__().table is not table
- def test_notcallable(self):
+ def test_not_instantiatable(self):
class Point(object):
pass
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert_raises(TypeError, alias)
- def test_instancemethods(self):
+ def test_instancemethod(self):
class Point(object):
def zero(self):
self.x, self.y = 0, 0
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert Point.zero
+ # TODO: I don't quite understand this
+ # still
if util.py2k:
- # TODO: what is this testing ??
assert not getattr(alias, 'zero')
+ else:
+ assert getattr(alias, 'zero')
- def test_classmethods(self):
+ def test_classmethod(self):
class Point(object):
@classmethod
def max_x(cls):
return 100
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert Point.max_x
assert alias.max_x
- assert Point.max_x() == alias.max_x()
+ assert Point.max_x() == alias.max_x() == 100
- def test_simpleproperties(self):
+ def test_simple_property(self):
class Point(object):
@property
def max_x(self):
return 100
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert Point.max_x
assert Point.max_x is alias.max_x
def test_descriptors(self):
- """Tortured..."""
class descriptor(object):
def __init__(self, fn):
def thing(self, arg):
return arg.center
- table = self.point_map(Point)
+ table = self._fixture(Point)
alias = aliased(Point)
assert Point.thing != (0, 0)
assert alias.thing != (0, 0)
assert alias.thing.method() == 'method'
- def test_hybrid_descriptors(self):
+ def _assert_has_table(self, expr, table):
from sqlalchemy import Column # override testlib's override
- import types
-
- class MethodDescriptor(object):
- def __init__(self, func):
- self.func = func
- def __get__(self, instance, owner):
- if instance is None:
- if util.py2k:
- args = (self.func, owner, owner.__class__)
- else:
- args = (self.func, owner)
- else:
- if util.py2k:
- args = (self.func, instance, owner)
- else:
- args = (self.func, instance)
- return types.MethodType(*args)
-
- class PropertyDescriptor(object):
- def __init__(self, fget, fset, fdel):
- self.fget = fget
- self.fset = fset
- self.fdel = fdel
- def __get__(self, instance, owner):
- if instance is None:
- return self.fget(owner)
- else:
- return self.fget(instance)
- def __set__(self, instance, value):
- self.fset(instance, value)
- def __delete__(self, instance):
- self.fdel(instance)
- hybrid = MethodDescriptor
- def hybrid_property(fget, fset=None, fdel=None):
- return PropertyDescriptor(fget, fset, fdel)
-
- def assert_table(expr, table):
- for child in expr.get_children():
- if isinstance(child, Column):
- assert child.table is table
+ for child in expr.get_children():
+ if isinstance(child, Column):
+ assert child.table is table
+ def test_hybrid_descriptor_one(self):
class Point(object):
def __init__(self, x, y):
self.x, self.y = x, y
- @hybrid
+
+ @hybrid_method
def left_of(self, other):
return self.x < other.x
- double_x = hybrid_property(lambda self: self.x * 2)
+ self._fixture(Point)
+ alias = aliased(Point)
+ sess = Session()
+
+ self.assert_compile(
+ sess.query(alias).filter(alias.left_of(Point)),
+ "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, "
+ "point_1.y AS point_1_y FROM point AS point_1, point "
+ "WHERE point_1.x < point.x"
+ )
+
+ def test_hybrid_descriptor_two(self):
+ class Point(object):
+ def __init__(self, x, y):
+ self.x, self.y = x, y
+
+ @hybrid_property
+ def double_x(self):
+ return self.x * 2
- table = self.point_map(Point)
+ self._fixture(Point)
alias = aliased(Point)
- alias_table = alias.x.__clause_element__().table
- assert table is not alias_table
- p1 = Point(-10, -10)
- p2 = Point(20, 20)
+ eq_(str(Point.double_x), "point.x * :x_1")
+ eq_(str(alias.double_x), "point_1.x * :x_1")
- assert p1.left_of(p2)
- assert p1.double_x == -20
+ sess = Session()
+
+ self.assert_compile(
+ sess.query(alias).filter(alias.double_x > Point.x),
+ "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, "
+ "point_1.y AS point_1_y FROM point AS point_1, point "
+ "WHERE point_1.x * :x_1 > point.x"
+ )
+
+ def test_hybrid_descriptor_three(self):
+ class Point(object):
+ def __init__(self, x, y):
+ self.x, self.y = x, y
- assert_table(Point.double_x, table)
- assert_table(alias.double_x, alias_table)
+ @hybrid_property
+ def x_alone(self):
+ return self.x
- assert_table(Point.left_of(p2), table)
- assert_table(alias.left_of(p2), alias_table)
+ self._fixture(Point)
+ alias = aliased(Point)
+
+ eq_(str(Point.x_alone), "Point.x")
+ eq_(str(alias.x_alone), "AliasedClass_Point.x")
+
+ assert Point.x_alone is Point.x
+
+ eq_(str(alias.x_alone == alias.x), "point_1.x = point_1.x")
+
+ a2 = aliased(Point)
+ eq_(str(a2.x_alone == alias.x), "point_1.x = point_2.x")
+
+ sess = Session()
+
+ self.assert_compile(
+ sess.query(alias).filter(alias.x_alone > Point.x),
+ "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, "
+ "point_1.y AS point_1_y FROM point AS point_1, point "
+ "WHERE point_1.x > point.x"
+ )
+
+ def test_proxy_descriptor_one(self):
+ class Point(object):
+ def __init__(self, x, y):
+ self.x, self.y = x, y
+
+ self._fixture(Point, properties={
+ 'x_syn': synonym("x")
+ })
+ alias = aliased(Point)
+
+ eq_(str(Point.x_syn), "Point.x_syn")
+ eq_(str(alias.x_syn), "AliasedClass_Point.x_syn")
+
+ sess = Session()
+ self.assert_compile(
+ sess.query(alias.x_syn).filter(alias.x_syn > Point.x_syn),
+ "SELECT point_1.x AS point_1_x FROM point AS point_1, point "
+ "WHERE point_1.x > point.x"
+ )
class IdentityKeyTest(_fixtures.FixtureTest):
run_inserts = None
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
