See the example ``examples/association/proxied_association.py``.
"""
-import itertools
import operator
import weakref
from .. import exc, orm, util
from ..orm import collections, interfaces
-from ..sql import not_, or_
+from ..sql import or_
from .. import inspect
def _default_getset(self, collection_class):
attr = self.value_attr
_getter = operator.attrgetter(attr)
- getter = lambda target: _getter(target) if target is not None else None
+
+ def getter(target):
+ return _getter(target) if target is not None else None
+
if collection_class is dict:
- setter = lambda o, k, v: setattr(o, attr, v)
+ def setter(o, k, v):
+ setattr(o, attr, v)
else:
- setter = lambda o, v: setattr(o, attr, v)
+ def setter(o, v):
+ setattr(o, attr, v)
return getter, setter
def _new(self, lazy_collection):
getattr(self.target_class, self.value_attr) != obj)
def __repr__(self):
- return "AssociationProxy(%r, %r)" % (self.target_collection, self.value_attr)
+ return "AssociationProxy(%r, %r)" % (
+ self.target_collection, self.value_attr)
class _lazy_collection(object):
return list(self) >= other
def __cmp__(self, other):
- return cmp(list(self), other)
+ return util.cmp(list(self), other)
def __add__(self, iterable):
try:
return dict(self) >= other
def __cmp__(self, other):
- return cmp(dict(self), other)
+ return util.cmp(dict(self), other)
def __repr__(self):
return repr(dict(self.items()))
def _get(self, object):
return self.getter(object)
- def _set(self, object, value):
- return self.setter(object, value)
-
def __len__(self):
return len(self.col)
import copy
import pickle
-from sqlalchemy import *
-from sqlalchemy.orm import *
+from sqlalchemy import Integer, ForeignKey, String, or_, MetaData
+from sqlalchemy.orm import relationship, configure_mappers, mapper, Session,\
+ collections, sessionmaker, aliased, clear_mappers, create_session
+from sqlalchemy import exc
from sqlalchemy.orm.collections import collection, attribute_mapped_collection
-from sqlalchemy.ext.associationproxy import *
+from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.associationproxy import _AssociationList
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing.util import gc_collect
assert True
-
class DefaultTest(_CollectionOperations):
collection_class = None
assert_raises(TypeError, set, [p1.children])
def test_set_comparisons(self):
- Parent, Child = self.Parent, self.Child
+ Parent = self.Parent
p1 = Parent('P1')
p1.children = ['a', 'b', 'c']
is_(set_0 != set_a, False)
def test_set_mutation(self):
- Parent, Child = self.Parent, self.Child
+ Parent = self.Parent
# mutations
for op in ('update', 'intersection_update',
collection_class = ObjectCollection
def test_basic(self):
- Parent, Child = self.Parent, self.Child
+ Parent = self.Parent
p = Parent('p1')
self.assert_(len(list(p.children)) == 0)
getter,
setter,
parent,
- )
+ )
class Parent(object):
children = association_proxy('_children', 'name',
p_copy = copy.copy(p)
del p
gc_collect()
- assert set(p_copy.kids) == set(['c1', 'c2']), p.kids
+ assert set(p_copy.kids) == set(['c1', 'c2']), p_copy.kids
def test_pickle_list(self):
mapper(Parent, self.parents,
singular_keyword = association_proxy("singular", "keyword")
# uselist assoc_proxy -> assoc_proxy -> scalar
- common_keyword_name = association_proxy("user_keywords", "keyword_name")
+ common_keyword_name = association_proxy(
+ "user_keywords", "keyword_name")
class Keyword(cls.Comparable):
def __init__(self, keyword):
user = association_proxy('user_keyword', 'user')
# uselist assoc_proxy -> collection -> assoc_proxy -> scalar object
- # (o2m relationship, associationproxy(m2o relationship, m2o relationship))
+ # (o2m relationship,
+ # associationproxy(m2o relationship, m2o relationship))
singulars = association_proxy("user_keywords", "singular")
class UserKeyword(cls.Comparable):
'quick', 'brown',
'fox', 'jumped', 'over',
'the', 'lazy',
- )
+ )
for ii in range(16):
user = User('user%d' % ii)
"Got None for value of column keywords.singular_id;"):
self._equivalent(
self.session.query(User).filter(
- User.singular_keywords.contains(self.kw)
+ User.singular_keywords.contains(self.kw)
),
self.session.query(User).filter(
- User.singular.has(
- Singular.keywords.contains(self.kw)
- )
+ User.singular.has(
+ Singular.keywords.contains(self.kw)
+ )
),
)
def test_filter_eq_nul_nul(self):
Keyword = self.classes.Keyword
- self._equivalent(self.session.query(Keyword).filter(Keyword.user
- == self.u),
- self.session.query(Keyword).
- filter(Keyword.user_keyword.has(user=self.u)))
+ self._equivalent(
+ self.session.query(Keyword).filter(Keyword.user == self.u),
+ self.session.query(Keyword).
+ filter(Keyword.user_keyword.has(user=self.u))
+ )
def test_filter_ne_nul_nul(self):
Keyword = self.classes.Keyword
- self._equivalent(self.session.query(Keyword).filter(
- Keyword.user != self.u),
- self.session.query(Keyword).filter(
- Keyword.user_keyword.has(Keyword.user != self.u)))
+ self._equivalent(
+ self.session.query(Keyword).filter(Keyword.user != self.u),
+ self.session.query(Keyword).filter(
+ Keyword.user_keyword.has(Keyword.user != self.u)))
def test_filter_eq_null_nul_nul(self):
UserKeyword, Keyword = self.classes.UserKeyword, self.classes.Keyword
self.session.query(User).filter(
User.singular_value == "singular4"),
self.session.query(User).filter(
- User.singular.has(Singular.value == "singular4")))
+ User.singular.has(Singular.value == "singular4")))
def test_filter_ne_None_nul(self):
User = self.classes.User
self._equivalent(
self.session.query(User).filter(User.singular_value.has()),
self.session.query(User).filter(
- User.singular.has(),
- )
+ User.singular.has(),
+ )
)
def test_nothas_nul(self):
self._equivalent(
self.session.query(User).filter(~User.singular_value.has()),
self.session.query(User).filter(
- ~User.singular.has(),
- )
+ ~User.singular.has(),
+ )
)
def test_filter_any_chained(self):
"FROM userkeywords "
"WHERE users.id = userkeywords.user_id AND (EXISTS (SELECT 1 "
"FROM keywords "
- "WHERE keywords.id = userkeywords.keyword_id AND (EXISTS (SELECT 1 "
+ "WHERE keywords.id = userkeywords.keyword_id AND "
+ "(EXISTS (SELECT 1 "
"FROM userkeywords "
- "WHERE keywords.id = userkeywords.keyword_id AND (EXISTS (SELECT 1 "
+ "WHERE keywords.id = userkeywords.keyword_id AND "
+ "(EXISTS (SELECT 1 "
"FROM users "
- "WHERE users.id = userkeywords.user_id AND users.name = :name_1)))))))",
+ "WHERE users.id = userkeywords.user_id AND users.name = :name_1)"
+ "))))))",
checkparams={'name_1': 'user7'}
)
User = self.classes.User
self.assert_compile(
self.session.query(User).join(
- User.keywords.local_attr,
- User.keywords.remote_attr),
+ User.keywords.local_attr,
+ User.keywords.remote_attr),
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users JOIN userkeywords ON users.id = "
User = self.classes.User
self.assert_compile(
self.session.query(User).join(
- *User.keywords.attr),
+ *User.keywords.attr),
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users JOIN userkeywords ON users.id = "