PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-05-26 10:32:33 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
55f930ef3d4e60bed02a2dad16e331fe42cfd12b
sqlalchemy/test/ext/test_associationproxy.py
T
Mike Bayer 1e278de4cc Post black reformatting 
Applied on top of a pure run of black -l 79 in
I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9,  this set of changes
resolves all remaining flake8 conditions for those codes
we have enabled in setup.cfg.

Included are resolutions for all remaining flake8 issues
including shadowed builtins, long lines, import order, unused
imports, duplicate imports, and docstring issues.

Change-Id: I4f72d3ba1380dd601610ff80b8fb06a2aff8b0fe
2019-01-06 18:23:11 -05:00

3209 lines
92 KiB
Python
Raw Blame History

import copy
import pickle
from sqlalchemy import exc
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import or_
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.engine import default
from sqlalchemy.ext.associationproxy import _AssociationList
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import aliased
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import collections
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import create_session
from sqlalchemy.orm import mapper
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.orm.collections import collection
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.mock import call
from sqlalchemy.testing.mock import Mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
class DictCollection(dict):
@collection.appender
def append(self, obj):
self[obj.foo] = obj
@collection.remover
def remove(self, obj):
del self[obj.foo]
class SetCollection(set):
pass
class ListCollection(list):
pass
class ObjectCollection(object):
def __init__(self):
self.values = list()
@collection.appender
def append(self, obj):
self.values.append(obj)
@collection.remover
def remove(self, obj):
self.values.remove(obj)
def __iter__(self):
return iter(self.values)
class AutoFlushTest(fixtures.TablesTest):
@classmethod
def define_tables(cls, metadata):
Table(
"parent",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
)
Table(
"association",
metadata,
Column("parent_id", ForeignKey("parent.id"), primary_key=True),
Column("child_id", ForeignKey("child.id"), primary_key=True),
Column("name", String(50)),
)
Table(
"child",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(50)),
)
def _fixture(self, collection_class, is_dict=False):
class Parent(object):
collection = association_proxy("_collection", "child")
class Child(object):
def __init__(self, name):
self.name = name
class Association(object):
if is_dict:
def __init__(self, key, child):
self.child = child
else:
def __init__(self, child):
self.child = child
mapper(
Parent,
self.tables.parent,
properties={
"_collection": relationship(
Association,
collection_class=collection_class,
backref="parent",
)
},
)
mapper(
Association,
self.tables.association,
properties={"child": relationship(Child, backref="association")},
)
mapper(Child, self.tables.child)
return Parent, Child, Association
def _test_premature_flush(self, collection_class, fn, is_dict=False):
Parent, Child, Association = self._fixture(
collection_class, is_dict=is_dict
)
session = Session(testing.db, autoflush=True, expire_on_commit=True)
p1 = Parent()
c1 = Child("c1")
c2 = Child("c2")
session.add(p1)
session.add(c1)
session.add(c2)
fn(p1.collection, c1)
session.commit()
fn(p1.collection, c2)
session.commit()
is_(c1.association[0].parent, p1)
is_(c2.association[0].parent, p1)
session.close()
def test_list_append(self):
self._test_premature_flush(
list, lambda collection, obj: collection.append(obj)
)
def test_list_extend(self):
self._test_premature_flush(
list, lambda collection, obj: collection.extend([obj])
)
def test_set_add(self):
self._test_premature_flush(
set, lambda collection, obj: collection.add(obj)
)
def test_set_extend(self):
self._test_premature_flush(
set, lambda collection, obj: collection.update([obj])
)
def test_dict_set(self):
def set_(collection, obj):
collection[obj.name] = obj
self._test_premature_flush(
collections.attribute_mapped_collection("name"), set_, is_dict=True
)
class _CollectionOperations(fixtures.TestBase):
def setup(self):
collection_class = self.collection_class
metadata = MetaData(testing.db)
parents_table = Table(
"Parent",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(128)),
)
children_table = Table(
"Children",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("Parent.id")),
Column("foo", String(128)),
Column("name", String(128)),
)
class Parent(object):
children = association_proxy("_children", "name")
def __init__(self, name):
self.name = name
class Child(object):
if collection_class and issubclass(collection_class, dict):
def __init__(self, foo, name):
self.foo = foo
self.name = name
else:
def __init__(self, name):
self.name = name
mapper(
Parent,
parents_table,
properties={
"_children": relationship(
Child,
lazy="joined",
backref="parent",
collection_class=collection_class,
)
},
)
mapper(Child, children_table)
metadata.create_all()
self.metadata = metadata
self.session = create_session()
self.Parent, self.Child = Parent, Child
def teardown(self):
self.metadata.drop_all()
def roundtrip(self, obj):
if obj not in self.session:
self.session.add(obj)
self.session.flush()
id_, type_ = obj.id, type(obj)
self.session.expunge_all()
return self.session.query(type_).get(id_)
def _test_sequence_ops(self):
Parent, Child = self.Parent, self.Child
p1 = Parent("P1")
def assert_index(expected, value, *args):
"""Assert index of child value is equal to expected.
If expected is None, assert that index raises ValueError.
"""
try:
index = p1.children.index(value, *args)
except ValueError:
self.assert_(expected is None)
else:
self.assert_(expected is not None)
self.assert_(index == expected)
self.assert_(not p1._children)
self.assert_(not p1.children)
ch = Child("regular")
p1._children.append(ch)
self.assert_(ch in p1._children)
self.assert_(len(p1._children) == 1)
self.assert_(p1.children)
self.assert_(len(p1.children) == 1)
self.assert_(ch not in p1.children)
self.assert_("regular" in p1.children)
assert_index(0, "regular")
assert_index(None, "regular", 1)
p1.children.append("proxied")
self.assert_("proxied" in p1.children)
self.assert_("proxied" not in p1._children)
self.assert_(len(p1.children) == 2)
self.assert_(len(p1._children) == 2)
self.assert_(p1._children[0].name == "regular")
self.assert_(p1._children[1].name == "proxied")
assert_index(0, "regular")
assert_index(1, "proxied")
assert_index(1, "proxied", 1)
assert_index(None, "proxied", 0, 1)
del p1._children[1]
self.assert_(len(p1._children) == 1)
self.assert_(len(p1.children) == 1)
self.assert_(p1._children[0] == ch)
assert_index(None, "proxied")
del p1.children[0]
self.assert_(len(p1._children) == 0)
self.assert_(len(p1.children) == 0)
assert_index(None, "regular")
p1.children = ["a", "b", "c"]
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
assert_index(0, "a")
assert_index(1, "b")
assert_index(2, "c")
del ch
p1 = self.roundtrip(p1)
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
assert_index(0, "a")
assert_index(1, "b")
assert_index(2, "c")
popped = p1.children.pop()
self.assert_(len(p1.children) == 2)
self.assert_(popped not in p1.children)
assert_index(None, popped)
p1 = self.roundtrip(p1)
self.assert_(len(p1.children) == 2)
self.assert_(popped not in p1.children)
assert_index(None, popped)
p1.children[1] = "changed-in-place"
self.assert_(p1.children[1] == "changed-in-place")
assert_index(1, "changed-in-place")
assert_index(None, "b")
inplace_id = p1._children[1].id
p1 = self.roundtrip(p1)
self.assert_(p1.children[1] == "changed-in-place")
assert p1._children[1].id == inplace_id
p1.children.append("changed-in-place")
self.assert_(p1.children.count("changed-in-place") == 2)
assert_index(1, "changed-in-place")
p1.children.remove("changed-in-place")
self.assert_(p1.children.count("changed-in-place") == 1)
assert_index(1, "changed-in-place")
p1 = self.roundtrip(p1)
self.assert_(p1.children.count("changed-in-place") == 1)
assert_index(1, "changed-in-place")
p1._children = []
self.assert_(len(p1.children) == 0)
assert_index(None, "changed-in-place")
after = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
p1.children = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
self.assert_(len(p1.children) == 10)
self.assert_([c.name for c in p1._children] == after)
for i, val in enumerate(after):
assert_index(i, val)
p1.children[2:6] = ["x"] * 4
after = ["a", "b", "x", "x", "x", "x", "g", "h", "i", "j"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
assert_index(2, "x")
assert_index(3, "x", 3)
assert_index(None, "x", 6)
p1.children[2:6] = ["y"]
after = ["a", "b", "y", "g", "h", "i", "j"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
assert_index(2, "y")
assert_index(None, "y", 3)
p1.children[2:3] = ["z"] * 4
after = ["a", "b", "z", "z", "z", "z", "g", "h", "i", "j"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children[2::2] = ["O"] * 4
after = ["a", "b", "O", "z", "O", "z", "O", "h", "O", "j"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
assert_raises(TypeError, set, [p1.children])
p1.children *= 0
after = []
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children += ["a", "b"]
after = ["a", "b"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children[:] = ["d", "e"]
after = ["d", "e"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children[:] = ["a", "b"]
p1.children += ["c"]
after = ["a", "b", "c"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children *= 1
after = ["a", "b", "c"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children *= 2
after = ["a", "b", "c", "a", "b", "c"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
p1.children = ["a"]
after = ["a"]
self.assert_(p1.children == after)
self.assert_([c.name for c in p1._children] == after)
self.assert_((p1.children * 2) == ["a", "a"])
self.assert_((2 * p1.children) == ["a", "a"])
self.assert_((p1.children * 0) == [])
self.assert_((0 * p1.children) == [])
self.assert_((p1.children + ["b"]) == ["a", "b"])
self.assert_((["b"] + p1.children) == ["b", "a"])
try:
p1.children + 123
assert False
except TypeError:
assert True
class DefaultTest(_CollectionOperations):
collection_class = None
def test_sequence_ops(self):
self._test_sequence_ops()
class ListTest(_CollectionOperations):
collection_class = list
def test_sequence_ops(self):
self._test_sequence_ops()
class CustomDictTest(_CollectionOperations):
collection_class = DictCollection
def test_mapping_ops(self):
Parent, Child = self.Parent, self.Child
p1 = Parent("P1")
self.assert_(not p1._children)
self.assert_(not p1.children)
ch = Child("a", "regular")
p1._children.append(ch)
self.assert_(ch in list(p1._children.values()))
self.assert_(len(p1._children) == 1)
self.assert_(p1.children)
self.assert_(len(p1.children) == 1)
self.assert_(ch not in p1.children)
self.assert_("a" in p1.children)
self.assert_(p1.children["a"] == "regular")
self.assert_(p1._children["a"] == ch)
p1.children["b"] = "proxied"
self.assert_("proxied" in list(p1.children.values()))
self.assert_("b" in p1.children)
self.assert_("proxied" not in p1._children)
self.assert_(len(p1.children) == 2)
self.assert_(len(p1._children) == 2)
self.assert_(p1._children["a"].name == "regular")
self.assert_(p1._children["b"].name == "proxied")
del p1._children["b"]
self.assert_(len(p1._children) == 1)
self.assert_(len(p1.children) == 1)
self.assert_(p1._children["a"] == ch)
del p1.children["a"]
self.assert_(len(p1._children) == 0)
self.assert_(len(p1.children) == 0)
p1.children = {"d": "v d", "e": "v e", "f": "v f"}
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
self.assert_(set(p1.children) == set(["d", "e", "f"]))
del ch
p1 = self.roundtrip(p1)
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
p1.children["e"] = "changed-in-place"
self.assert_(p1.children["e"] == "changed-in-place")
inplace_id = p1._children["e"].id
p1 = self.roundtrip(p1)
self.assert_(p1.children["e"] == "changed-in-place")
self.assert_(p1._children["e"].id == inplace_id)
p1._children = {}
self.assert_(len(p1.children) == 0)
try:
p1._children = []
self.assert_(False)
except TypeError:
self.assert_(True)
try:
p1._children = None
self.assert_(False)
except TypeError:
self.assert_(True)
assert_raises(TypeError, set, [p1.children])
class SetTest(_CollectionOperations):
collection_class = set
def test_set_operations(self):
Parent, Child = self.Parent, self.Child
p1 = Parent("P1")
self.assert_(not p1._children)
self.assert_(not p1.children)
ch1 = Child("regular")
p1._children.add(ch1)
self.assert_(ch1 in p1._children)
self.assert_(len(p1._children) == 1)
self.assert_(p1.children)
self.assert_(len(p1.children) == 1)
self.assert_(ch1 not in p1.children)
self.assert_("regular" in p1.children)
p1.children.add("proxied")
self.assert_("proxied" in p1.children)
self.assert_("proxied" not in p1._children)
self.assert_(len(p1.children) == 2)
self.assert_(len(p1._children) == 2)
self.assert_(
set([o.name for o in p1._children]) == set(["regular", "proxied"])
)
ch2 = None
for o in p1._children:
if o.name == "proxied":
ch2 = o
break
p1._children.remove(ch2)
self.assert_(len(p1._children) == 1)
self.assert_(len(p1.children) == 1)
self.assert_(p1._children == set([ch1]))
p1.children.remove("regular")
self.assert_(len(p1._children) == 0)
self.assert_(len(p1.children) == 0)
p1.children = ["a", "b", "c"]
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
del ch1
p1 = self.roundtrip(p1)
self.assert_(len(p1._children) == 3)
self.assert_(len(p1.children) == 3)
self.assert_("a" in p1.children)
self.assert_("b" in p1.children)
self.assert_("d" not in p1.children)
self.assert_(p1.children == set(["a", "b", "c"]))
assert_raises(KeyError, p1.children.remove, "d")
self.assert_(len(p1.children) == 3)
p1.children.discard("d")
self.assert_(len(p1.children) == 3)
p1 = self.roundtrip(p1)
self.assert_(len(p1.children) == 3)
popped = p1.children.pop()
self.assert_(len(p1.children) == 2)
self.assert_(popped not in p1.children)
p1 = self.roundtrip(p1)
self.assert_(len(p1.children) == 2)
self.assert_(popped not in p1.children)
p1.children = ["a", "b", "c"]
p1 = self.roundtrip(p1)
self.assert_(p1.children == set(["a", "b", "c"]))
p1.children.discard("b")
p1 = self.roundtrip(p1)
self.assert_(p1.children == set(["a", "c"]))
p1.children.remove("a")
p1 = self.roundtrip(p1)
self.assert_(p1.children == set(["c"]))
p1._children = set()
self.assert_(len(p1.children) == 0)
try:
p1._children = []
self.assert_(False)
except TypeError:
self.assert_(True)
try:
p1._children = None
self.assert_(False)
except TypeError:
self.assert_(True)
assert_raises(TypeError, set, [p1.children])
def test_set_comparisons(self):
Parent = self.Parent
p1 = Parent("P1")
p1.children = ["a", "b", "c"]
control = set(["a", "b", "c"])
for other in (
set(["a", "b", "c"]),
set(["a", "b", "c", "d"]),
set(["a"]),
set(["a", "b"]),
set(["c", "d"]),
set(["e", "f", "g"]),
set(),
):
eq_(p1.children.union(other), control.union(other))
eq_(p1.children.difference(other), control.difference(other))
eq_((p1.children - other), (control - other))
eq_(p1.children.intersection(other), control.intersection(other))
eq_(
p1.children.symmetric_difference(other),
control.symmetric_difference(other),
)
eq_(p1.children.issubset(other), control.issubset(other))
eq_(p1.children.issuperset(other), control.issuperset(other))
self.assert_((p1.children == other) == (control == other))
self.assert_((p1.children != other) == (control != other))
self.assert_((p1.children < other) == (control < other))
self.assert_((p1.children <= other) == (control <= other))
self.assert_((p1.children > other) == (control > other))
self.assert_((p1.children >= other) == (control >= other))
@testing.requires.python_fixed_issue_8743
def test_set_comparison_empty_to_empty(self):
# test issue #3265 which was fixed in Python version 2.7.8
Parent = self.Parent
p1 = Parent("P1")
p1.children = []
p2 = Parent("P2")
p2.children = []
set_0 = set()
set_a = p1.children
set_b = p2.children
is_(set_a == set_a, True)
is_(set_a == set_b, True)
is_(set_a == set_0, True)
is_(set_0 == set_a, True)
is_(set_a != set_a, False)
is_(set_a != set_b, False)
is_(set_a != set_0, False)
is_(set_0 != set_a, False)
def test_set_mutation(self):
Parent = self.Parent
# mutations
for op in (
"update",
"intersection_update",
"difference_update",
"symmetric_difference_update",
):
for base in (["a", "b", "c"], []):
for other in (
set(["a", "b", "c"]),
set(["a", "b", "c", "d"]),
set(["a"]),
set(["a", "b"]),
set(["c", "d"]),
set(["e", "f", "g"]),
set(),
):
p = Parent("p")
p.children = base[:]
control = set(base[:])
getattr(p.children, op)(other)
getattr(control, op)(other)
try:
self.assert_(p.children == control)
except Exception:
print("Test %s.%s(%s):" % (set(base), op, other))
print("want", repr(control))
print("got", repr(p.children))
raise
p = self.roundtrip(p)
try:
self.assert_(p.children == control)
except Exception:
print("Test %s.%s(%s):" % (base, op, other))
print("want", repr(control))
print("got", repr(p.children))
raise
# in-place mutations
for op in ("|=", "-=", "&=", "^="):
for base in (["a", "b", "c"], []):
for other in (
set(["a", "b", "c"]),
set(["a", "b", "c", "d"]),
set(["a"]),
set(["a", "b"]),
set(["c", "d"]),
set(["e", "f", "g"]),
frozenset(["e", "f", "g"]),
set(),
):
p = Parent("p")
p.children = base[:]
control = set(base[:])
exec("p.children %s other" % op)
exec("control %s other" % op)
try:
self.assert_(p.children == control)
except Exception:
print("Test %s %s %s:" % (set(base), op, other))
print("want", repr(control))
print("got", repr(p.children))
raise
p = self.roundtrip(p)
try:
self.assert_(p.children == control)
except Exception:
print("Test %s %s %s:" % (base, op, other))
print("want", repr(control))
print("got", repr(p.children))
raise
class CustomSetTest(SetTest):
collection_class = SetCollection
class CustomObjectTest(_CollectionOperations):
collection_class = ObjectCollection
def test_basic(self):
Parent = self.Parent
p = Parent("p1")
self.assert_(len(list(p.children)) == 0)
p.children.append("child")
self.assert_(len(list(p.children)) == 1)
p = self.roundtrip(p)
self.assert_(len(list(p.children)) == 1)
# We didn't provide an alternate _AssociationList implementation
# for our ObjectCollection, so indexing will fail.
assert_raises(TypeError, p.children.__getitem__, 1)
class ProxyFactoryTest(ListTest):
def setup(self):
metadata = MetaData(testing.db)
parents_table = Table(
"Parent",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(128)),
)
children_table = Table(
"Children",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("Parent.id")),
Column("foo", String(128)),
Column("name", String(128)),
)
class CustomProxy(_AssociationList):
def __init__(self, lazy_collection, creator, value_attr, parent):
getter, setter = parent._default_getset(lazy_collection)
_AssociationList.__init__(
self, lazy_collection, creator, getter, setter, parent
)
class Parent(object):
children = association_proxy(
"_children",
"name",
proxy_factory=CustomProxy,
proxy_bulk_set=CustomProxy.extend,
)
def __init__(self, name):
self.name = name
class Child(object):
def __init__(self, name):
self.name = name
mapper(
Parent,
parents_table,
properties={
"_children": relationship(
Child, lazy="joined", collection_class=list
)
},
)
mapper(Child, children_table)
metadata.create_all()
self.metadata = metadata
self.session = create_session()
self.Parent, self.Child = Parent, Child
def test_sequence_ops(self):
self._test_sequence_ops()
class ScalarTest(fixtures.TestBase):
@testing.provide_metadata
def test_scalar_proxy(self):
metadata = self.metadata
parents_table = Table(
"Parent",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(128)),
)
children_table = Table(
"Children",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("Parent.id")),
Column("foo", String(128)),
Column("bar", String(128)),
Column("baz", String(128)),
)
class Parent(object):
foo = association_proxy("child", "foo")
bar = association_proxy(
"child", "bar", creator=lambda v: Child(bar=v)
)
baz = association_proxy(
"child", "baz", creator=lambda v: Child(baz=v)
)
def __init__(self, name):
self.name = name
class Child(object):
def __init__(self, **kw):
for attr in kw:
setattr(self, attr, kw[attr])
mapper(
Parent,
parents_table,
properties={
"child": relationship(
Child, lazy="joined", backref="parent", uselist=False
)
},
)
mapper(Child, children_table)
metadata.create_all()
session = create_session()
def roundtrip(obj):
if obj not in session:
session.add(obj)
session.flush()
id_, type_ = obj.id, type(obj)
session.expunge_all()
return session.query(type_).get(id_)
p = Parent("p")
eq_(p.child, None)
eq_(p.foo, None)
p.child = Child(foo="a", bar="b", baz="c")
self.assert_(p.foo == "a")
self.assert_(p.bar == "b")
self.assert_(p.baz == "c")
p.bar = "x"
self.assert_(p.foo == "a")
self.assert_(p.bar == "x")
self.assert_(p.baz == "c")
p = roundtrip(p)
self.assert_(p.foo == "a")
self.assert_(p.bar == "x")
self.assert_(p.baz == "c")
p.child = None
eq_(p.foo, None)
# Bogus creator for this scalar type
assert_raises(TypeError, setattr, p, "foo", "zzz")
p.bar = "yyy"
self.assert_(p.foo is None)
self.assert_(p.bar == "yyy")
self.assert_(p.baz is None)
del p.child
p = roundtrip(p)
self.assert_(p.child is None)
p.baz = "xxx"
self.assert_(p.foo is None)
self.assert_(p.bar is None)
self.assert_(p.baz == "xxx")
p = roundtrip(p)
self.assert_(p.foo is None)
self.assert_(p.bar is None)
self.assert_(p.baz == "xxx")
# Ensure an immediate __set__ works.
p2 = Parent("p2")
p2.bar = "quux"
@testing.provide_metadata
def test_empty_scalars(self):
metadata = self.metadata
a = Table(
"a",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String(50)),
)
a2b = Table(
"a2b",
metadata,
Column("id", Integer, primary_key=True),
Column("id_a", Integer, ForeignKey("a.id")),
Column("id_b", Integer, ForeignKey("b.id")),
Column("name", String(50)),
)
b = Table(
"b",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String(50)),
)
class A(object):
a2b_name = association_proxy("a2b_single", "name")
b_single = association_proxy("a2b_single", "b")
class A2B(object):
pass
class B(object):
pass
mapper(
A, a, properties=dict(a2b_single=relationship(A2B, uselist=False))
)
mapper(A2B, a2b, properties=dict(b=relationship(B)))
mapper(B, b)
a1 = A()
assert a1.a2b_name is None
assert a1.b_single is None
def test_custom_getset(self):
metadata = MetaData()
p = Table(
"p",
metadata,
Column("id", Integer, primary_key=True),
Column("cid", Integer, ForeignKey("c.id")),
)
c = Table(
"c",
metadata,
Column("id", Integer, primary_key=True),
Column("foo", String(128)),
)
get = Mock()
set_ = Mock()
class Parent(object):
foo = association_proxy(
"child", "foo", getset_factory=lambda cc, parent: (get, set_)
)
class Child(object):
def __init__(self, foo):
self.foo = foo
mapper(Parent, p, properties={"child": relationship(Child)})
mapper(Child, c)
p1 = Parent()
eq_(p1.foo, get(None))
p1.child = child = Child(foo="x")
eq_(p1.foo, get(child))
p1.foo = "y"
eq_(set_.mock_calls, [call(child, "y")])
class LazyLoadTest(fixtures.TestBase):
def setup(self):
metadata = MetaData(testing.db)
parents_table = Table(
"Parent",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(128)),
)
children_table = Table(
"Children",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("Parent.id")),
Column("foo", String(128)),
Column("name", String(128)),
)
class Parent(object):
children = association_proxy("_children", "name")
def __init__(self, name):
self.name = name
class Child(object):
def __init__(self, name):
self.name = name
mapper(Child, children_table)
metadata.create_all()
self.metadata = metadata
self.session = create_session()
self.Parent, self.Child = Parent, Child
self.table = parents_table
def teardown(self):
self.metadata.drop_all()
def roundtrip(self, obj):
self.session.add(obj)
self.session.flush()
id_, type_ = obj.id, type(obj)
self.session.expunge_all()
return self.session.query(type_).get(id_)
def test_lazy_list(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(
Child, lazy="select", collection_class=list
)
},
)
p = Parent("p")
p.children = ["a", "b", "c"]
p = self.roundtrip(p)
# Is there a better way to ensure that the association_proxy
# didn't convert a lazy load to an eager load? This does work though.
self.assert_("_children" not in p.__dict__)
self.assert_(len(p._children) == 3)
self.assert_("_children" in p.__dict__)
def test_eager_list(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(
Child, lazy="joined", collection_class=list
)
},
)
p = Parent("p")
p.children = ["a", "b", "c"]
p = self.roundtrip(p)
self.assert_("_children" in p.__dict__)
self.assert_(len(p._children) == 3)
def test_slicing_list(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(
Child, lazy="select", collection_class=list
)
},
)
p = Parent("p")
p.children = ["a", "b", "c"]
p = self.roundtrip(p)
self.assert_(len(p._children) == 3)
eq_("b", p.children[1])
eq_(["b", "c"], p.children[-2:])
def test_lazy_scalar(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(Child, lazy="select", uselist=False)
},
)
p = Parent("p")
p.children = "value"
p = self.roundtrip(p)
self.assert_("_children" not in p.__dict__)
self.assert_(p._children is not None)
def test_eager_scalar(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(Child, lazy="joined", uselist=False)
},
)
p = Parent("p")
p.children = "value"
p = self.roundtrip(p)
self.assert_("_children" in p.__dict__)
self.assert_(p._children is not None)
class Parent(object):
def __init__(self, name):
self.name = name
class Child(object):
def __init__(self, name):
self.name = name
class KVChild(object):
def __init__(self, name, value):
self.name = name
self.value = value
class ReconstitutionTest(fixtures.TestBase):
def setup(self):
metadata = MetaData(testing.db)
parents = Table(
"parents",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(30)),
)
children = Table(
"children",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("parent_id", Integer, ForeignKey("parents.id")),
Column("name", String(30)),
)
metadata.create_all()
parents.insert().execute(name="p1")
self.metadata = metadata
self.parents = parents
self.children = children
Parent.kids = association_proxy("children", "name")
def teardown(self):
self.metadata.drop_all()
clear_mappers()
def test_weak_identity_map(self):
mapper(
Parent, self.parents, properties=dict(children=relationship(Child))
)
mapper(Child, self.children)
session = create_session(weak_identity_map=True)
def add_child(parent_name, child_name):
parent = session.query(Parent).filter_by(name=parent_name).one()
parent.kids.append(child_name)
add_child("p1", "c1")
gc_collect()
add_child("p1", "c2")
session.flush()
p = session.query(Parent).filter_by(name="p1").one()
assert set(p.kids) == set(["c1", "c2"]), p.kids
def test_copy(self):
mapper(
Parent, self.parents, properties=dict(children=relationship(Child))
)
mapper(Child, self.children)
p = Parent("p1")
p.kids.extend(["c1", "c2"])
p_copy = copy.copy(p)
del p
gc_collect()
assert set(p_copy.kids) == set(["c1", "c2"]), p_copy.kids
def test_pickle_list(self):
mapper(
Parent, self.parents, properties=dict(children=relationship(Child))
)
mapper(Child, self.children)
p = Parent("p1")
p.kids.extend(["c1", "c2"])
r1 = pickle.loads(pickle.dumps(p))
assert r1.kids == ["c1", "c2"]
# can't do this without parent having a cycle
# r2 = pickle.loads(pickle.dumps(p.kids))
# assert r2 == ['c1', 'c2']
def test_pickle_set(self):
mapper(
Parent,
self.parents,
properties=dict(
children=relationship(Child, collection_class=set)
),
)
mapper(Child, self.children)
p = Parent("p1")
p.kids.update(["c1", "c2"])
r1 = pickle.loads(pickle.dumps(p))
assert r1.kids == set(["c1", "c2"])
# can't do this without parent having a cycle
# r2 = pickle.loads(pickle.dumps(p.kids))
# assert r2 == set(['c1', 'c2'])
def test_pickle_dict(self):
mapper(
Parent,
self.parents,
properties=dict(
children=relationship(
KVChild,
collection_class=collections.mapped_collection(
PickleKeyFunc("name")
),
)
),
)
mapper(KVChild, self.children)
p = Parent("p1")
p.kids.update({"c1": "v1", "c2": "v2"})
assert p.kids == {"c1": "c1", "c2": "c2"}
r1 = pickle.loads(pickle.dumps(p))
assert r1.kids == {"c1": "c1", "c2": "c2"}
# can't do this without parent having a cycle
# r2 = pickle.loads(pickle.dumps(p.kids))
# assert r2 == {'c1': 'c1', 'c2': 'c2'}
class PickleKeyFunc(object):
def __init__(self, name):
self.name = name
def __call__(self, obj):
return getattr(obj, self.name)
class ComparatorTest(fixtures.MappedTest, AssertsCompiledSQL):
__dialect__ = "default"
run_inserts = "once"
run_deletes = None
run_setup_mappers = "once"
run_setup_classes = "once"
@classmethod
def define_tables(cls, metadata):
Table(
"userkeywords",
metadata,
Column(
"keyword_id",
Integer,
ForeignKey("keywords.id"),
primary_key=True,
),
Column("user_id", Integer, ForeignKey("users.id")),
Column("value", String(50)),
)
Table(
"users",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("name", String(64)),
Column("singular_id", Integer, ForeignKey("singular.id")),
)
Table(
"keywords",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("keyword", String(64)),
Column("singular_id", Integer, ForeignKey("singular.id")),
)
Table(
"singular",
metadata,
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("value", String(50)),
)
@classmethod
def setup_classes(cls):
class User(cls.Comparable):
def __init__(self, name):
self.name = name
# o2m -> m2o
# uselist -> nonuselist
keywords = association_proxy(
"user_keywords",
"keyword",
creator=lambda k: UserKeyword(keyword=k),
)
# m2o -> o2m
# nonuselist -> uselist
singular_keywords = association_proxy("singular", "keywords")
# m2o -> scalar
# nonuselist
singular_value = association_proxy("singular", "value")
# o2m -> scalar
singular_collection = association_proxy("user_keywords", "value")
# uselist assoc_proxy -> assoc_proxy -> obj
common_users = association_proxy("user_keywords", "common_users")
# non uselist assoc_proxy -> assoc_proxy -> obj
common_singular = association_proxy("singular", "keyword")
# non uselist assoc_proxy -> assoc_proxy -> scalar
singular_keyword = association_proxy("singular", "keyword")
# uselist assoc_proxy -> assoc_proxy -> scalar
common_keyword_name = association_proxy(
"user_keywords", "keyword_name"
)
class Keyword(cls.Comparable):
def __init__(self, keyword):
self.keyword = keyword
# o2o -> m2o
# nonuselist -> nonuselist
user = association_proxy("user_keyword", "user")
# uselist assoc_proxy -> collection -> assoc_proxy -> scalar object
# (o2m relationship,
# associationproxy(m2o relationship, m2o relationship))
singulars = association_proxy("user_keywords", "singular")
class UserKeyword(cls.Comparable):
def __init__(self, user=None, keyword=None):
self.user = user
self.keyword = keyword
common_users = association_proxy("keyword", "user")
keyword_name = association_proxy("keyword", "keyword")
singular = association_proxy("user", "singular")
class Singular(cls.Comparable):
def __init__(self, value=None):
self.value = value
keyword = association_proxy("keywords", "keyword")
@classmethod
def setup_mappers(cls):
(
users,
Keyword,
UserKeyword,
singular,
userkeywords,
User,
keywords,
Singular,
) = (
cls.tables.users,
cls.classes.Keyword,
cls.classes.UserKeyword,
cls.tables.singular,
cls.tables.userkeywords,
cls.classes.User,
cls.tables.keywords,
cls.classes.Singular,
)
mapper(User, users, properties={"singular": relationship(Singular)})
mapper(
Keyword,
keywords,
properties={
"user_keyword": relationship(UserKeyword, uselist=False),
"user_keywords": relationship(UserKeyword),
},
)
mapper(
UserKeyword,
userkeywords,
properties={
"user": relationship(User, backref="user_keywords"),
"keyword": relationship(Keyword),
},
)
mapper(
Singular, singular, properties={"keywords": relationship(Keyword)}
)
@classmethod
def insert_data(cls):
UserKeyword, User, Keyword, Singular = (
cls.classes.UserKeyword,
cls.classes.User,
cls.classes.Keyword,
cls.classes.Singular,
)
session = sessionmaker()()
words = ("quick", "brown", "fox", "jumped", "over", "the", "lazy")
for ii in range(16):
user = User("user%d" % ii)
if ii % 2 == 0:
user.singular = Singular(
value=("singular%d" % ii) if ii % 4 == 0 else None
)
session.add(user)
for jj in words[(ii % len(words)) : ((ii + 3) % len(words))]:
k = Keyword(jj)
user.keywords.append(k)
if ii % 2 == 0:
user.singular.keywords.append(k)
user.user_keywords[-1].value = "singular%d" % ii
orphan = Keyword("orphan")
orphan.user_keyword = UserKeyword(keyword=orphan, user=None)
session.add(orphan)
keyword_with_nothing = Keyword("kwnothing")
session.add(keyword_with_nothing)
session.commit()
cls.u = user
cls.kw = user.keywords[0]
cls.session = session
def _equivalent(self, q_proxy, q_direct):
proxy_sql = q_proxy.statement.compile(dialect=default.DefaultDialect())
direct_sql = q_direct.statement.compile(
dialect=default.DefaultDialect()
)
eq_(str(proxy_sql), str(direct_sql))
eq_(q_proxy.all(), q_direct.all())
def test_filter_any_criterion_ul_scalar(self):
UserKeyword, User = self.classes.UserKeyword, self.classes.User
q1 = self.session.query(User).filter(
User.singular_collection.any(UserKeyword.value == "singular8")
)
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users "
"WHERE EXISTS (SELECT 1 "
"FROM userkeywords "
"WHERE users.id = userkeywords.user_id AND "
"userkeywords.value = :value_1)",
checkparams={"value_1": "singular8"},
)
q2 = self.session.query(User).filter(
User.user_keywords.any(UserKeyword.value == "singular8")
)
self._equivalent(q1, q2)
def test_filter_any_kwarg_ul_nul(self):
UserKeyword, User = self.classes.UserKeyword, self.classes.User
self._equivalent(
self.session.query(User).filter(
User.keywords.any(keyword="jumped")
),
self.session.query(User).filter(
User.user_keywords.any(
UserKeyword.keyword.has(keyword="jumped")
)
),
)
def test_filter_has_kwarg_nul_nul(self):
UserKeyword, Keyword = self.classes.UserKeyword, self.classes.Keyword
self._equivalent(
self.session.query(Keyword).filter(Keyword.user.has(name="user2")),
self.session.query(Keyword).filter(
Keyword.user_keyword.has(UserKeyword.user.has(name="user2"))
),
)
def test_filter_has_kwarg_nul_ul(self):
User, Singular = self.classes.User, self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_keywords.any(keyword="jumped")
),
self.session.query(User).filter(
User.singular.has(Singular.keywords.any(keyword="jumped"))
),
)
def test_filter_any_criterion_ul_nul(self):
UserKeyword, User, Keyword = (
self.classes.UserKeyword,
self.classes.User,
self.classes.Keyword,
)
self._equivalent(
self.session.query(User).filter(
User.keywords.any(Keyword.keyword == "jumped")
),
self.session.query(User).filter(
User.user_keywords.any(
UserKeyword.keyword.has(Keyword.keyword == "jumped")
)
),
)
def test_filter_has_criterion_nul_nul(self):
UserKeyword, User, Keyword = (
self.classes.UserKeyword,
self.classes.User,
self.classes.Keyword,
)
self._equivalent(
self.session.query(Keyword).filter(
Keyword.user.has(User.name == "user2")
),
self.session.query(Keyword).filter(
Keyword.user_keyword.has(
UserKeyword.user.has(User.name == "user2")
)
),
)
def test_filter_any_criterion_nul_ul(self):
User, Keyword, Singular = (
self.classes.User,
self.classes.Keyword,
self.classes.Singular,
)
self._equivalent(
self.session.query(User).filter(
User.singular_keywords.any(Keyword.keyword == "jumped")
),
self.session.query(User).filter(
User.singular.has(
Singular.keywords.any(Keyword.keyword == "jumped")
)
),
)
def test_filter_contains_ul_nul(self):
User = self.classes.User
self._equivalent(
self.session.query(User).filter(User.keywords.contains(self.kw)),
self.session.query(User).filter(
User.user_keywords.any(keyword=self.kw)
),
)
def test_filter_contains_nul_ul(self):
User, Singular = self.classes.User, self.classes.Singular
with expect_warnings(
"Got None for value of column keywords.singular_id;"
):
self._equivalent(
self.session.query(User).filter(
User.singular_keywords.contains(self.kw)
),
self.session.query(User).filter(
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)
),
)
def test_filter_ne_nul_nul(self):
Keyword = self.classes.Keyword
UserKeyword = self.classes.UserKeyword
self._equivalent(
self.session.query(Keyword).filter(Keyword.user != self.u),
self.session.query(Keyword).filter(
Keyword.user_keyword.has(UserKeyword.user != self.u)
),
)
def test_filter_eq_null_nul_nul(self):
UserKeyword, Keyword = self.classes.UserKeyword, self.classes.Keyword
self._equivalent(
self.session.query(Keyword).filter(Keyword.user == None), # noqa
self.session.query(Keyword).filter(
or_(
Keyword.user_keyword.has(UserKeyword.user == None),
Keyword.user_keyword == None,
)
),
)
def test_filter_ne_null_nul_nul(self):
UserKeyword, Keyword = self.classes.UserKeyword, self.classes.Keyword
self._equivalent(
self.session.query(Keyword).filter(Keyword.user != None), # noqa
self.session.query(Keyword).filter(
Keyword.user_keyword.has(UserKeyword.user != None)
),
)
def test_filter_object_eq_None_nul(self):
UserKeyword = self.classes.UserKeyword
User = self.classes.User
self._equivalent(
self.session.query(UserKeyword).filter(
UserKeyword.singular == None
), # noqa
self.session.query(UserKeyword).filter(
or_(
UserKeyword.user.has(User.singular == None),
UserKeyword.user_id == None,
)
),
)
def test_filter_column_eq_None_nul(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_value == None
), # noqa
self.session.query(User).filter(
or_(
User.singular.has(Singular.value == None),
User.singular == None,
)
),
)
def test_filter_object_ne_value_nul(self):
UserKeyword = self.classes.UserKeyword
User = self.classes.User
Singular = self.classes.Singular
s4 = self.session.query(Singular).filter_by(value="singular4").one()
self._equivalent(
self.session.query(UserKeyword).filter(UserKeyword.singular != s4),
self.session.query(UserKeyword).filter(
UserKeyword.user.has(User.singular != s4)
),
)
def test_filter_column_ne_value_nul(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_value != "singular4"
),
self.session.query(User).filter(
User.singular.has(Singular.value != "singular4")
),
)
def test_filter_eq_value_nul(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_value == "singular4"
),
self.session.query(User).filter(
User.singular.has(Singular.value == "singular4")
),
)
def test_filter_ne_None_nul(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_value != None
), # noqa
self.session.query(User).filter(
User.singular.has(Singular.value != None)
),
)
def test_has_nul(self):
# a special case where we provide an empty has() on a
# non-object-targeted association proxy.
User = self.classes.User
self.classes.Singular
self._equivalent(
self.session.query(User).filter(User.singular_value.has()),
self.session.query(User).filter(User.singular.has()),
)
def test_nothas_nul(self):
# a special case where we provide an empty has() on a
# non-object-targeted association proxy.
User = self.classes.User
self.classes.Singular
self._equivalent(
self.session.query(User).filter(~User.singular_value.has()),
self.session.query(User).filter(~User.singular.has()),
)
def test_filter_any_chained(self):
User = self.classes.User
UserKeyword, User = self.classes.UserKeyword, self.classes.User
Keyword = self.classes.Keyword
q1 = self.session.query(User).filter(
User.common_users.any(User.name == "user7")
)
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users "
"WHERE EXISTS (SELECT 1 "
"FROM userkeywords "
"WHERE users.id = userkeywords.user_id AND (EXISTS (SELECT 1 "
"FROM keywords "
"WHERE keywords.id = userkeywords.keyword_id AND "
"(EXISTS (SELECT 1 "
"FROM userkeywords "
"WHERE keywords.id = userkeywords.keyword_id AND "
"(EXISTS (SELECT 1 "
"FROM users "
"WHERE users.id = userkeywords.user_id AND users.name = :name_1)"
"))))))",
checkparams={"name_1": "user7"},
)
q2 = self.session.query(User).filter(
User.user_keywords.any(
UserKeyword.keyword.has(
Keyword.user_keyword.has(
UserKeyword.user.has(User.name == "user7")
)
)
)
)
self._equivalent(q1, q2)
def test_filter_has_chained_has_to_any(self):
User = self.classes.User
Singular = self.classes.Singular
Keyword = self.classes.Keyword
q1 = self.session.query(User).filter(
User.common_singular.has(Keyword.keyword == "brown")
)
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users "
"WHERE EXISTS (SELECT 1 "
"FROM singular "
"WHERE singular.id = users.singular_id AND (EXISTS (SELECT 1 "
"FROM keywords "
"WHERE singular.id = keywords.singular_id AND "
"keywords.keyword = :keyword_1)))",
checkparams={"keyword_1": "brown"},
)
q2 = self.session.query(User).filter(
User.singular.has(
Singular.keywords.any(Keyword.keyword == "brown")
)
)
self._equivalent(q1, q2)
def test_filter_has_scalar_raises(self):
User = self.classes.User
assert_raises_message(
exc.ArgumentError,
r"Can't apply keyword arguments to column-targeted",
User.singular_keyword.has,
keyword="brown",
)
def test_filter_eq_chained_has_to_any(self):
User = self.classes.User
Keyword = self.classes.Keyword
Singular = self.classes.Singular
q1 = self.session.query(User).filter(User.singular_keyword == "brown")
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users "
"WHERE EXISTS (SELECT 1 "
"FROM singular "
"WHERE singular.id = users.singular_id AND (EXISTS (SELECT 1 "
"FROM keywords "
"WHERE singular.id = keywords.singular_id "
"AND keywords.keyword = :keyword_1)))",
checkparams={"keyword_1": "brown"},
)
q2 = self.session.query(User).filter(
User.singular.has(
Singular.keywords.any(Keyword.keyword == "brown")
)
)
self._equivalent(q1, q2)
def test_filter_contains_chained_any_to_has(self):
User = self.classes.User
Keyword = self.classes.Keyword
UserKeyword = self.classes.UserKeyword
q1 = self.session.query(User).filter(
User.common_keyword_name.contains("brown")
)
self.assert_compile(
q1,
"SELECT users.id AS users_id, users.name AS users_name, "
"users.singular_id AS users_singular_id "
"FROM users "
"WHERE EXISTS (SELECT 1 "
"FROM userkeywords "
"WHERE users.id = userkeywords.user_id AND (EXISTS (SELECT 1 "
"FROM keywords "
"WHERE keywords.id = userkeywords.keyword_id AND "
"keywords.keyword = :keyword_1)))",
checkparams={"keyword_1": "brown"},
)
q2 = self.session.query(User).filter(
User.user_keywords.any(
UserKeyword.keyword.has(Keyword.keyword == "brown")
)
)
self._equivalent(q1, q2)
def test_filter_contains_chained_any_to_has_to_eq(self):
User = self.classes.User
Keyword = self.classes.Keyword
UserKeyword = self.classes.UserKeyword
Singular = self.classes.Singular
singular = self.session.query(Singular).order_by(Singular.id).first()
q1 = self.session.query(Keyword).filter(
Keyword.singulars.contains(singular)
)
self.assert_compile(
q1,
"SELECT keywords.id AS keywords_id, "
"keywords.keyword AS keywords_keyword, "
"keywords.singular_id AS keywords_singular_id "
"FROM keywords "
"WHERE EXISTS (SELECT 1 "
"FROM userkeywords "
"WHERE keywords.id = userkeywords.keyword_id AND "
"(EXISTS (SELECT 1 "
"FROM users "
"WHERE users.id = userkeywords.user_id AND "
":param_1 = users.singular_id)))",
checkparams={"param_1": singular.id},
)
q2 = self.session.query(Keyword).filter(
Keyword.user_keywords.any(
UserKeyword.user.has(User.singular == singular)
)
)
self._equivalent(q1, q2)
def test_has_criterion_nul(self):
# but we don't allow that with any criterion...
User = self.classes.User
self.classes.Singular
assert_raises_message(
exc.ArgumentError,
r"Non-empty has\(\) not allowed",
User.singular_value.has,
User.singular_value == "singular4",
)
def test_has_kwargs_nul(self):
# ... or kwargs
User = self.classes.User
self.classes.Singular
assert_raises_message(
exc.ArgumentError,
r"Can't apply keyword arguments to column-targeted",
User.singular_value.has,
singular_value="singular4",
)
def test_filter_scalar_object_contains_fails_nul_nul(self):
Keyword = self.classes.Keyword
assert_raises(
exc.InvalidRequestError, lambda: Keyword.user.contains(self.u)
)
def test_filter_scalar_object_any_fails_nul_nul(self):
Keyword = self.classes.Keyword
assert_raises(
exc.InvalidRequestError, lambda: Keyword.user.any(name="user2")
)
def test_filter_scalar_column_like(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(User.singular_value.like("foo")),
self.session.query(User).filter(
User.singular.has(Singular.value.like("foo"))
),
)
def test_filter_scalar_column_contains(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(
User.singular_value.contains("foo")
),
self.session.query(User).filter(
User.singular.has(Singular.value.contains("foo"))
),
)
def test_filter_scalar_column_eq(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(User.singular_value == "foo"),
self.session.query(User).filter(
User.singular.has(Singular.value == "foo")
),
)
def test_filter_scalar_column_ne(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(User.singular_value != "foo"),
self.session.query(User).filter(
User.singular.has(Singular.value != "foo")
),
)
def test_filter_scalar_column_eq_nul(self):
User = self.classes.User
Singular = self.classes.Singular
self._equivalent(
self.session.query(User).filter(User.singular_value == None),
self.session.query(User).filter(
or_(
User.singular.has(Singular.value == None),
User.singular == None,
)
),
)
def test_filter_collection_has_fails_ul_nul(self):
User = self.classes.User
assert_raises(
exc.InvalidRequestError, lambda: User.keywords.has(keyword="quick")
)
def test_filter_collection_eq_fails_ul_nul(self):
User = self.classes.User
assert_raises(
exc.InvalidRequestError, lambda: User.keywords == self.kw
)
def test_filter_collection_ne_fails_ul_nul(self):
User = self.classes.User
assert_raises(
exc.InvalidRequestError, lambda: User.keywords != self.kw
)
def test_join_separate_attr(self):
User = self.classes.User
self.assert_compile(
self.session.query(User).join(
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 = "
"userkeywords.user_id JOIN keywords ON keywords.id = "
"userkeywords.keyword_id",
)
def test_join_single_attr(self):
User = self.classes.User
self.assert_compile(
self.session.query(User).join(*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 = "
"userkeywords.user_id JOIN keywords ON keywords.id = "
"userkeywords.keyword_id",
)
class DictOfTupleUpdateTest(fixtures.TestBase):
def setup(self):
class B(object):
def __init__(self, key, elem):
self.key = key
self.elem = elem
class A(object):
elements = association_proxy("orig", "elem", creator=B)
m = MetaData()
a = Table("a", m, Column("id", Integer, primary_key=True))
b = Table(
"b",
m,
Column("id", Integer, primary_key=True),
Column("aid", Integer, ForeignKey("a.id")),
Column("elem", String),
)
mapper(
A,
a,
properties={
"orig": relationship(
B, collection_class=attribute_mapped_collection("key")
)
},
)
mapper(B, b)
self.A = A
self.B = B
def test_update_one_elem_dict(self):
a1 = self.A()
a1.elements.update({("B", 3): "elem2"})
eq_(a1.elements, {("B", 3): "elem2"})
def test_update_multi_elem_dict(self):
a1 = self.A()
a1.elements.update({("B", 3): "elem2", ("C", 4): "elem3"})
eq_(a1.elements, {("B", 3): "elem2", ("C", 4): "elem3"})
def test_update_one_elem_list(self):
a1 = self.A()
a1.elements.update([(("B", 3), "elem2")])
eq_(a1.elements, {("B", 3): "elem2"})
def test_update_multi_elem_list(self):
a1 = self.A()
a1.elements.update([(("B", 3), "elem2"), (("C", 4), "elem3")])
eq_(a1.elements, {("B", 3): "elem2", ("C", 4): "elem3"})
def test_update_one_elem_varg(self):
a1 = self.A()
assert_raises_message(
ValueError,
"dictionary update sequence requires " "2-element tuples",
a1.elements.update,
(("B", 3), "elem2"),
)
def test_update_multi_elem_varg(self):
a1 = self.A()
assert_raises_message(
TypeError,
"update expected at most 1 arguments, got 2",
a1.elements.update,
(("B", 3), "elem2"),
(("C", 4), "elem3"),
)
class AttributeAccessTest(fixtures.TestBase):
def teardown(self):
clear_mappers()
def test_resolve_aliased_class(self):
Base = declarative_base()
class A(Base):
__tablename__ = "a"
id = Column(Integer, primary_key=True)
value = Column(String)
class B(Base):
__tablename__ = "b"
id = Column(Integer, primary_key=True)
a_id = Column(Integer, ForeignKey(A.id))
a = relationship(A)
a_value = association_proxy("a", "value")
spec = aliased(B).a_value
is_(spec.owning_class, B)
spec = B.a_value
is_(spec.owning_class, B)
def test_resolved_w_subclass(self):
# test for issue #4185, as well as several below
Base = declarative_base()
class Mixin(object):
@declared_attr
def children(cls):
return association_proxy("_children", "value")
# 1. build parent, Mixin.children gets invoked, we add
# Parent.children
class Parent(Mixin, Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
# 2. declarative builds up SubParent, scans through all attributes
# over all classes. Hits Mixin, hits "children", accesses "children"
# in terms of the class, e.g. SubParent.children. SubParent isn't
# mapped yet. association proxy then sets up "owning_class"
# as NoneType.
class SubParent(Parent):
__tablename__ = "subparent"
id = Column(Integer, ForeignKey(Parent.id), primary_key=True)
configure_mappers()
# 3. which would break here.
p1 = Parent()
eq_(p1.children, [])
def test_resolved_to_correct_class_one(self):
Base = declarative_base()
class Parent(Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
children = association_proxy("_children", "value")
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
class SubParent(Parent):
__tablename__ = "subparent"
id = Column(Integer, ForeignKey(Parent.id), primary_key=True)
is_(SubParent.children.owning_class, SubParent)
is_(Parent.children.owning_class, Parent)
def test_resolved_to_correct_class_two(self):
Base = declarative_base()
class Parent(Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
class SubParent(Parent):
__tablename__ = "subparent"
id = Column(Integer, ForeignKey(Parent.id), primary_key=True)
children = association_proxy("_children", "value")
is_(SubParent.children.owning_class, SubParent)
def test_resolved_to_correct_class_three(self):
Base = declarative_base()
class Parent(Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
class SubParent(Parent):
__tablename__ = "subparent"
id = Column(Integer, ForeignKey(Parent.id), primary_key=True)
children = association_proxy("_children", "value")
class SubSubParent(SubParent):
__tablename__ = "subsubparent"
id = Column(Integer, ForeignKey(SubParent.id), primary_key=True)
is_(SubParent.children.owning_class, SubParent)
is_(SubSubParent.children.owning_class, SubSubParent)
def test_resolved_to_correct_class_four(self):
Base = declarative_base()
class Parent(Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
children = association_proxy(
"_children", "value", creator=lambda value: Child(value=value)
)
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
class SubParent(Parent):
__tablename__ = "subparent"
id = Column(Integer, ForeignKey(Parent.id), primary_key=True)
sp = SubParent()
sp.children = "c"
is_(SubParent.children.owning_class, SubParent)
is_(Parent.children.owning_class, Parent)
def test_resolved_to_correct_class_five(self):
Base = declarative_base()
class Mixin(object):
children = association_proxy("_children", "value")
class Parent(Mixin, Base):
__tablename__ = "parent"
id = Column(Integer, primary_key=True)
_children = relationship("Child")
class Child(Base):
__tablename__ = "child"
parent_id = Column(
Integer, ForeignKey(Parent.id), primary_key=True
)
value = Column(String)
# this triggers the owning routine, doesn't fail
Mixin.children
p1 = Parent()
c1 = Child(value="c1")
p1._children.append(c1)
is_(Parent.children.owning_class, Parent)
eq_(p1.children, ["c1"])
def _test_never_assign_nonetype(self):
foo = association_proxy("x", "y")
foo._calc_owner(None, None)
is_(foo.owning_class, None)
class Bat(object):
foo = association_proxy("x", "y")
Bat.foo
is_(Bat.foo.owning_class, None)
b1 = Bat()
assert_raises_message(
exc.InvalidRequestError,
"This association proxy has no mapped owning class; "
"can't locate a mapped property",
getattr,
b1,
"foo",
)
is_(Bat.foo.owning_class, None)
# after all that, we can map it
mapper(
Bat,
Table("bat", MetaData(), Column("x", Integer, primary_key=True)),
)
# answer is correct
is_(Bat.foo.owning_class, Bat)
class ScalarRemoveTest(object):
useobject = None
cascade_scalar_deletes = None
uselist = None
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
class A(Base):
__tablename__ = "test_a"
id = Column(Integer, primary_key=True)
ab = relationship("AB", backref="a", uselist=cls.uselist)
b = association_proxy(
"ab",
"b",
creator=lambda b: AB(b=b),
cascade_scalar_deletes=cls.cascade_scalar_deletes,
)
if cls.useobject:
class B(Base):
__tablename__ = "test_b"
id = Column(Integer, primary_key=True)
ab = relationship("AB", backref="b")
class AB(Base):
__tablename__ = "test_ab"
a_id = Column(Integer, ForeignKey(A.id), primary_key=True)
b_id = Column(Integer, ForeignKey(B.id), primary_key=True)
else:
class AB(Base):
__tablename__ = "test_ab"
b = Column(Integer)
a_id = Column(Integer, ForeignKey(A.id), primary_key=True)
def test_set_nonnone_to_none(self):
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
a1 = A()
b1 = B() if self.useobject else 5
if self.uselist:
a1.b.append(b1)
else:
a1.b = b1
if self.uselist:
assert isinstance(a1.ab[0], AB)
else:
assert isinstance(a1.ab, AB)
if self.uselist:
a1.b.remove(b1)
else:
a1.b = None
if self.uselist:
eq_(a1.ab, [])
else:
if self.cascade_scalar_deletes:
assert a1.ab is None
else:
assert isinstance(a1.ab, AB)
assert a1.ab.b is None
def test_set_none_to_none(self):
if self.uselist:
return
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
a1 = A()
a1.b = None
assert a1.ab is None
def test_del_already_nonpresent(self):
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
a1 = A()
if self.uselist:
del a1.b
eq_(a1.ab, [])
else:
def go():
del a1.b
assert_raises_message(
AttributeError, "A.ab object does not have a value", go
)
def test_del(self):
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
b1 = B() if self.useobject else 5
a1 = A()
if self.uselist:
a1.b.append(b1)
else:
a1.b = b1
if self.uselist:
assert isinstance(a1.ab[0], AB)
else:
assert isinstance(a1.ab, AB)
del a1.b
if self.uselist:
eq_(a1.ab, [])
else:
assert a1.ab is None
def test_del_no_proxy(self):
if not self.uselist:
return
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
b1 = B() if self.useobject else 5
a1 = A()
a1.b.append(b1)
del a1.ab
# this is what it does for now, so maintain that w/ assoc proxy
eq_(a1.ab, [])
def test_del_already_nonpresent_no_proxy(self):
if not self.uselist:
return
if self.useobject:
A, AB, B = self.classes("A", "AB", "B")
else:
A, AB = self.classes("A", "AB")
a1 = A()
del a1.ab
# this is what it does for now, so maintain that w/ assoc proxy
eq_(a1.ab, [])
class ScalarRemoveListObjectCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = True
cascade_scalar_deletes = True
uselist = True
class ScalarRemoveScalarObjectCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = True
cascade_scalar_deletes = True
uselist = False
class ScalarRemoveListScalarCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = False
cascade_scalar_deletes = True
uselist = True
class ScalarRemoveScalarScalarCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = False
cascade_scalar_deletes = True
uselist = False
class ScalarRemoveListObjectNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = True
cascade_scalar_deletes = False
uselist = True
class ScalarRemoveScalarObjectNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = True
cascade_scalar_deletes = False
uselist = False
class ScalarRemoveListScalarNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = False
cascade_scalar_deletes = False
uselist = True
class ScalarRemoveScalarScalarNoCascade(
ScalarRemoveTest, fixtures.DeclarativeMappedTest
):
useobject = False
cascade_scalar_deletes = False
uselist = False
class InfoTest(fixtures.TestBase):
def test_constructor(self):
assoc = association_proxy("a", "b", info={"some_assoc": "some_value"})
eq_(assoc.info, {"some_assoc": "some_value"})
def test_empty(self):
assoc = association_proxy("a", "b")
eq_(assoc.info, {})
def test_via_cls(self):
class Foob(object):
assoc = association_proxy("a", "b")
eq_(Foob.assoc.info, {})
Foob.assoc.info["foo"] = "bar"
eq_(Foob.assoc.info, {"foo": "bar"})
class OnlyRelationshipTest(fixtures.DeclarativeMappedTest):
run_define_tables = None
run_create_tables = None
run_inserts = None
run_deletes = None
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
class Foo(Base):
__tablename__ = "foo"
id = Column(Integer, primary_key=True)
foo = Column(String) # assume some composite datatype
bar = association_proxy("foo", "attr")
def test_setattr(self):
Foo = self.classes.Foo
f1 = Foo()
assert_raises_message(
NotImplementedError,
"association proxy to a non-relationship "
"intermediary is not supported",
setattr,
f1,
"bar",
"asdf",
)
def test_getattr(self):
Foo = self.classes.Foo
f1 = Foo()
assert_raises_message(
NotImplementedError,
"association proxy to a non-relationship "
"intermediary is not supported",
getattr,
f1,
"bar",
)
def test_get_class_attr(self):
Foo = self.classes.Foo
assert_raises_message(
NotImplementedError,
"association proxy to a non-relationship "
"intermediary is not supported",
getattr,
Foo,
"bar",
)
class MultiOwnerTest(
fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
):
__dialect__ = "default"
run_define_tables = "each"
run_create_tables = None
run_inserts = None
run_deletes = None
run_setup_classes = "each"
run_setup_mappers = "each"
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
class A(Base):
__tablename__ = "a"
id = Column(Integer, primary_key=True)
type = Column(String(5), nullable=False)
d_values = association_proxy("ds", "value")
__mapper_args__ = {"polymorphic_on": type}
class B(A):
__tablename__ = "b"
id = Column(ForeignKey("a.id"), primary_key=True)
c1_id = Column(ForeignKey("c1.id"))
ds = relationship("D", primaryjoin="D.b_id == B.id")
__mapper_args__ = {"polymorphic_identity": "b"}
class C(A):
__tablename__ = "c"
id = Column(ForeignKey("a.id"), primary_key=True)
ds = relationship(
"D", primaryjoin="D.c_id == C.id", back_populates="c"
)
__mapper_args__ = {"polymorphic_identity": "c"}
class C1(C):
__tablename__ = "c1"
id = Column(ForeignKey("c.id"), primary_key=True)
csub_only_data = relationship("B") # uselist=True relationship
ds = relationship(
"D", primaryjoin="D.c1_id == C1.id", back_populates="c"
)
__mapper_args__ = {"polymorphic_identity": "c1"}
class C2(C):
__tablename__ = "c2"
id = Column(ForeignKey("c.id"), primary_key=True)
csub_only_data = Column(String(50)) # scalar Column
ds = relationship(
"D", primaryjoin="D.c2_id == C2.id", back_populates="c"
)
__mapper_args__ = {"polymorphic_identity": "c2"}
class D(Base):
__tablename__ = "d"
id = Column(Integer, primary_key=True)
value = Column(String(50))
b_id = Column(ForeignKey("b.id"))
c_id = Column(ForeignKey("c.id"))
c1_id = Column(ForeignKey("c1.id"))
c2_id = Column(ForeignKey("c2.id"))
c = relationship("C", primaryjoin="D.c_id == C.id")
c_data = association_proxy("c", "csub_only_data")
def _assert_raises_ambiguous(self, fn, *arg, **kw):
assert_raises_message(
AttributeError,
"Association proxy D.c refers to an attribute 'csub_only_data'",
fn,
*arg,
**kw
)
def test_column_collection_expressions(self):
B, C, C2 = self.classes("B", "C", "C2")
self.assert_compile(
B.d_values.contains("b1"),
"EXISTS (SELECT 1 FROM d, b WHERE d.b_id = b.id "
"AND (d.value LIKE '%' || :value_1 || '%'))",
)
self.assert_compile(
C2.d_values.contains("c2"),
"EXISTS (SELECT 1 FROM d, c2 WHERE d.c2_id = c2.id "
"AND (d.value LIKE '%' || :value_1 || '%'))",
)
self.assert_compile(
C.d_values.contains("c1"),
"EXISTS (SELECT 1 FROM d, c WHERE d.c_id = c.id "
"AND (d.value LIKE '%' || :value_1 || '%'))",
)
def test_subclass_only_owner_none(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D()
self._assert_raises_ambiguous(getattr, d1, "c_data")
def test_subclass_only_owner_assign(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C2())
d1.c_data = "some c2"
eq_(d1.c_data, "some c2")
def test_subclass_only_owner_get(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C2(csub_only_data="some c2"))
eq_(d1.c_data, "some c2")
def test_subclass_only_owner_none_raise(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D()
self._assert_raises_ambiguous(getattr, d1, "c_data")
def test_subclass_only_owner_delete(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C2(csub_only_data="some c2"))
del d1.c_data
self._assert_raises_ambiguous(getattr, d1, "c_data")
def test_subclass_only_owner_assign_raises(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C())
self._assert_raises_ambiguous(setattr, d1, "c_data", "some c1")
def test_subclass_only_owner_get_raises(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C())
self._assert_raises_ambiguous(getattr, d1, "c_data")
def test_subclass_only_owner_delete_raises(self):
D, C, C2 = self.classes("D", "C", "C2")
d1 = D(c=C2(csub_only_data="some c2"))
eq_(d1.c_data, "some c2")
# now switch
d1.c = C()
self._assert_raises_ambiguous(delattr, d1, "c_data")
def test_subclasses_conflicting_types(self):
B, D, C, C1, C2 = self.classes("B", "D", "C", "C1", "C2")
bs = [B(), B()]
d1 = D(c=C1(csub_only_data=bs))
d2 = D(c=C2(csub_only_data="some c2"))
association_proxy_object = inspect(D).all_orm_descriptors["c_data"]
inst1 = association_proxy_object.for_class(D, d1)
inst2 = association_proxy_object.for_class(D, d2)
eq_(inst1._target_is_object, True)
eq_(inst2._target_is_object, False)
# both instances are cached
inst0 = association_proxy_object.for_class(D)
eq_(inst0._lookup_cache, {C1: inst1, C2: inst2})
# cache works
is_(association_proxy_object.for_class(D, d1), inst1)
is_(association_proxy_object.for_class(D, d2), inst2)
def test_col_expressions_not_available(self):
D, = self.classes("D")
self._assert_raises_ambiguous(lambda: D.c_data == 5)
def test_rel_expressions_not_available(self):
B, D, = self.classes("B", "D")
self._assert_raises_ambiguous(lambda: D.c_data.any(B.id == 5))
class ScopeBehaviorTest(fixtures.DeclarativeMappedTest):
# test some GC scenarios, including issue #4268
@classmethod
def setup_classes(cls):
Base = cls.DeclarativeBasic
class A(Base):
__tablename__ = "a"
id = Column(Integer, primary_key=True)
data = Column(String(50))
bs = relationship("B")
b_dyn = relationship("B", lazy="dynamic")
b_data = association_proxy("bs", "data")
b_dynamic_data = association_proxy("bs", "data")
class B(Base):
__tablename__ = "b"
id = Column(Integer, primary_key=True)
aid = Column(ForeignKey("a.id"))
data = Column(String(50))
@classmethod
def insert_data(cls):
A, B = cls.classes("A", "B")
s = Session(testing.db)
s.add_all(
[
A(id=1, bs=[B(data="b1"), B(data="b2")]),
A(id=2, bs=[B(data="b3"), B(data="b4")]),
]
)
s.commit()
s.close()
def test_plain_collection_gc(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.bs # noqa
del a1
gc_collect()
assert (A, (1,), None) not in s.identity_map
@testing.fails("dynamic relationship strong references parent")
def test_dynamic_collection_gc(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_dyn # noqa
del a1
gc_collect()
# also fails, AppenderQuery holds onto parent
assert (A, (1,), None) not in s.identity_map
@testing.fails("association proxy strong references parent")
def test_associated_collection_gc(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_data # noqa
del a1
gc_collect()
assert (A, (1,), None) not in s.identity_map
@testing.fails("association proxy strong references parent")
def test_associated_dynamic_gc(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_dynamic_data # noqa
del a1
gc_collect()
assert (A, (1,), None) not in s.identity_map
def test_plain_collection_iterate(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.bs
del a1
gc_collect()
assert len(a1bs) == 2
def test_dynamic_collection_iterate(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_dyn # noqa
del a1
gc_collect()
assert len(list(a1bs)) == 2
def test_associated_collection_iterate(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_data
del a1
gc_collect()
assert len(a1bs) == 2
def test_associated_dynamic_iterate(self):
A, B = self.classes("A", "B")
s = Session(testing.db)
a1 = s.query(A).filter_by(id=1).one()
a1bs = a1.b_dynamic_data
del a1
gc_collect()
assert len(a1bs) == 2
Reference in New Issue View Git Blame Copy Permalink