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):
- if self.adapter:
- return self.adapter(self.prop.columns[0])
- else:
- return self.prop.columns[0]._annotate({"parententity": self.mapper})
-
+ 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, adapter=None):
+ def __init__(self, prop, mapper, of_type=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):
- elem = self.prop.parent._with_polymorphic_selectable
- if self.adapter:
- return self.adapter(elem)
- else:
- return elem
+ return self.prop.parent._with_polymorphic_selectable
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 ~self._criterion_exists()
+ return ~sql.exists([1], self.prop.primaryjoin)
else:
- return self.prop._optimized_compare(None, adapt_source=self.adapter)
+ return self.prop._optimized_compare(None)
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:
- import pdb
- pdb.set_trace()
- return self.prop._optimized_compare(other, adapt_source=self.adapter)
+ return self.prop._optimized_compare(other)
def _criterion_exists(self, criterion=None, **kwargs):
if getattr(self, '_of_type', None):
else:
to_selectable = None
- 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)
+ pj, sj, source, dest, secondary, target_adapter = self.prop._create_joins(dest_polymorphic=True, dest_selectable=to_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})
-
- crit = j & criterion
-
- return sql.exists([1], crit, from_obj=dest).correlate(source)
+ return sql.exists([1], j & criterion, 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, adapt_source=self.adapter)
+ clause = self.prop._optimized_compare(other)
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_(
- adapt(x) != state_bindparam(state, y),
- adapt(x) == None)
+ x !=
+ self.prop.mapper._get_committed_state_attr_by_column(state, y),
+ 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, adapt_source=None):
+ def _optimized_compare(self, value, value_is_parent=False):
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, adapt_source=adapt_source)
+ return self._get_strategy(strategies.LazyLoader).lazy_clause(value, reverse_direction=not value_is_parent, alias_secondary=True)
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, adapt_source=None):
+ def lazy_clause(self, state, reverse_direction=False, alias_secondary=False):
if state is None:
- return self._lazy_none_clause(reverse_direction, adapt_source=adapt_source)
+ return self._lazy_none_clause(reverse_direction)
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)
- 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):
+ return visitors.cloned_traverse(criterion, {}, {'bindparam':visit_bindparam})
+
+ def _lazy_none_clause(self, reverse_direction=False):
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_
- criterion = visitors.cloned_traverse(criterion, {}, {'binary':visit_binary})
- if adapt_source:
- criterion = adapt_source(criterion)
- return criterion
+ return visitors.cloned_traverse(criterion, {}, {'binary':visit_binary})
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)
-
- 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)
-
+ comparator = AliasedComparator(self, self.__adapter, existing.comparator)
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 or '_halt_adapt' in elem._annotations:
- elem = elem._annotate({'_orm_adapt':True, '_halt_adapt':False})
+ elif '_orm_adapt' not in elem._annotations:
+ elem = elem._annotate({'_orm_adapt':True})
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==None,
- "NOT (EXISTS (SELECT 1 FROM nodes WHERE nodes_1.id = nodes.parent_id))"
- )
+ #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))"
+ #)
# 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, but I think I want this to fail
+ # fails
#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),
- 'nodes_1.parent_id != :parent_id_1 OR nodes_1.parent_id IS NULL'
- )
+ #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.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")
- self._test(ualias.id.between(User.id, User.id), "users_1.id BETWEEN users.id AND users.id")
+ # fails:
+ # 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, visitors
+from sqlalchemy.sql import util as sql_util
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
