property() occurred. [ticket:2188].
Also in 0.6.9
+ - The join condition produced by with_parent
+ as well as when using a "dynamic" relationship
+ against a parent will generate unique
+ bindparams, rather than incorrectly repeating
+ the same bindparam. [ticket:2207].
+ Also in 0.6.9.
+
- Added the same "columns-only" check to
- mapper.polymorphic_on as used in
+ mapper.polymorphic_on as used when
+ receiving user arguments to
relationship.order_by, foreign_keys,
remote_side, etc.
def init(self):
super(LazyLoader, self).init()
- self.__lazywhere, \
- self.__bind_to_col, \
+ self._lazywhere, \
+ self._bind_to_col, \
self._equated_columns = self._create_lazy_clause(self.parent_property)
- self.logger.info("%s lazy loading clause %s", self, self.__lazywhere)
+ self._rev_lazywhere, \
+ self._rev_bind_to_col, \
+ self._rev_equated_columns = self._create_lazy_clause(
+ self.parent_property,
+ reverse_direction=True)
+
+ self.logger.info("%s lazy loading clause %s", self, self._lazywhere)
# determine if our "lazywhere" clause is the same as the mapper's
# get() clause. then we can just use mapper.get()
#from sqlalchemy.orm import query
self.use_get = not self.uselist and \
self.mapper._get_clause[0].compare(
- self.__lazywhere,
+ self._lazywhere,
use_proxies=True,
equivalents=self.mapper._equivalent_columns
)
if not reverse_direction:
criterion, bind_to_col, rev = \
- self.__lazywhere, \
- self.__bind_to_col, \
+ self._lazywhere, \
+ self._bind_to_col, \
self._equated_columns
else:
criterion, bind_to_col, rev = \
- LazyLoader._create_lazy_clause(
- self.parent_property,
- reverse_direction=reverse_direction)
+ self._rev_lazywhere, \
+ self._rev_bind_to_col, \
+ self._rev_equated_columns
if reverse_direction:
mapper = self.parent_property.mapper
sess = sessionlib._state_session(state)
if sess is not None and sess._flushing:
def visit_bindparam(bindparam):
- if bindparam.key in bind_to_col:
+ if bindparam._identifying_key in bind_to_col:
bindparam.callable = \
- lambda: mapper._get_committed_state_attr_by_column(
- state, dict_, bind_to_col[bindparam.key])
+ lambda: mapper._get_committed_state_attr_by_column(
+ state, dict_,
+ bind_to_col[bindparam._identifying_key])
else:
def visit_bindparam(bindparam):
- if bindparam.key in bind_to_col:
- bindparam.callable = lambda: mapper._get_state_attr_by_column(
- state, dict_, bind_to_col[bindparam.key])
+ if bindparam._identifying_key in bind_to_col:
+ bindparam.callable = \
+ lambda: mapper._get_state_attr_by_column(
+ state, dict_,
+ bind_to_col[bindparam._identifying_key])
if self.parent_property.secondary is not None and alias_secondary:
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._lazywhere, \
+ self._bind_to_col,\
self._equated_columns
else:
criterion, bind_to_col, rev = \
- LazyLoader._create_lazy_clause(
- self.parent_property,
- reverse_direction=reverse_direction)
+ self._rev_lazywhere, \
+ self._rev_bind_to_col, \
+ self._rev_equated_columns
criterion = sql_util.adapt_criterion_to_null(criterion, bind_to_col)
if equated in binds:
return None
if col not in binds:
- binds[col] = sql.bindparam(None, None, type_=col.type)
+ binds[col] = sql.bindparam(None, None, type_=col.type, unique=True)
return binds[col]
return None
else:
self.key = key or _generated_label('%%(%d param)s'
% id(self))
+
+ # identifiying key that won't change across
+ # clones, used to identify the bind's logical
+ # identity
+ self._identifying_key = self.key
+
+ # key that was passed in the first place, used to
+ # generate new keys
self._orig_key = key or 'param'
+
self.unique = unique
self.value = value
self.callable = callable_
"""given criterion containing bind params, convert selected elements to IS NULL."""
def visit_binary(binary):
- if isinstance(binary.left, expression._BindParamClause) and binary.left.key in nulls:
+ if isinstance(binary.left, expression._BindParamClause) \
+ and binary.left._identifying_key in nulls:
# reverse order if the NULL is on the left side
binary.left = binary.right
binary.right = expression.null()
binary.operator = operators.is_
binary.negate = operators.isnot
- elif isinstance(binary.right, expression._BindParamClause) and binary.right.key in nulls:
+ elif isinstance(binary.right, expression._BindParamClause) \
+ and binary.right._identifying_key in nulls:
binary.right = expression.null()
binary.operator = operators.is_
binary.negate = operators.isnot
import operator
from sqlalchemy import MetaData, null, exists, text, union, literal, \
literal_column, func, between, Unicode, desc, and_, bindparam, \
- select, distinct
+ select, distinct, or_
from sqlalchemy import exc as sa_exc, util
from sqlalchemy.sql import compiler, table, column
from sqlalchemy.sql import expression
-class FilterTest(QueryTest):
+class FilterTest(QueryTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
def test_basic(self):
User = self.classes.User
#assert [User(id=7), User(id=9), User(id=10)] == sess.query(User).filter(User.addresses!=address).all()
+ def test_unique_binds_join_cond(self):
+ """test that binds used when the lazyclause is used in criterion are unique"""
+
+ User, Address = self.classes.User, self.classes.Address
+ sess = Session()
+ a1, a2 = sess.query(Address).order_by(Address.id)[0:2]
+ self.assert_compile(
+ sess.query(User).filter(User.addresses.contains(a1)).union(
+ sess.query(User).filter(User.addresses.contains(a2))
+ ),
+ "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS "
+ "anon_1_users_name FROM (SELECT users.id AS users_id, "
+ "users.name AS users_name FROM users WHERE users.id = :param_1 "
+ "UNION SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users WHERE users.id = :param_2) AS anon_1",
+ checkparams = {u'param_1': 7, u'param_2': 8}
+ )
+
def test_any(self):
User, Address = self.classes.User, self.classes.Address
eq_(s.query(User.id, "name").order_by(User.id).all(), [(7, u'jack'), (8, u'ed'), (9, u'fred'), (10, u'chuck')])
-class ParentTest(QueryTest):
+class ParentTest(QueryTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
def test_o2m(self):
User, orders, Order = (self.classes.User,
self.tables.orders,
User(id=o1.user_id)
)
+ def test_unique_binds_union(self):
+ """bindparams used in the 'parent' query are unique"""
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = Session()
+ u1, u2 = sess.query(User).order_by(User.id)[0:2]
+ q1 = sess.query(Address).with_parent(u1, 'addresses')
+ q2 = sess.query(Address).with_parent(u2, 'addresses')
+ self.assert_compile(
+ q1.union(q2),
+ "SELECT anon_1.addresses_id AS anon_1_addresses_id, "
+ "anon_1.addresses_user_id AS anon_1_addresses_user_id, "
+ "anon_1.addresses_email_address AS "
+ "anon_1_addresses_email_address FROM (SELECT addresses.id AS "
+ "addresses_id, addresses.user_id AS addresses_user_id, "
+ "addresses.email_address AS addresses_email_address FROM "
+ "addresses WHERE :param_1 = addresses.user_id UNION SELECT "
+ "addresses.id AS addresses_id, addresses.user_id AS "
+ "addresses_user_id, addresses.email_address AS addresses_email_address "
+ "FROM addresses WHERE :param_2 = addresses.user_id) AS anon_1",
+ checkparams={u'param_1': 7, u'param_2': 8},
+ )
+
+ def test_unique_binds_or(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ sess = Session()
+ u1, u2 = sess.query(User).order_by(User.id)[0:2]
+
+ self.assert_compile(
+ sess.query(Address).filter(
+ or_(with_parent(u1, 'addresses'), with_parent(u2, 'addresses'))
+ ),
+ "SELECT addresses.id AS addresses_id, addresses.user_id AS "
+ "addresses_user_id, addresses.email_address AS "
+ "addresses_email_address FROM addresses WHERE "
+ ":param_1 = addresses.user_id OR :param_2 = addresses.user_id",
+ checkparams={u'param_1': 7, u'param_2': 8},
+ )
class SynonymTest(QueryTest):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
