PropComparator.
"""
+ def __init__(self, prop, mapper, adapter=None):
+ self.prop = self.property = prop
+ self.mapper = mapper
+ self.adapter = adapter
+
def __clause_element__(self):
raise NotImplementedError("%r" % self)
+ def adapted(self, adapter):
+ return self.__class__(self.prop, self.mapper, adapter)
+
@staticmethod
def any_op(a, b, **kwargs):
return a.any(b, **kwargs)
def has_op(a, b, **kwargs):
return a.has(b, **kwargs)
- def __init__(self, prop, mapper):
- self.prop = self.property = prop
- self.mapper = mapper
-
@staticmethod
def of_type_op(a, class_):
return a.of_type(class_)
class ColumnComparator(PropComparator):
@util.memoized_instancemethod
def __clause_element__(self):
- return self.prop.columns[0]._annotate({"parententity": self.mapper})
-
+ if self.adapter:
+ return self.adapter(self.prop.columns[0])
+ else:
+ return self.prop.columns[0]._annotate({"parententity": self.mapper})
+
def operate(self, op, *other, **kwargs):
return op(self.__clause_element__(), *other, **kwargs)
self._is_backref = _is_backref
class Comparator(PropComparator):
- def __init__(self, prop, mapper, of_type=None):
+ def __init__(self, prop, mapper, of_type=None, adapter=None):
self.prop = self.property = prop
self.mapper = mapper
+ self.adapter = adapter
if of_type:
self._of_type = _class_to_mapper(of_type)
+ def adapted(self, adapter):
+ return PropertyLoader.Comparator(self.prop, self.mapper, getattr(self, '_of_type', None), adapter)
+
@property
def parententity(self):
return self.prop.parent
def __clause_element__(self):
- return self.prop.parent._with_polymorphic_selectable
+ elem = self.prop.parent._with_polymorphic_selectable
+ if self.adapter:
+ return self.adapter(elem)
+ else:
+ return elem
def operate(self, op, *other, **kwargs):
return op(self, *other, **kwargs)
def __eq__(self, other):
if other is None:
if self.prop.direction in [ONETOMANY, MANYTOMANY]:
- return ~sql.exists([1], self.prop.primaryjoin)
+ return ~self._criterion_exists()
else:
- return self.prop._optimized_compare(None)
+ return self.prop._optimized_compare(None, adapt_source=self.adapter)
elif self.prop.uselist:
raise sa_exc.InvalidRequestError("Can't compare a collection to an object or collection; use contains() to test for membership.")
else:
- return self.prop._optimized_compare(other)
+ import pdb
+ pdb.set_trace()
+ return self.prop._optimized_compare(other, adapt_source=self.adapter)
def _criterion_exists(self, criterion=None, **kwargs):
if getattr(self, '_of_type', None):
else:
to_selectable = None
- pj, sj, source, dest, secondary, target_adapter = self.prop._create_joins(dest_polymorphic=True, dest_selectable=to_selectable)
+ if self.adapter:
+ source_selectable = self.__clause_element__()
+ else:
+ source_selectable = None
+ pj, sj, source, dest, secondary, target_adapter = \
+ self.prop._create_joins(dest_polymorphic=True, dest_selectable=to_selectable, source_selectable=source_selectable)
for k in kwargs:
crit = self.prop.mapper.class_manager.get_inst(k) == kwargs[k]
else:
criterion = criterion & crit
+ # why are we annotating ????
if sj:
j = _orm_annotate(pj) & sj
else:
# to anything in the enclosing query.
if criterion:
criterion = criterion._annotate({'_halt_adapt': True})
- return sql.exists([1], j & criterion, from_obj=dest).correlate(source)
+
+ crit = j & criterion
+
+ return sql.exists([1], crit, from_obj=dest).correlate(source)
def any(self, criterion=None, **kwargs):
if not self.prop.uselist:
def contains(self, other, **kwargs):
if not self.prop.uselist:
raise sa_exc.InvalidRequestError("'contains' not implemented for scalar attributes. Use ==")
- clause = self.prop._optimized_compare(other)
+ clause = self.prop._optimized_compare(other, adapt_source=self.adapter)
if self.prop.secondaryjoin:
clause.negation_clause = self.__negated_contains_or_equals(other)
if self.prop.direction == MANYTOONE:
state = attributes.instance_state(other)
strategy = self.prop._get_strategy(strategies.LazyLoader)
+
+ def state_bindparam(state, col):
+ o = state.obj() # strong ref
+ return lambda: self.prop.mapper._get_committed_attr_by_column(o, col)
+
+ def adapt(col):
+ if self.adapter:
+ return self.adapter(col)
+ else:
+ return col
+
if strategy.use_get:
return sql.and_(*[
sql.or_(
- x !=
- self.prop.mapper._get_committed_state_attr_by_column(state, y),
- x == None)
+ adapt(x) != state_bindparam(state, y),
+ adapt(x) == None)
for (x, y) in self.prop.local_remote_pairs])
criterion = sql.and_(*[x==y for (x, y) in zip(self.prop.mapper.primary_key, self.prop.mapper.primary_key_from_instance(other))])
else:
return op(self.comparator, value)
- def _optimized_compare(self, value, value_is_parent=False):
+ def _optimized_compare(self, value, value_is_parent=False, adapt_source=None):
if value is not None:
value = attributes.instance_state(value)
- return self._get_strategy(strategies.LazyLoader).lazy_clause(value, reverse_direction=not value_is_parent, alias_secondary=True)
+ return self._get_strategy(strategies.LazyLoader).\
+ lazy_clause(value, reverse_direction=not value_is_parent, alias_secondary=True, adapt_source=adapt_source)
def __str__(self):
return str(self.parent.class_.__name__) + "." + self.key
"Specify a 'primaryjoin' expression. If this is a "
"many-to-many relation, 'secondaryjoin' is needed as well." % (self))
+ self.primaryjoin = self.primaryjoin._annotate({}) #{'_halt_adapt':True})
+# if self.secondaryjoin:
+# self.secondaryjoin = self.secondaryjoin._annotate({}) #{'_halt_adapt':True})
+
def _col_is_part_of_mappings(self, column):
if self.secondary is None:
return self.parent.mapped_table.c.contains_column(column) or \
self.is_class_level = True
self._register_attribute(self.parent.class_, callable_=self.class_level_loader)
- def lazy_clause(self, state, reverse_direction=False, alias_secondary=False):
+ def lazy_clause(self, state, reverse_direction=False, alias_secondary=False, adapt_source=None):
if state is None:
- return self._lazy_none_clause(reverse_direction)
+ return self._lazy_none_clause(reverse_direction, adapt_source=adapt_source)
if not reverse_direction:
(criterion, bind_to_col, rev) = (self.__lazywhere, self.__bind_to_col, self._equated_columns)
if self.parent_property.secondary and alias_secondary:
criterion = sql_util.ClauseAdapter(self.parent_property.secondary.alias()).traverse(criterion)
- return visitors.cloned_traverse(criterion, {}, {'bindparam':visit_bindparam})
-
- def _lazy_none_clause(self, reverse_direction=False):
+ criterion = visitors.cloned_traverse(criterion, {}, {'bindparam':visit_bindparam})
+ if adapt_source:
+ criterion = adapt_source(criterion)
+ return criterion
+
+ def _lazy_none_clause(self, reverse_direction=False, adapt_source=None):
if not reverse_direction:
(criterion, bind_to_col, rev) = (self.__lazywhere, self.__bind_to_col, self._equated_columns)
else:
binary.right = expression.null()
binary.operator = operators.is_
- return visitors.cloned_traverse(criterion, {}, {'binary':visit_binary})
+ criterion = visitors.cloned_traverse(criterion, {}, {'binary':visit_binary})
+ if adapt_source:
+ criterion = adapt_source(criterion)
+ return criterion
def class_level_loader(self, state, options=None, path=None):
if not mapperutil._state_has_identity(state):
def __adapt_prop(self, prop):
existing = getattr(self.__target, prop.key)
- comparator = AliasedComparator(self, self.__adapter, existing.comparator)
+
+ adapter = sql_util.ClauseAdapter(self.__alias, equivalents=self.__mapper._equivalent_columns, exclude=getattr(prop, 'remote_side', None))
+ def adapt(elem):
+ return adapter.traverse(elem)._annotate({'parententity': self})
+ comparator = existing.comparator.adapted(adapt)
+
queryattr = attributes.QueryableAttribute(
existing.impl, parententity=self, comparator=comparator)
setattr(self, prop.key, queryattr)
return '<AliasedClass at 0x%x; %s>' % (
id(self), self.__target.__name__)
-class AliasedComparator(PropComparator):
- def __init__(self, aliasedclass, adapter, comparator):
- self.aliasedclass = aliasedclass
- self.comparator = comparator
- self.adapter = adapter
- self.__clause_element = self.adapter.traverse(self.comparator.__clause_element__())._annotate({'parententity': aliasedclass})
-
- def __clause_element__(self):
- return self.__clause_element
-
- def operate(self, op, *other, **kwargs):
- return self.adapter.traverse(self.comparator.operate(op, *other, **kwargs))
-
- def reverse_operate(self, op, other, **kwargs):
- return self.adapter.traverse(self.comparator.reverse_operate(op, *other, **kwargs))
-
def _orm_annotate(element, exclude=None):
"""Deep copy the given ClauseElement, annotating each element with the "_orm_adapt" flag.
def clone(elem):
if exclude and elem in exclude:
elem = elem._clone()
- elif '_orm_adapt' not in elem._annotations:
- elem = elem._annotate({'_orm_adapt':True})
+ elif '_orm_adapt' not in elem._annotations or '_halt_adapt' in elem._annotations:
+ elem = elem._annotate({'_orm_adapt':True, '_halt_adapt':False})
elem._copy_internals(clone=clone)
return elem
def test_polymorphic_any(self):
sess = create_session()
+ # test that the aliasing on "Person" does not bleed into the
+ # EXISTS clause generated by any()
+
+ # fails with this line, which simulates the usage of declarative
+ # need to establish whether or not primaryjoin is _orm_annotated or not
+ # in general
+ #Company.employees.property.primaryjoin = Company.company_id==Person.company_id
+ #self.assertEquals(
+ # sess.query(Company).join(Company.employees, aliased=True).filter(Person.name=='dilbert').\
+ # filter(Company.employees.any(Person.name=='wally')).all(), [c1]
+ #)
+
+ self.assertEquals(
+ sess.query(Company).join(Company.employees, aliased=True).filter(Person.name=='dilbert').\
+ filter(Company.employees.any(Person.name=='vlad')).all(), []
+ )
+ return
+
self.assertEquals(
sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).one(),
c2
})
mapper(Keyword, keywords)
+ compile_mappers()
+ #class_mapper(User).add_property('addresses', relation(Address, primaryjoin=User.id==Address.user_id, order_by=Address.id, backref='user'))
+
class UnicodeSchemaTest(QueryTest):
keep_mappers = False
)
# fails, needs autoaliasing
- #self._test(
- # Node.children==None,
- # "NOT (EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id))"
- #)
+ self._test(
+ Node.children==None,
+ "NOT (EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id))"
+ )
self._test(
Node.parent==None,
"nodes_1.parent_id IS NULL"
)
- # fails, needs autoaliasing
- #self._test(
- # Node.children==[Node(id=1), Node(id=2)],
- # "(EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id AND nodes_1.id = :id_1)) "
- # "AND (EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id AND nodes_1.id = :id_2))"
- #)
-
# fails, overaliases
- #self._test(
- # nalias.children==[Node(id=1), Node(id=2)],
- # "(EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id AND nodes_1.id = :id_1)) "
- # "AND (EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE nodes.id = nodes_1.parent_id AND nodes_1.id = :id_2))"
- #)
-
- # fails, overaliases
- #self._test(
- # nalias.children==None,
- # "NOT (EXISTS (SELECT 1 FROM nodes AS nodes WHERE nodes_1.id = nodes.parent_id))"
- #)
+ self._test(
+ nalias.children==None,
+ "NOT (EXISTS (SELECT 1 FROM nodes WHERE nodes_1.id = nodes.parent_id))"
+ )
# fails
- #self._test(
- # nalias.children.any(Node.data=='some data'),
- # "EXISTS (SELECT 1 FROM nodes WHERE "
- # "nodes_1.id = nodes.parent_id AND nodes.data = :data_1)")
+ self._test(
+ nalias.children.any(Node.data=='some data'),
+ "EXISTS (SELECT 1 FROM nodes WHERE "
+ "nodes_1.id = nodes.parent_id AND nodes.data = :data_1)")
- # fails
+ # fails, but I think I want this to fail
#self._test(
# Node.children.any(nalias.data=='some data'),
# "EXISTS (SELECT 1 FROM nodes AS nodes_1 WHERE "
# )
# fails, overaliases
- #self._test(
- # nalias.parent.has(Node.data=='some data'),
- # "EXISTS (SELECT 1 FROM nodes WHERE nodes.id = nodes_1.parent_id AND nodes.data = :data_1)"
- #)
+ self._test(
+ nalias.parent.has(Node.data=='some data'),
+ "EXISTS (SELECT 1 FROM nodes WHERE nodes.id = nodes_1.parent_id AND nodes.data = :data_1)"
+ )
self._test(
Node.parent.has(Node.data=='some data'),
# fails
# (also why are we doing an EXISTS for this??)
- #self._test(
- # nalias.parent != Node(id=7),
- # 'NOT (EXISTS (SELECT 1 FROM nodes WHERE nodes.id = nodes_1.parent_id AND nodes.id = :id_1))'
- #)
+ self._test(
+ nalias.parent != Node(id=7),
+ 'nodes_1.parent_id != :parent_id_1 OR nodes_1.parent_id IS NULL'
+ )
self._test(
nalias.children.contains(Node(id=7)), "nodes_1.id = :param_1"
def test_selfref_between(self):
ualias = aliased(User)
self._test(User.id.between(ualias.id, ualias.id), "users.id BETWEEN users_1.id AND users_1.id")
- # fails:
- # self._test(ualias.id.between(User.id, User.id), "users_1.id BETWEEN users.id AND users.id")
+ self._test(ualias.id.between(User.id, User.id), "users_1.id BETWEEN users.id AND users.id")
def test_clauses(self):
for (expr, compare) in (
def test_binds(self):
assert [User(id=8), User(id=9)] == create_session().query(User).filter("id in (:id1, :id2)").params(id1=8, id2=9).all()
+
+class FooTest(FixtureTest):
+ keep_data = True
+
+ def test_filter_by(self):
+ clear_mappers()
+ sess = create_session(bind=testing.db)
+ from sqlalchemy.ext.declarative import declarative_base
+ Base = declarative_base(bind=testing.db)
+ class User(Base, _base.ComparableEntity):
+ __table__ = users
+
+ class Address(Base, _base.ComparableEntity):
+ __table__ = addresses
+
+ compile_mappers()
+# Address.user = relation(User, primaryjoin="User.id==Address.user_id")
+ Address.user = relation(User, primaryjoin=User.id==Address.user_id)
+# Address.user = relation(User, primaryjoin=users.c.id==addresses.c.user_id)
+ compile_mappers()
+# Address.user.property.primaryjoin = User.id==Address.user_id
+ user = sess.query(User).get(8)
+ print sess.query(Address).filter_by(user=user).all()
+ assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(Address).filter_by(user=user).all()
+
class FilterTest(QueryTest):
def test_basic(self):
assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).all()
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from testlib import *
-from sqlalchemy.sql import util as sql_util
+from sqlalchemy.sql import util as sql_util, visitors
from sqlalchemy import exc
metadata = MetaData()
assert inner.corresponding_column(t2.c.col1, require_embedded=False) is inner.corresponding_column(t2.c.col1, require_embedded=True) is inner.c.col1
assert inner.corresponding_column(t1.c.col1, require_embedded=False) is inner.corresponding_column(t1.c.col1, require_embedded=True) is inner.c.col1
+ def test_annotated_visit(self):
+ from sqlalchemy.sql import table, column
+ table1 = table('table1', column("col1"), column("col2"))
+
+ bin = table1.c.col1 == bindparam('foo', value=None)
+ assert str(bin) == "table1.col1 = :foo"
+ def visit_binary(b):
+ b.right = table1.c.col2
+ b2 = visitors.cloned_traverse(bin, {}, {'binary':visit_binary})
+ assert str(b2) == "table1.col1 = table1.col2"
+
+ b3 = visitors.cloned_traverse(bin._annotate({}), {}, {'binary':visit_binary})
+ assert str(b3) == "table1.col1 = table1.col2"
+
+
if __name__ == "__main__":
testenv.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
