from orm import _base
from testlib.testing import eq_
+
class AssociationTest(_base.MappedTest):
run_setup_classes = 'once'
run_setup_mappers = 'once'
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
- sess.save(item1)
- sess.save(item2)
+ sess.add_all((item1, item2))
sess.flush()
saved = repr([item1, item2])
sess.clear()
item1 = Item('item1')
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
- sess.save(item1)
+ sess.add(item1)
sess.flush()
red_keyword = item1.keywords[1].keyword
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
- sess.save(item1)
- sess.save(item2)
+ sess.add_all((item1, item2))
sess.flush()
red_keyword = item1.keywords[1].keyword
item1.keywords.append(KeywordAssociation(Keyword('blue'), 'blue_assoc'))
item1.keywords.append(KeywordAssociation(Keyword('red'), 'red_assoc'))
item2.keywords.append(KeywordAssociation(Keyword('green'), 'green_assoc'))
- sess.save(item1)
- sess.save(item2)
+ sess.add_all((item1, item2))
sess.flush()
eq_(self.tables.item_keywords.count().scalar(), 3)
-"""eager loading unittests derived from mailing list-reported problems and trac tickets."""
+"""Exercises for eager loading.
+Derived from mailing list-reported problems and trac tickets.
+
+"""
import testenv; testenv.configure_for_tests()
import random, datetime
from sqlalchemy import *
# an owner
o=Owner()
- s.save(o)
+ s.add(o)
# owner a has 3 tests, one of which he has specified options for
c=Category()
c.name='Some Category'
- s.save(c)
+ s.add(c)
for i in range(3):
t=Test()
t.owner=o
t.category=c
- s.save(t)
+ s.add(t)
if i==1:
op=Option()
op.someoption=True
p = Middle('test1')
p.left.append(Left('tag1'))
p.right.append(Right('tag2'))
- session.save(p)
+ session.add(p)
session.flush()
session.clear()
obj = session.query(Left).filter_by(tag='tag1').one()
for x in range(5):
d=Data()
d.a=x
- s.save(d)
+ s.add(d)
data.append(d)
for x in range(10):
stat=Stat()
stat.data = data[rid]
stat.somedata=somedata
- s.save(stat)
+ s.add(stat)
s.flush()
d2.employees.append(Employee(name=e))
sess = create_session()
- sess.save(d1)
- sess.save(d2)
+ sess.add_all((d1, d2))
sess.flush()
q = sess.query(Department)
d.comments = [Comment('uid1', 'comment')]
d2 = DerivedII('uid2', 'xx', 'z')
d2.comments = [Comment('uid2', 'comment')]
- sess.save(d)
- sess.save(d2)
+ sess.add_all((d, d2))
sess.flush()
sess.clear()
# this eager load sets up an AliasedClauses for the "comment" relationship,
d = Design()
sess = create_session()
- sess.save(d)
+ sess.add(d)
sess.flush()
sess.clear()
x = sess.query(Design).get(1)
ent21 = Entry(name='ent21', account=acc2, transaction=tx1)
ent22 = Entry(name='ent22', account=acc2, transaction=tx2)
- session.save(acc1)
+ session.add(acc1)
session.flush()
session.clear()
u = User(name='jack', orders=[
Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
[Order(description="order 3"), Order(description="order 4")])
o5 = Order(description="order 5")
- sess.save(o5)
+ sess.add(o5)
try:
sess.flush()
assert False
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.delete(u)
u = User(name='jack',
addresses=[Address(email_address="address1"),
Address(email_address="address2")])
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
assert addresses.count().scalar() == 2
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
o = u.orders[0]
assert o in sess
u2 = User(name='newuser', orders=[o])
- sess.save(u2)
+ sess.add(u2)
sess.flush()
sess.clear()
assert users.count().scalar() == 1
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
assert users.count().scalar() == 1
assert orders.count().scalar() == 2
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
assert users.count().scalar() == 1
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
- sess.save(u)
+ sess.add(u)
sess.flush()
assert users.count().scalar() == 1
assert orders.count().scalar() == 2
def test_pending_expunge(self):
sess = create_session()
someuser = User(name='someuser')
- sess.save(someuser)
+ sess.add(someuser)
sess.flush()
someuser.pref = p1 = Pref(data='somepref')
assert p1 in sess
def test_cascade_delete(self):
sess = create_session()
x = T1(data='t1a', t2=T2(data='t2a', t3=T3(data='t3a')))
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.delete(x)
x1 = T1(data='t1', )
x2 = T2(data='t2')
x3 = T3(data='t3')
- sess.save(x1)
- sess.save(x2)
- sess.save(x3)
+ sess.add_all((x1, x2, x3))
sess.flush()
sess.delete(x1)
sess = create_session()
x1 = T1(data='t1', t2=T2(data='t2'))
x3 = T3(data='t3')
- sess.save(x1)
- sess.save(x3)
+ sess.add_all((x1, x3))
sess.flush()
sess.delete(x1)
def test_preserves_orphans_onelevel(self):
sess = create_session()
x2 = T1(data='t1b', t2=T2(data='t2b', t3=T3(data='t3b')))
- sess.save(x2)
+ sess.add(x2)
sess.flush()
x2.t2 = None
def test_preserves_orphans_onelevel_postremove(self):
sess = create_session()
x2 = T1(data='t1b', t2=T2(data='t2b', t3=T3(data='t3b')))
- sess.save(x2)
+ sess.add(x2)
sess.flush()
sess.delete(x2)
def test_preserves_orphans_twolevel(self):
sess = create_session()
x = T1(data='t1a', t2=T2(data='t2a', t3=T3(data='t3a')))
- sess.save(x)
+ sess.add(x)
sess.flush()
x.t2.t3 = None
def test_cascade_delete(self):
sess = create_session()
x = T1(data='t1a', t2=T2(data='t2a', t3=T3(data='t3a')))
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.delete(x)
def test_deletes_orphans_onelevel(self):
sess = create_session()
x2 = T1(data='t1b', t2=T2(data='t2b', t3=T3(data='t3b')))
- sess.save(x2)
+ sess.add(x2)
sess.flush()
x2.t2 = None
def test_deletes_orphans_twolevel(self):
sess = create_session()
x = T1(data='t1a', t2=T2(data='t2a', t3=T3(data='t3a')))
- sess.save(x)
+ sess.add(x)
sess.flush()
x.t2.t3 = None
def test_finds_orphans_twolevel(self):
sess = create_session()
x = T1(data='t1a', t2=T2(data='t2a', t3=T3(data='t3a')))
- sess.save(x)
+ sess.add(x)
sess.flush()
x.t2.t3 = None
sess = create_session()
b1 = B(data='b1')
a1 = A(data='a1', bs=[b1])
- sess.save(a1)
+ sess.add(a1)
sess.flush()
a1.bs.remove(b1)
sess = create_session()
b1 = B(data='b1', cs=[C(data='c1')])
a1 = A(data='a1', bs=[b1])
- sess.save(a1)
+ sess.add(a1)
sess.flush()
a1.bs.remove(b1)
sess = create_session()
a1 = A(data='a1', bs=[B(data='b1')])
- sess.save(a1)
+ sess.add(a1)
sess.flush()
sess.delete(a1)
))
s = create_session()
a = Address()
- s.save(a)
+ s.add(a)
try:
s.flush()
except orm_exc.FlushError, e:
s = create_session()
u = User()
- s.save(u)
+ s.add(u)
s.flush()
a = Address()
))
s = create_session()
u = User(name='u1', addresses=[Address(email_address='ad1')])
- s.save(u)
+ s.add(u)
a1 = u.addresses[0]
u.addresses.remove(a1)
assert a1 in s
s = create_session()
order = Order(name="order1")
- s.save(order)
+ s.add(order)
attr = Attribute(name="attr1")
item = Item(name="item1", attributes=[attr])
a = Account(balance=0)
sr = SalesRep(name="John")
- [s.save(x) for x in [a,sr]]
+ s.add_all((a, sr))
s.flush()
c = Customer(name="Jane")
session = create_session()
h1 = Home(description='home1', address=Address(street='address1'))
b1 = Business(description='business1', address=Address(street='address2'))
- [session.save(x) for x in [h1,b1]]
+ session.add_all((h1,b1))
session.flush()
session.clear()
session = create_session()
a1 = Address()
- session.save(a1)
+ session.add(a1)
try:
session.flush()
assert False
a1 = A(name='a1', bs=[B(name='b1'), B(name='b2'), B(name='b3')])
sess = create_session()
- sess.save(a1)
+ sess.add(a1)
sess.flush()
sess.clear()
p.children['foo'] = Child('foo', 'value')
p.children['bar'] = Child('bar', 'value')
session = create_session()
- session.save(p)
+ session.add(p)
session.flush()
pid = p.id
session.clear()
p.children[('foo', '2')] = Child('foo', '2', 'value 2')
session = create_session()
- session.save(p)
+ session.add(p)
session.flush()
pid = p.id
session.clear()
f.bars.add(Bar())
f.bars.add(Bar())
sess = create_session()
- sess.save(f)
+ sess.add(f)
sess.flush()
sess.clear()
f = sess.query(Foo).get(f.col1)
f.bars.set(Bar())
f.bars.set(Bar())
sess = create_session()
- sess.save(f)
+ sess.add(f)
sess.flush()
sess.clear()
f = sess.query(Foo).get(f.col1)
col.append_with_event(Bar('a'))
col.append_with_event(Bar('b'))
sess = create_session()
- sess.save(f)
+ sess.add(f)
sess.flush()
sess.clear()
f = sess.query(Foo).get(f.col1)
assert control == list(p1.children)
sess = create_session()
- sess.save(p1)
+ sess.add(p1)
sess.flush()
sess.clear()
-import testenv; testenv.configure_for_tests()
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from testlib import *
-from testlib.tables import *
-from orm import _base
-
-"""
-Tests cyclical mapper relationships.
+"""Tests cyclical mapper relationships.
We might want to try an automated generate of much of this, all combos of
T1<->T2, with o2m or m2o between them, and a third T3 with o2m/m2o to one/both
T1/T2.
+
"""
+import testenv; testenv.configure_for_tests()
+from testlib import testing
+from testlib.sa import Table, Column, Integer, String, ForeignKey
+from testlib.sa.orm import mapper, relation, backref, create_session
+from testlib.testing import eq_
+from orm import _base
+
+class SelfReferentialTest(_base.MappedTest):
+ """A self-referential mapper with an additional list of child objects."""
+
+ def define_tables(self, metadata):
+ Table('t1', metadata,
+ Column('c1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('parent_c1', Integer, ForeignKey('t1.c1')),
+ Column('data', String(20)))
+ Table('t2', metadata,
+ Column('c1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('c1id', Integer, ForeignKey('t1.c1')),
+ Column('data', String(20)))
+
+ def setup_classes(self):
+ class C1(_base.BasicEntity):
+ def __init__(self, data=None):
+ self.data = data
-class Tester(object):
- def __init__(self, data=None):
- self.data = data
- print repr(self) + " (%d)" % (id(self))
- def __repr__(self):
- return "%s(%s)" % (self.__class__.__name__, repr(self.data))
-
-class SelfReferentialTest(_base.ORMTest):
- """tests a self-referential mapper, with an additional list of child objects."""
- def setUpAll(self):
- global t1, t2, metadata
- metadata = MetaData(testing.db)
- t1 = Table('t1', metadata,
- Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True),
- Column('parent_c1', Integer, ForeignKey('t1.c1')),
- Column('data', String(20))
- )
- t2 = Table('t2', metadata,
- Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
- Column('c1id', Integer, ForeignKey('t1.c1')),
- Column('data', String(20))
- )
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def setUp(self):
- clear_mappers()
+ class C2(_base.BasicEntity):
+ def __init__(self, data=None):
+ self.data = data
+ @testing.resolve_artifact_names
def testsingle(self):
- class C1(Tester):
- pass
- m1 = mapper(C1, t1, properties = {
+ mapper(C1, t1, properties = {
'c1s':relation(C1, cascade="all"),
- 'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, remote_side=t1.c.c1, lazy=True, uselist=False)
- })
+ 'parent':relation(C1,
+ primaryjoin=t1.c.parent_c1 == t1.c.c1,
+ remote_side=t1.c.c1,
+ lazy=True,
+ uselist=False)})
a = C1('head c1')
a.c1s.append(C1('another c1'))
+
sess = create_session( )
- sess.save(a)
+ sess.add(a)
sess.flush()
sess.delete(a)
sess.flush()
+ @testing.resolve_artifact_names
def testmanytooneonly(self):
- """test that the circular dependency sort can assemble a many-to-one dependency processor
- when only the object on the "many" side is actually in the list of modified objects.
- this requires that the circular sort add the other side of the relation into the UOWTransaction
- so that the dependency operation can be tacked onto it.
+ """
+
+ test that the circular dependency sort can assemble a many-to-one
+ dependency processor when only the object on the "many" side is
+ actually in the list of modified objects. this requires that the
+ circular sort add the other side of the relation into the
+ UOWTransaction so that the dependency operation can be tacked onto it.
+
+ This also affects inheritance relationships since they rely upon
+ circular sort as well.
- This also affects inheritance relationships since they rely upon circular sort as well.
"""
- class C1(Tester):
- pass
mapper(C1, t1, properties={
- 'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, remote_side=t1.c.c1)
- })
- sess = create_session()
+ 'parent':relation(C1,
+ primaryjoin=t1.c.parent_c1 == t1.c.c1,
+ remote_side=t1.c.c1)})
+
c1 = C1()
- sess.save(c1)
+
+ sess = create_session()
+ sess.add(c1)
sess.flush()
sess.clear()
c1 = sess.query(C1).get(c1.c1)
c2 = C1()
c2.parent = c1
- sess.save(c2)
+ sess.add(c2)
sess.flush()
assert c2.parent_c1==c1.c1
+ @testing.resolve_artifact_names
def testcycle(self):
- class C1(Tester):
- pass
- class C2(Tester):
- pass
-
- m1 = mapper(C1, t1, properties = {
+ mapper(C1, t1, properties = {
'c1s' : relation(C1, cascade="all"),
- 'c2s' : relation(mapper(C2, t2), cascade="all, delete-orphan")
- })
+ 'c2s' : relation(mapper(C2, t2), cascade="all, delete-orphan")})
a = C1('head c1')
a.c1s.append(C1('child1'))
a.c1s[1].c2s.append(C2('child2 data1'))
a.c1s[1].c2s.append(C2('child2 data2'))
sess = create_session( )
- sess.save(a)
+ sess.add(a)
sess.flush()
sess.delete(a)
sess.flush()
-class SelfReferentialNoPKTest(TestBase, AssertsExecutionResults):
- """test self-referential relationship that joins on a column other than the primary key column"""
- def setUpAll(self):
- global table, meta
- meta = MetaData(testing.db)
- table = Table('item', meta,
+
+class SelfReferentialNoPKTest(_base.MappedTest):
+ """A self-referential relationship that joins on a column other than the primary key column"""
+
+ def define_tables(self, metadata):
+ Table('item', metadata,
Column('id', Integer, primary_key=True),
Column('uuid', String(32), unique=True, nullable=False),
- Column('parent_uuid', String(32), ForeignKey('item.uuid'), nullable=True),
- )
- meta.create_all()
- def tearDown(self):
- table.delete().execute()
- def tearDownAll(self):
- meta.drop_all()
- def testbasic(self):
- class TT(object):
+ Column('parent_uuid', String(32), ForeignKey('item.uuid'),
+ nullable=True))
+
+ def setup_classes(self):
+ class TT(_base.BasicEntity):
def __init__(self):
self.uuid = hex(id(self))
- mapper(TT, table, properties={'children':relation(TT, remote_side=[table.c.parent_uuid], backref=backref('parent', remote_side=[table.c.uuid]))})
- s = create_session()
+
+ @testing.resolve_artifact_names
+ def setup_mappers(self):
+ mapper(TT, item, properties={
+ 'children': relation(
+ TT,
+ remote_side=[item.c.parent_uuid],
+ backref=backref('parent', remote_side=[item.c.uuid]))})
+
+ @testing.resolve_artifact_names
+ def testbasic(self):
t1 = TT()
t1.children.append(TT())
t1.children.append(TT())
- s.save(t1)
+
+ s = create_session()
+ s.add(t1)
s.flush()
s.clear()
t = s.query(TT).filter_by(id=t1.id).one()
- assert t.children[0].parent_uuid == t1.uuid
+ eq_(t.children[0].parent_uuid, t1.uuid)
+
+ @testing.resolve_artifact_names
def testlazyclause(self):
- class TT(object):
- def __init__(self):
- self.uuid = hex(id(self))
- mapper(TT, table, properties={'children':relation(TT, remote_side=[table.c.parent_uuid], backref=backref('parent', remote_side=[table.c.uuid]))})
s = create_session()
t1 = TT()
t2 = TT()
t1.children.append(t2)
- s.save(t1)
+ s.add(t1)
s.flush()
s.clear()
t = s.query(TT).filter_by(id=t2.id).one()
- assert t.uuid == t2.uuid
- assert t.parent.uuid == t1.uuid
-
-class InheritTestOne(_base.ORMTest):
- def setUpAll(self):
- global parent, child1, child2, meta
- meta = MetaData(testing.db)
- parent = Table("parent", meta,
+ eq_(t.uuid, t2.uuid)
+ eq_(t.parent.uuid, t1.uuid)
+
+
+class InheritTestOne(_base.MappedTest):
+ def define_tables(self, metadata):
+ Table("parent", metadata,
Column("id", Integer, primary_key=True),
Column("parent_data", String(50)),
- Column("type", String(10))
- )
-
- child1 = Table("child1", meta,
- Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
- Column("child1_data", String(50))
- )
-
- child2 = Table("child2", meta,
- Column("id", Integer, ForeignKey("parent.id"), primary_key=True),
- Column("child1_id", Integer, ForeignKey("child1.id"), nullable=False),
- Column("child2_data", String(50))
- )
- meta.create_all()
-
- def tearDownAll(self):
- meta.drop_all()
-
- def testmanytooneonly(self):
- """test similar to SelfReferentialTest.testmanytooneonly"""
-
- class Parent(object):
+ Column("type", String(10)))
+
+ Table("child1", metadata,
+ Column("id", Integer, ForeignKey("parent.id"),
+ primary_key=True),
+ Column("child1_data", String(50)))
+
+ Table("child2", metadata,
+ Column("id", Integer, ForeignKey("parent.id"),
+ primary_key=True),
+ Column("child1_id", Integer, ForeignKey("child1.id"),
+ nullable=False),
+ Column("child2_data", String(50)))
+
+ def setup_classes(self):
+ class Parent(_base.BasicEntity):
pass
- mapper(Parent, parent)
-
class Child1(Parent):
pass
- mapper(Child1, child1, inherits=Parent)
-
class Child2(Parent):
pass
- mapper(Child2, child2, properties={
- "child1": relation(Child1,
- primaryjoin=child2.c.child1_id==child1.c.id,
- )
- },inherits=Parent)
+ @testing.resolve_artifact_names
+ def setup_mappers(self):
+ mapper(Parent, parent)
+ mapper(Child1, child1, inherits=Parent)
+ mapper(Child2, child2, inherits=Parent, properties=dict(
+ child1=relation(Child1,
+ primaryjoin=child2.c.child1_id == child1.c.id)))
+
+ @testing.resolve_artifact_names
+ def testmanytooneonly(self):
+ """test similar to SelfReferentialTest.testmanytooneonly"""
session = create_session()
c1 = Child1()
c1.child1_data = "qwerty"
- session.save(c1)
+ session.add(c1)
session.flush()
session.clear()
c2 = Child2()
c2.child1 = c1
c2.child2_data = "asdfgh"
- session.save(c2)
+ session.add(c2)
+
# the flush will fail if the UOW does not set up a many-to-one DP
- # attached to a task corresponding to c1, since "child1_id" is not nullable
+ # attached to a task corresponding to c1, since "child1_id" is not
+ # nullable
session.flush()
+
class InheritTestTwo(_base.MappedTest):
- """the fix in BiDirectionalManyToOneTest raised this issue, regarding
- the 'circular sort' containing UOWTasks that were still polymorphic, which could
- create duplicate entries in the final sort
-
+ """
+
+ The fix in BiDirectionalManyToOneTest raised this issue, regarding the
+ 'circular sort' containing UOWTasks that were still polymorphic, which
+ could create duplicate entries in the final sort
+
"""
def define_tables(self, metadata):
- global a, b, c
- a = Table('a', metadata,
+ Table('a', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- Column('cid', Integer, ForeignKey('c.id')),
- )
+ Column('cid', Integer, ForeignKey('c.id')))
- b = Table('b', metadata,
+ Table('b', metadata,
Column('id', Integer, ForeignKey("a.id"), primary_key=True),
- Column('data', String(30)),
- )
+ Column('data', String(30)))
- c = Table('c', metadata,
+ Table('c', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- Column('aid', Integer, ForeignKey('a.id', use_alter=True, name="foo")),
- )
-
- def test_flush(self):
- class A(object):pass
- class B(A):pass
- class C(object):pass
+ Column('aid', Integer,
+ ForeignKey('a.id', use_alter=True, name="foo")))
+
+ def setup_classes(self):
+ class A(_base.BasicEntity):
+ pass
+
+ class B(A):
+ pass
+
+ class C(_base.BasicEntity):
+ pass
+ @testing.resolve_artifact_names
+ def test_flush(self):
mapper(A, a, properties={
- 'cs':relation(C, primaryjoin=a.c.cid==c.c.id)
- })
+ 'cs':relation(C, primaryjoin=a.c.cid==c.c.id)})
+
+ mapper(B, b, inherits=A, inherit_condition=b.c.id == a.c.id)
- mapper(B, b, inherits=A, inherit_condition=b.c.id==a.c.id, properties={
- })
mapper(C, c, properties={
- 'arel':relation(A, primaryjoin=a.c.id==c.c.aid)
- })
+ 'arel':relation(A, primaryjoin=a.c.id == c.c.aid)})
sess = create_session()
bobj = B()
- sess.save(bobj)
+ sess.add(bobj)
cobj = C()
- sess.save(cobj)
+ sess.add(cobj)
sess.flush()
class BiDirectionalManyToOneTest(_base.MappedTest):
+
def define_tables(self, metadata):
- global t1, t2, t3, t4
- t1 = Table('t1', metadata,
+ Table('t1', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- Column('t2id', Integer, ForeignKey('t2.id'))
- )
- t2 = Table('t2', metadata,
+ Column('t2id', Integer, ForeignKey('t2.id')))
+ Table('t2', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- Column('t1id', Integer, ForeignKey('t1.id', use_alter=True, name="foo_fk"))
- )
- t3 = Table('t3', metadata,
+ Column('t1id', Integer,
+ ForeignKey('t1.id', use_alter=True, name="foo_fk")))
+ Table('t3', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
Column('t1id', Integer, ForeignKey('t1.id'), nullable=False),
- Column('t2id', Integer, ForeignKey('t2.id'), nullable=False),
- )
+ Column('t2id', Integer, ForeignKey('t2.id'), nullable=False))
- def test_reflush(self):
- class T1(object):pass
- class T2(object):pass
- class T3(object):pass
+ def setup_classes(self):
+ class T1(_base.BasicEntity):
+ pass
+ class T2(_base.BasicEntity):
+ pass
+ class T3(_base.BasicEntity):
+ pass
+ @testing.resolve_artifact_names
+ def setup_mappers(self):
mapper(T1, t1, properties={
- 't2':relation(T2, primaryjoin=t1.c.t2id==t2.c.id)
- })
+ 't2':relation(T2, primaryjoin=t1.c.t2id == t2.c.id)})
mapper(T2, t2, properties={
- 't1':relation(T1, primaryjoin=t2.c.t1id==t1.c.id)
- })
+ 't1':relation(T1, primaryjoin=t2.c.t1id == t1.c.id)})
mapper(T3, t3, properties={
't1':relation(T1),
- 't2':relation(T2)
- })
+ 't2':relation(T2)})
+ @testing.resolve_artifact_names
+ def test_reflush(self):
o1 = T1()
o1.t2 = T2()
sess = create_session()
- sess.save(o1)
+ sess.add(o1)
sess.flush()
- # the bug here is that the dependency sort comes up with T1/T2 in a cycle, but there
- # are no T1/T2 objects to be saved. therefore no "cyclical subtree" gets generated,
- # and one or the other of T1/T2 gets lost, and processors on T3 dont fire off.
- # the test will then fail because the FK's on T3 are not nullable.
+ # the bug here is that the dependency sort comes up with T1/T2 in a
+ # cycle, but there are no T1/T2 objects to be saved. therefore no
+ # "cyclical subtree" gets generated, and one or the other of T1/T2
+ # gets lost, and processors on T3 dont fire off. the test will then
+ # fail because the FK's on T3 are not nullable.
o3 = T3()
o3.t1 = o1
o3.t2 = o1.t2
- sess.save(o3)
+ sess.add(o3)
sess.flush()
+ @testing.resolve_artifact_names
def test_reflush_2(self):
- """a variant on test_reflush()"""
- class T1(object):pass
- class T2(object):pass
- class T3(object):pass
-
- mapper(T1, t1, properties={
- 't2':relation(T2, primaryjoin=t1.c.t2id==t2.c.id)
- })
- mapper(T2, t2, properties={
- 't1':relation(T1, primaryjoin=t2.c.t1id==t1.c.id)
- })
- mapper(T3, t3, properties={
- 't1':relation(T1),
- 't2':relation(T2)
- })
-
+ """A variant on test_reflush()"""
o1 = T1()
o1.t2 = T2()
sess = create_session()
- sess.save(o1)
+ sess.add(o1)
sess.flush()
# in this case, T1, T2, and T3 tasks will all be in the cyclical
# as well.
o1a = T1()
o2a = T2()
- sess.save(o1a)
- sess.save(o2a)
+ sess.add(o1a)
+ sess.add(o2a)
o3b = T3()
o3b.t1 = o1a
o3b.t2 = o2a
- sess.save(o3b)
+ sess.add(o3b)
o3 = T3()
o3.t1 = o1
o3.t2 = o1.t2
- sess.save(o3)
+ sess.add(o3)
sess.flush()
-class BiDirectionalOneToManyTest(TestBase, AssertsExecutionResults):
+
+class BiDirectionalOneToManyTest(_base.MappedTest):
"""tests two mappers with a one-to-many relation to each other."""
- def setUpAll(self):
- global t1, t2, metadata
- metadata = MetaData(testing.db)
- t1 = Table('t1', metadata,
- Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True),
- Column('c2', Integer, ForeignKey('t2.c1'))
- )
- t2 = Table('t2', metadata,
- Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
- Column('c2', Integer, ForeignKey('t1.c1', use_alter=True, name='t1c1_fk'))
- )
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def tearDown(self):
- clear_mappers()
+
+ def define_tables(self, metadata):
+ Table('t1', metadata,
+ Column('c1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('c2', Integer, ForeignKey('t2.c1')))
+
+ Table('t2', metadata,
+ Column('c1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('c2', Integer,
+ ForeignKey('t1.c1', use_alter=True, name='t1c1_fk')))
+
+ def setup_classes(self):
+ class C1(_base.BasicEntity):
+ pass
+
+ class C2(_base.BasicEntity):
+ pass
+
+ @testing.resolve_artifact_names
def testcycle(self):
- class C1(object):pass
- class C2(object):pass
-
- m2 = mapper(C2, t2, properties={
- 'c1s': relation(C1, primaryjoin=t2.c.c1==t1.c.c2, uselist=True)
- })
- m1 = mapper(C1, t1, properties = {
- 'c2s' : relation(C2, primaryjoin=t1.c.c1==t2.c.c2, uselist=True)
- })
+ mapper(C2, t2, properties={
+ 'c1s': relation(C1,
+ primaryjoin=t2.c.c1 == t1.c.c2,
+ uselist=True)})
+ mapper(C1, t1, properties={
+ 'c2s': relation(C2,
+ primaryjoin=t1.c.c1 == t2.c.c2,
+ uselist=True)})
+
a = C1()
b = C2()
c = C1()
d.c1s.append(c)
b.c1s.append(c)
sess = create_session()
- [sess.save(x) for x in [a,b,c,d,e,f]]
+ sess.add_all((a, b, c, d, e, f))
sess.flush()
-class BiDirectionalOneToManyTest2(_base.ORMTest):
- """tests two mappers with a one-to-many relation to each other, with a second one-to-many on one of the mappers"""
- def setUpAll(self):
- global t1, t2, t3, metadata
- metadata = MetaData(testing.db)
- t1 = Table('t1', metadata,
- Column('c1', Integer, Sequence('t1c1_id_seq', optional=True), primary_key=True),
- Column('c2', Integer, ForeignKey('t2.c1')),
- )
- t2 = Table('t2', metadata,
- Column('c1', Integer, Sequence('t2c1_id_seq', optional=True), primary_key=True),
- Column('c2', Integer, ForeignKey('t1.c1', use_alter=True, name='t1c1_fq')),
- )
- t3 = Table('t1_data', metadata,
- Column('c1', Integer, Sequence('t1dc1_id_seq', optional=True), primary_key=True),
- Column('t1id', Integer, ForeignKey('t1.c1')),
- Column('data', String(20)))
- metadata.create_all()
-
- def tearDown(self):
- clear_mappers()
-
- def tearDownAll(self):
- metadata.drop_all()
- def testcycle(self):
- class C1(object):pass
- class C2(object):pass
- class C1Data(object):
- def __init__(self, data=None):
- self.data = data
+class BiDirectionalOneToManyTest2(_base.MappedTest):
+ """Two mappers with a one-to-many relation to each other, with a second one-to-many on one of the mappers"""
- m2 = mapper(C2, t2, properties={
- 'c1s': relation(C1, primaryjoin=t2.c.c1==t1.c.c2, uselist=True)
- })
- m1 = mapper(C1, t1, properties = {
- 'c2s' : relation(C2, primaryjoin=t1.c.c1==t2.c.c2, uselist=True),
- 'data' : relation(mapper(C1Data, t3))
- })
+ def define_tables(self, metadata):
+ Table('t1', metadata,
+ Column('c1', Integer, primary_key=True),
+ Column('c2', Integer, ForeignKey('t2.c1')),
+ test_needs_autoincrement=True)
+
+ Table('t2', metadata,
+ Column('c1', Integer, primary_key=True),
+ Column('c2', Integer,
+ ForeignKey('t1.c1', use_alter=True, name='t1c1_fq')),
+ test_needs_autoincrement=True)
+
+ Table('t1_data', metadata,
+ Column('c1', Integer, primary_key=True),
+ Column('t1id', Integer, ForeignKey('t1.c1')),
+ Column('data', String(20)),
+ test_needs_autoincrement=True)
+
+ def setup_classes(self):
+ class C1(_base.BasicEntity):
+ pass
+ class C2(_base.BasicEntity):
+ pass
+
+ class C1Data(_base.BasicEntity):
+ pass
+
+ @testing.resolve_artifact_names
+ def setup_mappers(self):
+ mapper(C2, t2, properties={
+ 'c1s': relation(C1,
+ primaryjoin=t2.c.c1 == t1.c.c2,
+ uselist=True)})
+ mapper(C1, t1, properties={
+ 'c2s': relation(C2,
+ primaryjoin=t1.c.c1 == t2.c.c2,
+ uselist=True),
+ 'data': relation(mapper(C1Data, t1_data))})
+
+ @testing.resolve_artifact_names
+ def testcycle(self):
a = C1()
b = C2()
c = C1()
a.c2s.append(b)
d.c1s.append(c)
b.c1s.append(c)
- a.data.append(C1Data('c1data1'))
- a.data.append(C1Data('c1data2'))
- c.data.append(C1Data('c1data3'))
+ a.data.append(C1Data(data='c1data1'))
+ a.data.append(C1Data(data='c1data2'))
+ c.data.append(C1Data(data='c1data3'))
sess = create_session()
- [sess.save(x) for x in [a,b,c,d,e,f]]
+ sess.add_all((a, b, c, d, e, f))
sess.flush()
sess.delete(d)
sess.delete(c)
sess.flush()
-class OneToManyManyToOneTest(_base.ORMTest):
- """tests two mappers, one has a one-to-many on the other mapper, the other has a separate many-to-one relationship to the first.
- two tests will have a row for each item that is dependent on the other. without the "post_update" flag, such relationships
- raise an exception when dependencies are sorted."""
- def setUpAll(self):
- global metadata
- metadata = MetaData(testing.db)
- global person
- global ball
- ball = Table('ball', metadata,
- Column('id', Integer, Sequence('ball_id_seq', optional=True), primary_key=True),
- Column('person_id', Integer, ForeignKey('person.id', use_alter=True, name='fk_person_id')),
- Column('data', String(30))
- )
- person = Table('person', metadata,
- Column('id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
- Column('favorite_ball_id', Integer, ForeignKey('ball.id')),
- Column('data', String(30))
- )
-
- metadata.create_all()
-
- def tearDownAll(self):
- metadata.drop_all()
-
- def tearDown(self):
- clear_mappers()
+class OneToManyManyToOneTest(_base.MappedTest):
+ """
- def testcycle(self):
- """this test has a peculiar aspect in that it doesnt create as many dependent
- relationships as the other tests, and revealed a small glitch in the circular dependency sorting.
-
- """
- class Person(object):
+ Tests two mappers, one has a one-to-many on the other mapper, the other
+ has a separate many-to-one relationship to the first. two tests will have
+ a row for each item that is dependent on the other. without the
+ "post_update" flag, such relationships raise an exception when
+ dependencies are sorted.
+
+ """
+
+ def define_tables(self, metadata):
+ Table('ball', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('person_id', Integer,
+ ForeignKey('person.id', use_alter=True, name='fk_person_id')),
+ Column('data', String(30)))
+
+ Table('person', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('favorite_ball_id', Integer, ForeignKey('ball.id')),
+ Column('data', String(30)))
+
+ def setup_classes(self):
+ class Person(_base.BasicEntity):
pass
- class Ball(object):
+ class Ball(_base.BasicEntity):
pass
- Ball.mapper = mapper(Ball, ball)
- Person.mapper = mapper(Person, person, properties= dict(
- balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, remote_side=ball.c.person_id),
- favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=ball.c.id),
- )
- )
+ @testing.resolve_artifact_names
+ def testcycle(self):
+ """
+ This test has a peculiar aspect in that it doesnt create as many
+ dependent relationships as the other tests, and revealed a small
+ glitch in the circular dependency sorting.
+
+ """
+ mapper(Ball, ball)
+ mapper(Person, person, properties=dict(
+ balls=relation(Ball,
+ primaryjoin=ball.c.person_id == person.c.id,
+ remote_side=ball.c.person_id),
+ favorite=relation(Ball,
+ primaryjoin=person.c.favorite_ball_id == ball.c.id,
+ remote_side=ball.c.id)))
b = Ball()
p = Person()
p.balls.append(b)
sess = create_session()
- sess.save(p)
+ sess.add(p)
sess.flush()
-
- def testpostupdate_m2o(self):
- """tests a cycle between two rows, with a post_update on the many-to-one"""
- class Person(object):
- def __init__(self, data):
- self.data = data
- class Ball(object):
- def __init__(self, data):
- self.data = data
-
- Ball.mapper = mapper(Ball, ball)
- Person.mapper = mapper(Person, person, properties= dict(
- balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, remote_side=ball.c.person_id, post_update=False, cascade="all, delete-orphan"),
- favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=person.c.favorite_ball_id, post_update=True),
- )
- )
-
- b = Ball('some data')
- p = Person('some data')
+ @testing.resolve_artifact_names
+ def testpostupdate_m2o(self):
+ """A cycle between two rows, with a post_update on the many-to-one"""
+ mapper(Ball, ball)
+ mapper(Person, person, properties=dict(
+ balls=relation(Ball,
+ primaryjoin=ball.c.person_id == person.c.id,
+ remote_side=ball.c.person_id,
+ post_update=False,
+ cascade="all, delete-orphan"),
+ favorite=relation(Ball,
+ primaryjoin=person.c.favorite_ball_id == ball.c.id,
+ remote_side=person.c.favorite_ball_id,
+ post_update=True)))
+
+ b = Ball(data='some data')
+ p = Person(data='some data')
p.balls.append(b)
- p.balls.append(Ball('some data'))
- p.balls.append(Ball('some data'))
- p.balls.append(Ball('some data'))
- p.favorateBall = b
+ p.balls.append(Ball(data='some data'))
+ p.balls.append(Ball(data='some data'))
+ p.balls.append(Ball(data='some data'))
+ p.favorite = b
sess = create_session()
- sess.save(b)
- sess.save(p)
-
- self.assert_sql(testing.db, lambda: sess.flush(), [
- (
- "INSERT INTO person (favorite_ball_id, data) VALUES (:favorite_ball_id, :data)",
- {'favorite_ball_id': None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- lambda ctx:{'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- lambda ctx:{'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- lambda ctx:{'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- lambda ctx:{'person_id':p.id, 'data':'some data'}
- ),
- (
- "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
- lambda ctx:{'favorite_ball_id':p.favorateBall.id,'person_id':p.id}
- )
- ],
- with_sequences= [
- (
- "INSERT INTO person (id, favorite_ball_id, data) VALUES (:id, :favorite_ball_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'favorite_ball_id': None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0],'person_id':p.id, 'data':'some data'}
- ),
- # heres the post update
- (
- "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
- lambda ctx:{'favorite_ball_id':p.favorateBall.id,'person_id':p.id}
- )
- ])
+ sess.add(b)
+ sess.add(p)
+
+ self.assert_sql(testing.db, sess.flush, [
+ ("INSERT INTO person (favorite_ball_id, data) "
+ "VALUES (:favorite_ball_id, :data)",
+ {'favorite_ball_id': None, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ lambda ctx:{'person_id':p.id, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ lambda ctx:{'person_id':p.id, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ lambda ctx:{'person_id':p.id, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ lambda ctx:{'person_id':p.id, 'data':'some data'}),
+
+ ("UPDATE person SET favorite_ball_id=:favorite_ball_id "
+ "WHERE person.id = :person_id",
+ lambda ctx:{'favorite_ball_id':p.favorite.id, 'person_id':p.id})
+ ],
+
+ with_sequences= [
+ ("INSERT INTO person (id, favorite_ball_id, data) "
+ "VALUES (:id, :favorite_ball_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'favorite_ball_id': None,
+ 'data':'some data'}),
+
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'person_id':p.id,
+ 'data':'some data'}),
+
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'person_id':p.id,
+ 'data':'some data'}),
+
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'person_id':p.id,
+ 'data':'some data'}),
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'person_id':p.id,
+ 'data':'some data'}),
+ # heres the post update
+ ("UPDATE person SET favorite_ball_id=:favorite_ball_id "
+ "WHERE person.id = :person_id",
+ lambda ctx: {'favorite_ball_id':p.favorite.id, 'person_id':p.id})])
+
sess.delete(p)
- self.assert_sql(testing.db, lambda: sess.flush(), [
+
+ self.assert_sql(testing.db, sess.flush, [
# heres the post update (which is a pre-update with deletes)
- (
- "UPDATE person SET favorite_ball_id=:favorite_ball_id WHERE person.id = :person_id",
- lambda ctx:{'person_id': p.id, 'favorite_ball_id': None}
- ),
- (
- "DELETE FROM ball WHERE ball.id = :id",
- None
- # order cant be predicted, but something like:
- #lambda ctx:[{'id': 1L}, {'id': 4L}, {'id': 3L}, {'id': 2L}]
- ),
- (
- "DELETE FROM person WHERE person.id = :id",
- lambda ctx:[{'id': p.id}]
- )
-
-
- ])
+ ("UPDATE person SET favorite_ball_id=:favorite_ball_id "
+ "WHERE person.id = :person_id",
+ lambda ctx: {'person_id': p.id, 'favorite_ball_id': None}),
- def testpostupdate_o2m(self):
- """tests a cycle between two rows, with a post_update on the one-to-many"""
- class Person(object):
- def __init__(self, data):
- self.data = data
+ ("DELETE FROM ball WHERE ball.id = :id",
+ None),
+ # order cant be predicted, but something like:
+ #lambda ctx:[{'id': 1L}, {'id': 4L}, {'id': 3L}, {'id': 2L}]),
- class Ball(object):
- def __init__(self, data):
- self.data = data
+ ("DELETE FROM person WHERE person.id = :id",
+ lambda ctx:[{'id': p.id}])])
- Ball.mapper = mapper(Ball, ball)
- Person.mapper = mapper(Person, person, properties= dict(
- balls = relation(Ball.mapper, primaryjoin=ball.c.person_id==person.c.id, remote_side=ball.c.person_id, cascade="all, delete-orphan", post_update=True, backref='person'),
- favorateBall = relation(Ball.mapper, primaryjoin=person.c.favorite_ball_id==ball.c.id, remote_side=person.c.favorite_ball_id),
- )
- )
- b = Ball('some data')
- p = Person('some data')
+ @testing.resolve_artifact_names
+ def testpostupdate_o2m(self):
+ """A cycle between two rows, with a post_update on the one-to-many"""
+
+ mapper(Ball, ball)
+ mapper(Person, person, properties=dict(
+ balls=relation(Ball,
+ primaryjoin=ball.c.person_id == person.c.id,
+ remote_side=ball.c.person_id,
+ cascade="all, delete-orphan",
+ post_update=True,
+ backref='person'),
+ favorite=relation(Ball,
+ primaryjoin=person.c.favorite_ball_id == ball.c.id,
+ remote_side=person.c.favorite_ball_id)))
+
+ b = Ball(data='some data')
+ p = Person(data='some data')
p.balls.append(b)
- b2 = Ball('some data')
+ b2 = Ball(data='some data')
p.balls.append(b2)
- b3 = Ball('some data')
+ b3 = Ball(data='some data')
p.balls.append(b3)
- b4 = Ball('some data')
+ b4 = Ball(data='some data')
p.balls.append(b4)
- p.favorateBall = b
+ p.favorite = b
sess = create_session()
- [sess.save(x) for x in [b,p,b2,b3,b4]]
-
- self.assert_sql(testing.db, lambda: sess.flush(), [
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- {'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- {'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- {'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (person_id, data) VALUES (:person_id, :data)",
- {'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO person (favorite_ball_id, data) VALUES (:favorite_ball_id, :data)",
- lambda ctx:{'favorite_ball_id':b.id, 'data':'some data'}
- ),
- # heres the post update on each one-to-many item
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b2.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b3.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b4.id}
- ),
- ],
- with_sequences=[
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO ball (id, person_id, data) VALUES (:id, :person_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'person_id':None, 'data':'some data'}
- ),
- (
- "INSERT INTO person (id, favorite_ball_id, data) VALUES (:id, :favorite_ball_id, :data)",
- lambda ctx:{'id':ctx.last_inserted_ids()[0], 'favorite_ball_id':b.id, 'data':'some data'}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b2.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b3.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id':p.id,'ball_id':b4.id}
- ),
- ])
+ sess.add_all((b,p,b2,b3,b4))
+
+ self.assert_sql(testing.db, sess.flush, [
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ {'person_id':None, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ {'person_id':None, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ {'person_id':None, 'data':'some data'}),
+
+ ("INSERT INTO ball (person_id, data) "
+ "VALUES (:person_id, :data)",
+ {'person_id':None, 'data':'some data'}),
+
+ ("INSERT INTO person (favorite_ball_id, data) "
+ "VALUES (:favorite_ball_id, :data)",
+ lambda ctx:{'favorite_ball_id':b.id, 'data':'some data'}),
+
+ # heres the post update on each one-to-many item
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b2.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b3.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b4.id})],
+
+ with_sequences=[
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx: {'id':ctx.last_inserted_ids()[0],
+ 'person_id':None,
+ 'data':'some data'}),
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx:{'id':ctx.last_inserted_ids()[0],
+ 'person_id':None,
+ 'data':'some data'}),
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx:{'id':ctx.last_inserted_ids()[0],
+ 'person_id':None,
+ 'data':'some data'}),
+ ("INSERT INTO ball (id, person_id, data) "
+ "VALUES (:id, :person_id, :data)",
+ lambda ctx:{'id':ctx.last_inserted_ids()[0],
+ 'person_id':None,
+ 'data':'some data'}),
+ ("INSERT INTO person (id, favorite_ball_id, data) "
+ "VALUES (:id, :favorite_ball_id, :data)",
+ lambda ctx:{'id':ctx.last_inserted_ids()[0],
+ 'favorite_ball_id':b.id,
+ 'data':'some data'}),
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b2.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b3.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id':p.id,'ball_id':b4.id})])
sess.delete(p)
- self.assert_sql(testing.db, lambda: sess.flush(), [
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id': None, 'ball_id': b.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id': None, 'ball_id': b2.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id': None, 'ball_id': b3.id}
- ),
- (
- "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
- lambda ctx:{'person_id': None, 'ball_id': b4.id}
- ),
- (
- "DELETE FROM person WHERE person.id = :id",
- lambda ctx:[{'id':p.id}]
- ),
- (
- "DELETE FROM ball WHERE ball.id = :id",
- lambda ctx:[{'id': b.id}, {'id': b2.id}, {'id': b3.id}, {'id': b4.id}]
- )
- ])
-
-class SelfReferentialPostUpdateTest(_base.ORMTest):
- """test using post_update on a single self-referential mapper"""
- def setUpAll(self):
- global metadata, node_table
- metadata = MetaData(testing.db)
- node_table = Table('node', metadata,
- Column('id', Integer, Sequence('nodeid_id_seq', optional=True), primary_key=True),
- Column('path', String(50), nullable=False),
- Column('parent_id', Integer, ForeignKey('node.id'), nullable=True),
- Column('prev_sibling_id', Integer, ForeignKey('node.id'), nullable=True),
- Column('next_sibling_id', Integer, ForeignKey('node.id'), nullable=True)
- )
- node_table.create()
- def tearDownAll(self):
- node_table.drop()
+ self.assert_sql(testing.db, sess.flush, [
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id': None, 'ball_id': b.id}),
- def testbasic(self):
- """test that post_update only fires off when needed.
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id': None, 'ball_id': b2.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id': None, 'ball_id': b3.id}),
+
+ ("UPDATE ball SET person_id=:person_id "
+ "WHERE ball.id = :ball_id",
+ lambda ctx:{'person_id': None, 'ball_id': b4.id}),
+
+ ("DELETE FROM person WHERE person.id = :id",
+ lambda ctx:[{'id':p.id}]),
+
+ ("DELETE FROM ball WHERE ball.id = :id",
+ lambda ctx:[{'id': b.id},
+ {'id': b2.id},
+ {'id': b3.id},
+ {'id': b4.id}])])
- this test case used to produce many superfluous update statements, particularly upon delete"""
- class Node(object):
+
+class SelfReferentialPostUpdateTest(_base.MappedTest):
+ """Post_update on a single self-referential mapper"""
+
+ def define_tables(self, metadata):
+ Table('node', metadata,
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('path', String(50), nullable=False),
+ Column('parent_id', Integer,
+ ForeignKey('node.id'), nullable=True),
+ Column('prev_sibling_id', Integer,
+ ForeignKey('node.id'), nullable=True),
+ Column('next_sibling_id', Integer,
+ ForeignKey('node.id'), nullable=True))
+
+ def setup_classes(self):
+ class Node(_base.BasicEntity):
def __init__(self, path=''):
self.path = path
- n_mapper = mapper(Node, node_table, properties={
+ @testing.resolve_artifact_names
+ def testbasic(self):
+ """Post_update only fires off when needed.
+
+ This test case used to produce many superfluous update statements,
+ particularly upon delete
+
+ """
+
+ mapper(Node, node, properties={
'children': relation(
Node,
- primaryjoin=node_table.c.id==node_table.c.parent_id,
+ primaryjoin=node.c.id==node.c.parent_id,
lazy=True,
cascade="all",
- backref=backref("parent", primaryjoin=node_table.c.parent_id==node_table.c.id, remote_side=node_table.c.id)
- ),
+ backref=backref("parent",
+ primaryjoin=node.c.parent_id == node.c.id,
+ remote_side=node.c.id)),
'prev_sibling': relation(
Node,
- primaryjoin=node_table.c.prev_sibling_id==node_table.c.id,
- remote_side=node_table.c.id,
+ primaryjoin=node.c.prev_sibling_id==node.c.id,
+ remote_side=node.c.id,
lazy=True,
- uselist=False
- ),
+ uselist=False),
'next_sibling': relation(
Node,
- primaryjoin=node_table.c.next_sibling_id==node_table.c.id,
- remote_side=node_table.c.id,
+ primaryjoin=node.c.next_sibling_id==node.c.id,
+ remote_side=node.c.id,
lazy=True,
uselist=False,
- post_update=True
- )
- })
+ post_update=True)})
session = create_session()
assert(about.prev_sibling is None)
append_child(root, stories)
append_child(root, bruce)
- session.save(root)
+ session.add(root)
session.flush()
remove_child(root, cats)
# pre-trigger lazy loader on 'cats' to make the test easier
cats.children
self.assert_sql(testing.db, lambda: session.flush(), [
- (
- "UPDATE node SET prev_sibling_id=:prev_sibling_id WHERE node.id = :node_id",
- lambda ctx:{'prev_sibling_id':about.id, 'node_id':stories.id}
- ),
- (
- "UPDATE node SET next_sibling_id=:next_sibling_id WHERE node.id = :node_id",
- lambda ctx:{'next_sibling_id':stories.id, 'node_id':about.id}
- ),
- (
- "UPDATE node SET next_sibling_id=:next_sibling_id WHERE node.id = :node_id",
- lambda ctx:{'next_sibling_id':None, 'node_id':cats.id}
- ),
- (
- "DELETE FROM node WHERE node.id = :id",
- lambda ctx:[{'id':cats.id}]
- ),
- ])
-
-class SelfReferentialPostUpdateTest2(_base.ORMTest):
- def setUpAll(self):
- global metadata, a_table
- metadata = MetaData(testing.db)
- a_table = Table("a", metadata,
- Column("id", Integer(), primary_key=True),
- Column("fui", String(128)),
- Column("b", Integer(), ForeignKey("a.id")),
- )
- a_table.create()
- def tearDownAll(self):
- a_table.drop()
+ ("UPDATE node SET prev_sibling_id=:prev_sibling_id "
+ "WHERE node.id = :node_id",
+ lambda ctx:{'prev_sibling_id':about.id, 'node_id':stories.id}),
+
+ ("UPDATE node SET next_sibling_id=:next_sibling_id "
+ "WHERE node.id = :node_id",
+ lambda ctx:{'next_sibling_id':stories.id, 'node_id':about.id}),
+
+ ("UPDATE node SET next_sibling_id=:next_sibling_id "
+ "WHERE node.id = :node_id",
+ lambda ctx:{'next_sibling_id':None, 'node_id':cats.id}),
+ ("DELETE FROM node WHERE node.id = :id",
+ lambda ctx:[{'id':cats.id}])])
+
+
+class SelfReferentialPostUpdateTest2(_base.MappedTest):
+
+ def define_tables(self, metadata):
+ Table("a_table", metadata,
+ Column("id", Integer(), primary_key=True),
+ Column("fui", String(128)),
+ Column("b", Integer(), ForeignKey("a_table.id")))
+ def setup_classes(self):
+ class A(_base.BasicEntity):
+ pass
+
+ @testing.resolve_artifact_names
def testbasic(self):
- """test that post_update remembers to be involved in update operations as well,
- since it replaces the normal dependency processing completely [ticket:413]"""
- class a(object):
- def __init__(self, fui):
- self.fui = fui
+ """
+ Test that post_update remembers to be involved in update operations as
+ well, since it replaces the normal dependency processing completely
+ [ticket:413]
+
+ """
- mapper(a, a_table, properties={
- 'foo': relation(a, remote_side=[a_table.c.id], post_update=True),
- })
+ mapper(A, a_table, properties={
+ 'foo': relation(A,
+ remote_side=[a_table.c.id],
+ post_update=True)})
session = create_session()
- f1 = a("f1")
- session.save(f1)
+ f1 = A(fui="f1")
+ session.add(f1)
session.flush()
- f2 = a("f2")
- f2.foo = f1
+ f2 = A(fui="f2", foo=f1)
+
# at this point f1 is already inserted. but we need post_update
# to fire off anyway
- session.save(f2)
+ session.add(f2)
session.flush()
-
session.clear()
- f1 = session.query(a).get(f1.id)
- f2 = session.query(a).get(f2.id)
+
+ f1 = session.query(A).get(f1.id)
+ f2 = session.query(A).get(f2.id)
assert f2.foo is f1
+
if __name__ == "__main__":
testenv.main()
o1 = Order(id=15, description="order 10")
i1 = Item(id=10, description="item 8")
o1.items.append(i1)
- sess.save(o1)
+ sess.add(o1)
sess.flush()
assert o1 in i1.orders.all()
u2 = User(name='ed')
u2.addresses.append(Address(email_address='foo@bar.com'))
u1.addresses.append(Address(email_address='lala@hoho.com'))
- sess.save(u1)
- sess.save(u2)
+ sess.add_all((u1, u2))
sess.flush()
sess.clear()
sess = create_session(autoexpire=False, autocommit=False, autoflush=True)
u1 = User(name='jack')
u1.addresses.append(Address(email_address='lala@hoho.com'))
- sess.save(u1)
+ sess.add(u1)
sess.flush()
sess.commit()
u1.addresses.append(Address(email_address='foo@bar.com'))
u.addresses.append(Address(email_address='d'))
u.addresses.append(Address(email_address='e'))
u.addresses.append(Address(email_address='f'))
- sess.save(u)
+ sess.add(u)
assert Address(email_address='c') == u.addresses[2]
sess.delete(u.addresses[2])
u.addresses.append(Address(email_address='d'))
u.addresses.append(Address(email_address='e'))
u.addresses.append(Address(email_address='f'))
- sess.save(u)
+ sess.add(u)
assert Address(email_address='c') == u.addresses[2]
sess.delete(u.addresses[2])
u.addresses.append(Address(email_address='d'))
u.addresses.append(Address(email_address='e'))
u.addresses.append(Address(email_address='f'))
- sess.save(u)
+ sess.add(u)
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
Address(email_address='d'), Address(email_address='e'), Address(email_address='f')] == sess.query(Address).all()
a.user = u
if saveuser:
- sess.save(u)
+ sess.add(u)
else:
- sess.save(a)
+ sess.add(a)
if not autoflush:
sess.flush()
address = Address()
address.email_address = 'joe@joesdomain.example'
address.user = user
- session.save(user)
+ session.add(user)
session.flush()
session.clear()
import testenv; testenv.configure_for_tests()
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from testlib import *
-from testlib.tables import *
-from testlib import fixtures
+from testlib import sa, testing
+from testlib.sa import Table, Column, Integer, String, ForeignKey
+from testlib.sa.orm import mapper, relation, backref, create_session
+from testlib.testing import eq_
from orm import _base
-class EntityTest(_base.ORMTest):
- """tests mappers that are constructed based on "entity names", which allows the same class
- to have multiple primary mappers """
-
- def setUpAll(self):
- global user1, user2, address1, address2, metadata, ctx
- metadata = MetaData(testing.db)
- ctx = scoped_session(create_session)
-
- user1 = Table('user1', metadata,
- Column('user_id', Integer, Sequence('user1_id_seq', optional=True),
- primary_key=True),
- Column('name', String(60), nullable=False)
- )
- user2 = Table('user2', metadata,
- Column('user_id', Integer, Sequence('user2_id_seq', optional=True),
- primary_key=True),
- Column('name', String(60), nullable=False)
- )
- address1 = Table('address1', metadata,
- Column('address_id', Integer,
- Sequence('address1_id_seq', optional=True),
- primary_key=True),
- Column('user_id', Integer, ForeignKey(user1.c.user_id),
- nullable=False),
- Column('email', String(100), nullable=False)
- )
- address2 = Table('address2', metadata,
- Column('address_id', Integer,
- Sequence('address2_id_seq', optional=True),
- primary_key=True),
- Column('user_id', Integer, ForeignKey(user2.c.user_id),
- nullable=False),
- Column('email', String(100), nullable=False)
- )
- metadata.create_all()
- def tearDownAll(self):
- metadata.drop_all()
- def tearDown(self):
- ctx.clear()
- clear_mappers()
- for t in metadata.table_iterator(reverse=True):
- t.delete().execute()
-
- def testbasic(self):
- """tests a pair of one-to-many mapper structures, establishing that both
- parent and child objects honor the "entity_name" attribute attached to the object
- instances."""
- class User(object):
- def __init__(self, **kw):
- pass
- class Address(object):
- def __init__(self, **kw):
- pass
-
- a1mapper = mapper(Address, address1, entity_name='address1', extension=ctx.extension)
- a2mapper = mapper(Address, address2, entity_name='address2', extension=ctx.extension)
- u1mapper = mapper(User, user1, entity_name='user1', properties ={
- 'addresses':relation(a1mapper)
- }, extension=ctx.extension)
- u2mapper =mapper(User, user2, entity_name='user2', properties={
- 'addresses':relation(a2mapper)
- }, extension=ctx.extension)
-
- u1 = User(_sa_entity_name='user1')
- u1.name = 'this is user 1'
- a1 = Address(_sa_entity_name='address1')
- a1.email='a1@foo.com'
+class EntityTest(_base.MappedTest):
+ """Mappers scoped to an entity_name"""
+
+ def define_tables(self, metadata):
+ Table('user1', metadata,
+ Column('user_id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('name', String(60), nullable=False))
+
+ Table('user2', metadata,
+ Column('user_id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('name', String(60), nullable=False))
+
+ Table('address1', metadata,
+ Column('address_id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('user_id', Integer, ForeignKey('user1.user_id'),
+ nullable=False),
+ Column('email', String(100), nullable=False))
+
+ Table('address2', metadata,
+ Column('address_id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('user_id', Integer, ForeignKey('user2.user_id'),
+ nullable=False),
+ Column('email', String(100), nullable=False))
+
+ def setup_classes(self):
+ class User(_base.BasicEntity):
+ pass
+
+ class Address(_base.BasicEntity):
+ pass
+
+ class Address1(_base.BasicEntity):
+ pass
+
+ class Address2(_base.BasicEntity):
+ pass
+
+ @testing.resolve_artifact_names
+ def test_entity_name_assignment_by_extension(self):
+ """
+ Tests a pair of one-to-many mapper structures, establishing that both
+ parent and child objects honor the "entity_name" attribute attached to
+ the object instances.
+
+ """
+ ctx = sa.orm.scoped_session(create_session)
+
+ ma1 = mapper(Address, address1, entity_name='address1',
+ extension=ctx.extension)
+ ma2 = mapper(Address, address2, entity_name='address2',
+ extension=ctx.extension)
+
+ mapper(User, user1, entity_name='user1',
+ extension=ctx.extension,
+ properties=dict(addresses=relation(ma1)))
+
+ mapper(User, user2, entity_name='user2',
+ extension=ctx.extension,
+ properties=dict(addresses=relation(ma2)))
+
+ u1 = User(name='this is user 1', _sa_entity_name='user1')
+ a1 = Address(email='a1@foo.com', _sa_entity_name='address1')
u1.addresses.append(a1)
- u2 = User(_sa_entity_name='user2')
- u2.name='this is user 2'
- a2 = Address(_sa_entity_name='address2')
- a2.email='a2@foo.com'
+ u2 = User(name='this is user 2', _sa_entity_name='user2')
+ a2 = Address(email='a2@foo.com', _sa_entity_name='address2')
u2.addresses.append(a2)
ctx.flush()
- assert user1.select().execute().fetchall() == [(u1.user_id, u1.name)]
- assert user2.select().execute().fetchall() == [(u2.user_id, u2.name)]
- assert address1.select().execute().fetchall() == [(a1.address_id, u1.user_id, 'a1@foo.com')]
- assert address2.select().execute().fetchall() == [(a1.address_id, u2.user_id, 'a2@foo.com')]
+ eq_(user1.select().execute().fetchall(), [(u1.user_id, u1.name)])
+ eq_(user2.select().execute().fetchall(), [(u2.user_id, u2.name)])
+ eq_(address1.select().execute().fetchall(),
+ [(a1.address_id, u1.user_id, 'a1@foo.com')])
+ eq_(address2.select().execute().fetchall(),
+ [(a1.address_id, u2.user_id, 'a2@foo.com')])
ctx.clear()
u1list = ctx.query(User, entity_name='user1').all()
u2list = ctx.query(User, entity_name='user2').all()
- assert len(u1list) == len(u2list) == 1
+ eq_(len(u1list), 1)
+ eq_(len(u2list), 1)
assert u1list[0] is not u2list[0]
- assert len(u1list[0].addresses) == len(u2list[0].addresses) == 1
+
+ eq_(len(u1list[0].addresses), 1)
+ eq_(len(u2list[0].addresses), 1)
u1 = ctx.query(User, entity_name='user1').first()
ctx.refresh(u1)
ctx.expire(u1)
+ @testing.resolve_artifact_names
+ def test_cascade(self):
+ ma1 = mapper(Address, address1, entity_name='address1')
+ ma2 = mapper(Address, address2, entity_name='address2')
- def testcascade(self):
- """same as testbasic but relies on session cascading"""
- class User(object):pass
- class Address(object):pass
-
- a1mapper = mapper(Address, address1, entity_name='address1')
- a2mapper = mapper(Address, address2, entity_name='address2')
- u1mapper = mapper(User, user1, entity_name='user1', properties ={
- 'addresses':relation(a1mapper)
- })
- u2mapper =mapper(User, user2, entity_name='user2', properties={
- 'addresses':relation(a2mapper)
- })
+ mapper(User, user1, entity_name='user1', properties=dict(
+ addresses=relation(ma1)))
+ mapper(User, user2, entity_name='user2', properties=dict(
+ addresses=relation(ma2)))
sess = create_session()
- u1 = User()
- u1.name = 'this is user 1'
- sess.save(u1, entity_name='user1')
- a1 = Address()
- a1.email='a1@foo.com'
+ u1 = User(name='this is user 1')
+ sess.add(u1, entity_name='user1')
+
+ a1 = Address(email='a1@foo.com')
u1.addresses.append(a1)
- u2 = User()
- u2.name='this is user 2'
- a2 = Address()
- a2.email='a2@foo.com'
+ u2 = User(name='this is user 2')
+ a2 = Address(email='a2@foo.com')
u2.addresses.append(a2)
- sess.save(u2, entity_name='user2')
- print u2.__dict__
+ sess.add(u2, entity_name='user2')
sess.flush()
- assert user1.select().execute().fetchall() == [(u1.user_id, u1.name)]
- assert user2.select().execute().fetchall() == [(u2.user_id, u2.name)]
- assert address1.select().execute().fetchall() == [(a1.address_id, u1.user_id, 'a1@foo.com')]
- assert address2.select().execute().fetchall() == [(a1.address_id, u2.user_id, 'a2@foo.com')]
+ eq_(user1.select().execute().fetchall(), [(u1.user_id, u1.name)])
+ eq_(user2.select().execute().fetchall(), [(u2.user_id, u2.name)])
+ eq_(address1.select().execute().fetchall(),
+ [(a1.address_id, u1.user_id, 'a1@foo.com')])
+ eq_(address2.select().execute().fetchall(),
+ [(a1.address_id, u2.user_id, 'a2@foo.com')])
sess.clear()
u1list = sess.query(User, entity_name='user1').all()
u2list = sess.query(User, entity_name='user2').all()
- assert len(u1list) == len(u2list) == 1
+ eq_(len(u1list), 1)
+ eq_(len(u2list), 1)
assert u1list[0] is not u2list[0]
- assert len(u1list[0].addresses) == len(u2list[0].addresses) == 1
-
- def testpolymorphic(self):
- """tests that entity_name can be used to have two kinds of relations on the same class."""
- class User(object):
- def __init__(self, **kw):
- pass
- class Address1(object):
- def __init__(self, **kw):
- pass
- class Address2(object):
- def __init__(self, **kw):
- pass
-
- a1mapper = mapper(Address1, address1, extension=ctx.extension)
- a2mapper = mapper(Address2, address2, extension=ctx.extension)
- u1mapper = mapper(User, user1, entity_name='user1', properties ={
- 'addresses':relation(a1mapper)
- }, extension=ctx.extension)
- u2mapper =mapper(User, user2, entity_name='user2', properties={
- 'addresses':relation(a2mapper)
- }, extension=ctx.extension)
-
- u1 = User(_sa_entity_name='user1')
- u1.name = 'this is user 1'
- a1 = Address1()
- a1.email='a1@foo.com'
+ eq_(len(u1list[0].addresses), 1)
+ eq_(len(u2list[0].addresses), 1)
+
+ @testing.resolve_artifact_names
+ def test_polymorphic(self):
+ """entity_name can be used to have two kinds of relations on the same class."""
+ ctx = sa.orm.scoped_session(create_session)
+
+ ma1 = mapper(Address1, address1, extension=ctx.extension)
+ ma2 = mapper(Address2, address2, extension=ctx.extension)
+
+ mapper(User, user1, entity_name='user1', extension=ctx.extension,
+ properties=dict(
+ addresses=relation(ma1)))
+
+ mapper(User, user2, entity_name='user2', extension=ctx.extension,
+ properties=dict(
+ addresses=relation(ma2)))
+
+ u1 = User(name='this is user 1', _sa_entity_name='user1')
+ a1 = Address1(email='a1@foo.com')
u1.addresses.append(a1)
- u2 = User(_sa_entity_name='user2')
- u2.name='this is user 2'
- a2 = Address2()
- a2.email='a2@foo.com'
+ u2 = User(name='this is user 2', _sa_entity_name='user2')
+ a2 = Address2(email='a2@foo.com')
u2.addresses.append(a2)
ctx.flush()
- assert user1.select().execute().fetchall() == [(u1.user_id, u1.name)]
- assert user2.select().execute().fetchall() == [(u2.user_id, u2.name)]
- assert address1.select().execute().fetchall() == [(a1.address_id, u1.user_id, 'a1@foo.com')]
- assert address2.select().execute().fetchall() == [(a1.address_id, u2.user_id, 'a2@foo.com')]
+ eq_(user1.select().execute().fetchall(), [(u1.user_id, u1.name)])
+ eq_(user2.select().execute().fetchall(), [(u2.user_id, u2.name)])
+ eq_(address1.select().execute().fetchall(),
+ [(a1.address_id, u1.user_id, 'a1@foo.com')])
+ eq_(address2.select().execute().fetchall(),
+ [(a1.address_id, u2.user_id, 'a2@foo.com')])
ctx.clear()
u1list = ctx.query(User, entity_name='user1').all()
u2list = ctx.query(User, entity_name='user2').all()
- assert len(u1list) == len(u2list) == 1
+ eq_(len(u1list), 1)
+ eq_(len(u2list), 1)
assert u1list[0] is not u2list[0]
- assert len(u1list[0].addresses) == len(u2list[0].addresses) == 1
- # the lazy load requires that setup_loader() check that the correct LazyLoader
- # is setting up for each load
+ eq_(len(u1list[0].addresses), 1)
+ eq_(len(u2list[0].addresses), 1)
+
+ # the lazy load requires that setup_loader() check that the correct
+ # LazyLoader is setting up for each load
assert isinstance(u1list[0].addresses[0], Address1)
assert isinstance(u2list[0].addresses[0], Address2)
+ @testing.resolve_artifact_names
def testpolymorphic_deferred(self):
- """test that deferred columns load properly using entity names"""
- class User(object):
- def __init__(self, **kwargs):
- pass
- u1mapper = mapper(User, user1, entity_name='user1', properties ={
- 'name':deferred(user1.c.name)
- }, extension=ctx.extension)
- u2mapper =mapper(User, user2, entity_name='user2', properties={
- 'name':deferred(user2.c.name)
- }, extension=ctx.extension)
-
- u1 = User(_sa_entity_name='user1')
- u1.name = 'this is user 1'
-
- u2 = User(_sa_entity_name='user2')
- u2.name='this is user 2'
+ """Deferred columns load properly using entity names"""
- ctx.flush()
- assert user1.select().execute().fetchall() == [(u1.user_id, u1.name)]
- assert user2.select().execute().fetchall() == [(u2.user_id, u2.name)]
+ mapper(User, user1, entity_name='user1', properties=dict(
+ name=sa.orm.deferred(user1.c.name)))
+ mapper(User, user2, entity_name='user2', properties=dict(
+ name=sa.orm.deferred(user2.c.name)))
- ctx.clear()
- u1list = ctx.query(User, entity_name='user1').all()
- u2list = ctx.query(User, entity_name='user2').all()
- assert len(u1list) == len(u2list) == 1
+ u1 = User(name='this is user 1')
+ u2 = User(name='this is user 2')
+
+ session = create_session()
+ session.add(u1, entity_name='user1')
+ session.add(u2, entity_name='user2')
+ session.flush()
+
+ eq_(user1.select().execute().fetchall(), [(u1.user_id, u1.name)])
+ eq_(user2.select().execute().fetchall(), [(u2.user_id, u2.name)])
+
+ session.clear()
+ u1list = session.query(User, entity_name='user1').all()
+ u2list = session.query(User, entity_name='user2').all()
+
+ eq_(len(u1list), 1)
+ eq_(len(u2list), 1)
assert u1list[0] is not u2list[0]
- # the deferred column load requires that setup_loader() check that the correct DeferredColumnLoader
- # is setting up for each load
- assert u1list[0].name == 'this is user 1'
- assert u2list[0].name == 'this is user 2'
+
+ # the deferred column load requires that setup_loader() check that the
+ # correct DeferredColumnLoader is setting up for each load
+ eq_(u1list[0].name, 'this is user 1')
+ eq_(u2list[0].name, 'this is user 2')
+
class SelfReferentialTest(_base.MappedTest):
def define_tables(self, metadata):
- global nodes
-
- nodes = Table('nodes', metadata,
- Column('id', Integer, primary_key=True),
- Column('parent_id', Integer, ForeignKey('nodes.id')),
- Column('data', String(50)),
- Column('type', String(50)),
- )
-
- # fails inconsistently. entity name needs deterministic
- # instrumentation.
- def dont_test_relation(self):
- class Node(fixtures.Base):
+ Table('nodes', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('parent_id', Integer, ForeignKey('nodes.id')),
+ Column('data', String(50)),
+ Column('type', String(50)))
+
+ def setup_classes(self):
+ class Node(_base.ComparableEntity):
pass
-
+
+ # fails inconsistently. entity name needs deterministic instrumentation.
+ @testing.resolve_artifact_names
+ def FIXME_test_relation(self):
foonodes = nodes.select().where(nodes.c.type=='foo').alias()
barnodes = nodes.select().where(nodes.c.type=='bar').alias()
-
+
# TODO: the order of instrumentation here is not deterministic;
# therefore the test fails sporadically since "Node.data" references
# different mappers at different times
m1 = mapper(Node, nodes)
m2 = mapper(Node, foonodes, entity_name='foo')
m3 = mapper(Node, barnodes, entity_name='bar')
-
- m1.add_property('foonodes', relation(m2, primaryjoin=nodes.c.id==foonodes.c.parent_id,
- backref=backref('foo_parent', remote_side=nodes.c.id, primaryjoin=nodes.c.id==foonodes.c.parent_id)))
- m1.add_property('barnodes', relation(m3, primaryjoin=nodes.c.id==barnodes.c.parent_id,
- backref=backref('bar_parent', remote_side=nodes.c.id, primaryjoin=nodes.c.id==barnodes.c.parent_id)))
-
+
+ m1.add_property('foonodes', relation(
+ m2,
+ primaryjoin=nodes.c.id == foonodes.c.parent_id,
+ backref=backref('foo_parent',
+ remote_side=nodes.c.id,
+ primaryjoin=nodes.c.id==foonodes.c.parent_id)))
+
+ m1.add_property('barnodes', relation(
+ m3,
+ primaryjoin=nodes.c.id==barnodes.c.parent_id,
+ backref=backref('bar_parent',
+ remote_side=nodes.c.id,
+ primaryjoin=nodes.c.id==barnodes.c.parent_id)))
+
sess = create_session()
-
+
n1 = Node(data='n1', type='bat')
n1.foonodes.append(Node(data='n2', type='foo'))
Node(data='n3', type='bar', bar_parent=n1)
- sess.save(n1)
+ sess.add(n1)
sess.flush()
sess.clear()
-
- self.assertEquals(sess.query(Node, entity_name="bar").one(), Node(data='n3'))
- self.assertEquals(sess.query(Node).filter(Node.data=='n1').one(), Node(data='n1', foonodes=[Node(data='n2')], barnodes=[Node(data='n3')]))
+
+ eq_(sess.query(Node, entity_name="bar").one(),
+ Node(data='n3'))
+ eq_(sess.query(Node).filter(Node.data=='n1').one(),
+ Node(data='n1',
+ foonodes=[Node(data='n2')],
+ barnodes=[Node(data='n3')]))
+
if __name__ == "__main__":
testenv.main()
-"""test attribute/instance expiration, deferral of attributes, etc."""
+"""Attribute/instance expiration, deferral of attributes, etc."""
import testenv; testenv.configure_for_tests()
import gc
from testlib.sa.orm import mapper, relation, create_session, attributes
from orm import _base, _fixtures
+
class ExpireTest(_fixtures.FixtureTest):
@testing.resolve_artifact_names
mapper(User, users)
sess = create_session()
u = User(id=15)
- sess.save(u)
+ sess.add(u)
self.assertRaises(sa.exc.InvalidRequestError, sess.expire, u, ['name'])
@testing.resolve_artifact_names
sess.expunge(u)
attributes.instance_state(u).key = None
assert 'name' not in u.__dict__
- sess.save(u)
+ sess.add(u)
assert u.name == 'jack'
@testing.resolve_artifact_names
a = Address(id=10, email_address='lala')
u.addresses.append(a)
- s.save(u)
+ s.add(u)
s.flush()
s.clear()
u = s.query(User).filter(User.name=='Justin').one()
import testenv; testenv.configure_for_tests()
-from sqlalchemy import MetaData, Table, Column, Integer, ForeignKey
-from sqlalchemy import util
-from sqlalchemy.orm import attributes
-from sqlalchemy.orm import create_session
-from sqlalchemy.orm import interfaces
-from sqlalchemy.orm import mapper
-from sqlalchemy.orm import relation
+from testlib import sa
+from testlib.sa import MetaData, Table, Column, Integer, ForeignKey
+from testlib.sa.orm import mapper, relation, create_session, attributes
from testlib.testing import eq_, ne_
from testlib.compat import _function_named
from orm import _base
class NativeInstrumentationTest(_base.ORMTest):
- @with_lookup_strategy(util.symbol('native'))
+ @with_lookup_strategy(sa.util.symbol('native'))
def test_register_reserved_attribute(self):
class T(object): pass
fails('install_descriptor', sa)
fails('install_descriptor', ma)
- @with_lookup_strategy(util.symbol('native'))
+ @with_lookup_strategy(sa.util.symbol('native'))
def test_mapped_stateattr(self):
t = Table('t', MetaData(),
Column('id', Integer, primary_key=True),
self.assertRaises(KeyError, mapper, T, t)
- @with_lookup_strategy(util.symbol('native'))
+ @with_lookup_strategy(sa.util.symbol('native'))
def test_mapped_managerattr(self):
t = Table('t', MetaData(),
Column('id', Integer, primary_key=True),
b.a = a
session = create_session()
- session.save(b)
+ session.add(b)
assert a in session, "base is %s" % base
def test_compileonattr_rel_backref_b(self):
b.a = a
session = create_session()
- session.save(a)
+ session.add(a)
assert b in session, 'base: %s' % base
def test_compileonattr_rel_entity_name(self):
def test_nativeext_interfaceexact(self):
class A(object):
- __sa_instrumentation_manager__ = interfaces.InstrumentationManager
+ __sa_instrumentation_manager__ = sa.orm.interfaces.InstrumentationManager
attributes.register_class(A)
ne_(type(attributes.manager_of_class(A)), attributes.ClassManager)
sess = create_session()
ad1 = Address(email_address='somenewaddress', id=12)
- sess.save(ad1)
+ sess.add(ad1)
sess.flush()
sess.clear()
p5 = Place('place5')
p6 = Place('place6')
p7 = Place('place7')
- [sess.save(x) for x in [p1,p2,p3,p4,p5,p6,p7]]
+ sess.add_all((p1, p2, p3, p4, p5, p6, p7))
p1.places.append(p2)
p1.places.append(p3)
p5.places.append(p6)
tran.outputs.append(Place('place2'))
tran.outputs.append(Place('place3'))
sess = create_session()
- sess.save(tran)
+ sess.add(tran)
sess.flush()
sess.clear()
p3.inputs.append(t2)
p1.outputs.append(t1)
sess = create_session()
- [sess.save(x) for x in [t1,t2,t3,p1,p2,p3]]
+ sess.add_all((t1, t2, t3,p1, p2, p3))
sess.flush()
self.assert_result([t1], Transition, {'outputs': (Place, [{'name':'place3'}, {'name':'place1'}])})
c3.students.append(s1)
self.assert_(len(s1.courses) == 3)
self.assert_(len(c1.students) == 1)
- sess.save(s1)
+ sess.add(s1)
sess.flush()
sess.clear()
s = sess.query(Student).filter_by(name='Student1').one()
s1.courses.append(c1)
s1.courses.append(c2)
c3.students.append(s1)
- sess.save(s1)
+ sess.add(s1)
sess.flush()
sess.delete(s1)
sess.flush()
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
for x in [a1,a2,a3]:
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.clear()
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
for x in [a1,a2,a3]:
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.clear()
b1 = B(col3='b1')
b2 = B(col3='b2')
for x in [a1,a2,b1, b2]:
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.clear()
a1.bs.append(b1)
a2.bs.append(b2)
for x in [a1,a2]:
- sess.save(x)
+ sess.add(x)
sess.flush()
sess.clear()
from testlib import sa, testing
from testlib.sa.util import OrderedSet
from testlib.sa.orm import mapper, relation, create_session
-from testlib.testing import eq_
+from testlib.testing import eq_, ne_
from orm import _base, _fixtures
on_load = self.on_load_tracker(User)
u = User(id=7, name='fred')
- assert on_load.called == 0
+ eq_(on_load.called, 0)
u2 = sess.merge(u)
- assert on_load.called == 1
+ eq_(on_load.called, 1)
assert u2 in sess
- self.assertEquals(u2, User(id=7, name='fred'))
+ eq_(u2, User(id=7, name='fred'))
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).first(), User(id=7, name='fred'))
+ eq_(sess.query(User).first(), User(id=7, name='fred'))
@testing.resolve_artifact_names
def test_transient_to_pending_collection(self):
- mapper(User, users, properties={'addresses':relation(Address, backref='user', collection_class=OrderedSet)})
+ mapper(User, users, properties={
+ 'addresses': relation(Address, backref='user',
+ collection_class=OrderedSet)})
mapper(Address, addresses)
on_load = self.on_load_tracker(User)
self.on_load_tracker(Address, on_load)
Address(id=1, email_address='fred1'),
Address(id=2, email_address='fred2'),
]))
- assert on_load.called == 0
+ eq_(on_load.called, 0)
sess = create_session()
sess.merge(u)
- assert on_load.called == 3
+ eq_(on_load.called, 3)
merged_users = [e for e in sess if isinstance(e, User)]
- assert len(merged_users) == 1
+ eq_(len(merged_users), 1)
assert merged_users[0] is not u
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).one(),
+ eq_(sess.query(User).one(),
User(id=7, name='fred', addresses=OrderedSet([
Address(id=1, email_address='fred1'),
Address(id=2, email_address='fred2'),
sess = create_session()
u = User(id=7, name='fred')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
- assert on_load.called == 0
+ eq_(on_load.called, 0)
_u2 = u2 = User(id=7, name='fred jones')
- assert on_load.called == 0
+ eq_(on_load.called, 0)
u2 = sess.merge(u2)
assert u2 is not _u2
- assert on_load.called == 1
+ eq_(on_load.called, 1)
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).first(), User(id=7, name='fred jones'))
- assert on_load.called == 2
+ eq_(sess.query(User).first(), User(id=7, name='fred jones'))
+ eq_(on_load.called, 2)
@testing.resolve_artifact_names
def test_transient_to_persistent_collection(self):
mapper(User, users, properties={
'addresses':relation(Address,
backref='user',
- collection_class=OrderedSet, cascade="all, delete-orphan")
+ collection_class=OrderedSet,
+ cascade="all, delete-orphan")
})
mapper(Address, addresses)
Address(id=2, email_address='fred2'),
]))
sess = create_session()
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
- assert on_load.called == 0
+ eq_(on_load.called, 0)
u = User(id=7, name='fred', addresses=OrderedSet([
Address(id=3, email_address='fred3'),
u = sess.merge(u)
- assert on_load.called == 5, on_load.called # 1. merges User object. updates into session.
- # 2.,3. merges Address ids 3 & 4, saves into session.
- # 4.,5. loads pre-existing elements in "addresses" collection,
- # marks as deleted, Address ids 1 and 2.
- self.assertEquals(u,
+ # 1. merges User object. updates into session.
+ # 2.,3. merges Address ids 3 & 4, saves into session.
+ # 4.,5. loads pre-existing elements in "addresses" collection,
+ # marks as deleted, Address ids 1 and 2.
+ eq_(on_load.called, 5)
+
+ eq_(u,
User(id=7, name='fred', addresses=OrderedSet([
Address(id=3, email_address='fred3'),
Address(id=4, email_address='fred4'),
)
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).one(),
+ eq_(sess.query(User).one(),
User(id=7, name='fred', addresses=OrderedSet([
Address(id=3, email_address='fred3'),
Address(id=4, email_address='fred4'),
@testing.resolve_artifact_names
def test_detached_to_persistent_collection(self):
- mapper(User, users, properties={'addresses':relation(Address, backref='user', collection_class=OrderedSet)})
+ mapper(User, users, properties={
+ 'addresses':relation(Address,
+ backref='user',
+ collection_class=OrderedSet)})
mapper(Address, addresses)
on_load = self.on_load_tracker(User)
self.on_load_tracker(Address, on_load)
Address(id=2, email_address='fred2'),
]))
sess = create_session()
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
u.addresses.add(Address(id=3, email_address='fred3'))
u.addresses.remove(a)
- assert on_load.called == 0
+ eq_(on_load.called, 0)
u = sess.merge(u)
- assert on_load.called == 4
+ eq_(on_load.called, 4)
sess.flush()
sess.clear()
- self.assertEquals(sess.query(User).first(),
+ eq_(sess.query(User).first(),
User(id=7, name='fred jones', addresses=OrderedSet([
Address(id=2, email_address='fred2'),
- Address(id=3, email_address='fred3'),
- ]))
- )
+ Address(id=3, email_address='fred3')])))
@testing.resolve_artifact_names
def test_unsaved_cascade(self):
- """test merge of a transient entity with two child transient entities, with a bidirectional relation."""
+ """Merge of a transient entity with two child transient entities, with a bidirectional relation."""
mapper(User, users, properties={
- 'addresses':relation(mapper(Address, addresses), cascade="all", backref="user")
+ 'addresses':relation(mapper(Address, addresses),
+ cascade="all", backref="user")
})
on_load = self.on_load_tracker(User)
self.on_load_tracker(Address, on_load)
u.addresses.append(a2)
u2 = sess.merge(u)
- assert on_load.called == 3
+ eq_(on_load.called, 3)
+
+ eq_(u,
+ User(id=7, name='fred', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@bar.com')]))
+ eq_(u2,
+ User(id=7, name='fred', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@bar.com')]))
- self.assertEquals(u, User(id=7, name='fred', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@bar.com')]))
- self.assertEquals(u2, User(id=7, name='fred', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@bar.com')]))
sess.flush()
sess.clear()
u2 = sess.query(User).get(7)
- self.assertEquals(u2, User(id=7, name='fred', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@bar.com')]))
- assert on_load.called == 6
+ eq_(u2, User(id=7, name='fred', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@bar.com')]))
+ eq_(on_load.called, 6)
@testing.resolve_artifact_names
def test_attribute_cascade(self):
- """test merge of a persistent entity with two child persistent entities."""
+ """Merge of a persistent entity with two child persistent entities."""
mapper(User, users, properties={
'addresses':relation(mapper(Address, addresses), backref='user')
# set up data and save
u = User(id=7, name='fred', addresses=[
Address(email_address='foo@bar.com'),
- Address(email_address = 'hoho@la.com')
- ])
- sess.save(u)
+ Address(email_address = 'hoho@la.com')])
+ sess.add(u)
sess.flush()
# assert data was saved
sess2 = create_session()
u2 = sess2.query(User).get(7)
- self.assertEquals(u2, User(id=7, name='fred', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@la.com')]))
+ eq_(u2,
+ User(id=7, name='fred', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@la.com')]))
# make local changes to data
u.name = 'fred2'
u.addresses[1].email_address = 'hoho@lalala.com'
- assert on_load.called == 3
+ eq_(on_load.called, 3)
# new session, merge modified data into session
sess3 = create_session()
u3 = sess3.merge(u)
- assert on_load.called == 6
+ eq_(on_load.called, 6)
# ensure local changes are pending
- self.assertEquals(u3, User(id=7, name='fred2', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@lalala.com')]))
+ eq_(u3, User(id=7, name='fred2', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@lalala.com')]))
# save merged data
sess3.flush()
# assert modified/merged data was saved
sess.clear()
u = sess.query(User).get(7)
- self.assertEquals(u, User(id=7, name='fred2', addresses=[Address(email_address='foo@bar.com'), Address(email_address='hoho@lalala.com')]))
- assert on_load.called == 9
+ eq_(u, User(id=7, name='fred2', addresses=[
+ Address(email_address='foo@bar.com'),
+ Address(email_address='hoho@lalala.com')]))
+ eq_(on_load.called, 9)
# merge persistent object into another session
sess4 = create_session()
sess4.flush()
# no changes; therefore flush should do nothing
self.assert_sql_count(testing.db, go, 0)
- assert on_load.called == 12
+ eq_(on_load.called, 12)
# test with "dontload" merge
sess5 = create_session()
# but also, dont_load wipes out any difference in committed state,
# so no flush at all
self.assert_sql_count(testing.db, go, 0)
- assert on_load.called == 15
+ eq_(on_load.called, 15)
sess4 = create_session()
u = sess4.merge(u, dont_load=True)
sess4.flush()
# afafds change flushes
self.assert_sql_count(testing.db, go, 1)
- assert on_load.called == 18
+ eq_(on_load.called, 18)
sess5 = create_session()
u2 = sess5.query(User).get(u.id)
- assert u2.name == 'fred2'
- assert u2.addresses[1].email_address == 'afafds'
- assert on_load.called == 21
+ eq_(u2.name, 'fred2')
+ eq_(u2.addresses[1].email_address, 'afafds')
+ eq_(on_load.called, 21)
@testing.resolve_artifact_names
def test_one_to_many_cascade(self):
a2 = Address(email_address='foo@quux')
u.addresses.extend([a1, a2])
- sess.save(u)
+ sess.add(u)
sess.flush()
eq_(on_load.called, 0)
o.items.append(i1)
o.items.append(i2)
- sess.save(o)
+ sess.add(o)
sess.flush()
eq_(on_load.called, 0)
a1.email_address='foo@bar.com'
u.address = a1
- sess.save(u)
+ sess.add(u)
sess.flush()
- assert on_load.called == 0
+ eq_(on_load.called, 0)
sess2 = create_session()
u2 = sess2.query(User).get(7)
- assert on_load.called == 1
+ eq_(on_load.called, 1)
u2.name = 'fred2'
u2.address.email_address = 'hoho@lalala.com'
- assert on_load.called == 2
+ eq_(on_load.called, 2)
u3 = sess.merge(u2)
- assert on_load.called == 2
+ eq_(on_load.called, 2)
assert u3 is u
@testing.resolve_artifact_names
@testing.resolve_artifact_names
def test_dontload_with_backrefs(self):
- """test that dontload populates relations in both directions without requiring a load"""
-
+ """dontload populates relations in both directions without requiring a load"""
mapper(User, users, properties={
'addresses':relation(mapper(Address, addresses), backref='user')
})
- u = User(id=7, name='fred', addresses=[Address(email_address='ad1'), Address(email_address='ad2')])
+ u = User(id=7, name='fred', addresses=[
+ Address(email_address='ad1'),
+ Address(email_address='ad2')])
sess = create_session()
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.close()
assert 'user' in u.addresses[1].__dict__
sess = create_session()
u2 = sess.merge(u, dont_load=True)
assert 'user' in u2.addresses[1].__dict__
- self.assertEquals(u2.addresses[1].user, User(id=7, name='fred'))
+ eq_(u2.addresses[1].user, User(id=7, name='fred'))
sess.expire(u2.addresses[1], ['user'])
assert 'user' not in u2.addresses[1].__dict__
sess = create_session()
u = sess.merge(u2, dont_load=True)
assert 'user' not in u.addresses[1].__dict__
- self.assertEquals(u.addresses[1].user, User(id=7, name='fred'))
+ eq_(u.addresses[1].user, User(id=7, name='fred'))
@testing.resolve_artifact_names
def test_dontload_with_eager(self):
- """this test illustrates that with dont_load=True, we can't just
- copy the committed_state of the merged instance over; since it references collection objects
- which themselves are to be merged. This committed_state would instead need to be piecemeal
- 'converted' to represent the correct objects.
- However, at the moment I'd rather not support this use case; if you are merging with dont_load=True,
- you're typically dealing with caching and the merged objects shouldnt be "dirty".
"""
+ This test illustrates that with dont_load=True, we can't just copy the
+ committed_state of the merged instance over; since it references
+ collection objects which themselves are to be merged. This
+ committed_state would instead need to be piecemeal 'converted' to
+ represent the correct objects. However, at the moment I'd rather not
+ support this use case; if you are merging with dont_load=True, you're
+ typically dealing with caching and the merged objects shouldnt be
+ 'dirty'.
+
+ """
mapper(User, users, properties={
'addresses':relation(mapper(Address, addresses))
})
a1.email_address='foo@bar.com'
u.addresses.append(a1)
- sess.save(u)
+ sess.add(u)
sess.flush()
sess2 = create_session()
@testing.resolve_artifact_names
def test_dont_load_disallows_dirty(self):
- """dont_load doesnt support 'dirty' objects right now (see test_dont_load_with_eager()).
- Therefore lets assert it."""
+ """dont_load doesnt support 'dirty' objects right now
+ (see test_dont_load_with_eager()). Therefore lets assert it.
+
+ """
mapper(User, users)
sess = create_session()
u = User()
u.id = 7
u.name = "fred"
- sess.save(u)
+ sess.add(u)
sess.flush()
u.name = 'ed'
sess2.merge(u, dont_load=True)
assert False
except sa.exc.InvalidRequestError, e:
- assert "merge() with dont_load=True option does not support objects marked as 'dirty'. flush() all changes on mapped instances before merging with dont_load=True." in str(e)
+ assert ("merge() with dont_load=True option does not support "
+ "objects marked as 'dirty'. flush() all changes on mapped "
+ "instances before merging with dont_load=True.") in str(e)
u2 = sess2.query(User).get(7)
@testing.resolve_artifact_names
def test_dont_load_sets_entityname(self):
- """test that a dont_load-merged entity has entity_name set, has_mapper() passes, and lazyloads work"""
+ """dont_load-merged entity has entity_name set, has_mapper() passes, and lazyloads work"""
+
mapper(User, users, properties={
- 'addresses':relation(mapper(Address, addresses),uselist = True)
- })
+ 'addresses':relation(mapper(Address, addresses), uselist=True)})
+
sess = create_session()
u = User()
u.id = 7
a1.email_address='foo@bar.com'
u.addresses.append(a1)
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.clear()
assert state.entity_name is not sa.orm.attributes.NO_ENTITY_NAME
assert sa.orm.mapperlib.has_mapper(u2)
def go():
- assert u2.addresses != []
- assert len(u2.addresses) == 1
+ ne_( u2.addresses, [])
+ eq_(len(u2.addresses), 1)
self.assert_sql_count(testing.db, go, 1)
@testing.resolve_artifact_names
def test_dont_load_sets_backrefs(self):
mapper(User, users, properties={
- 'addresses':relation(mapper(Address, addresses),backref='user')
- })
+ 'addresses':relation(mapper(Address, addresses),backref='user')})
+
sess = create_session()
u = User()
u.id = 7
a1.email_address='foo@bar.com'
u.addresses.append(a1)
- sess.save(u)
+ sess.add(u)
sess.flush()
assert u.addresses[0].user is u
@testing.resolve_artifact_names
def test_dont_load_preserves_parents(self):
- """test that merge with dont_load does not trigger a 'delete-orphan' operation.
-
- merge with dont_load sets attributes without using events. this means the
- 'hasparent' flag is not propagated to the newly merged instance. in fact this
- works out OK, because the '_state.parents' collection on the newly
- merged instance is empty; since the mapper doesn't see an active 'False' setting
- in this collection when _is_orphan() is called, it does not count as an orphan
- (i.e. this is the 'optimistic' logic in mapper._is_orphan().)
- """
+ """Merge with dont_load does not trigger a 'delete-orphan' operation.
+
+ merge with dont_load sets attributes without using events. this means
+ the 'hasparent' flag is not propagated to the newly merged instance.
+ in fact this works out OK, because the '_state.parents' collection on
+ the newly merged instance is empty; since the mapper doesn't see an
+ active 'False' setting in this collection when _is_orphan() is called,
+ it does not count as an orphan (i.e. this is the 'optimistic' logic in
+ mapper._is_orphan().)
+ """
mapper(User, users, properties={
- 'addresses':relation(mapper(Address, addresses),backref='user', cascade="all, delete-orphan")
- })
+ 'addresses':relation(mapper(Address, addresses),
+ backref='user', cascade="all, delete-orphan")})
sess = create_session()
u = User()
u.id = 7
a1 = Address()
a1.email_address='foo@bar.com'
u.addresses.append(a1)
- sess.save(u)
+ sess.add(u)
sess.flush()
assert u.addresses[0].user is u
sa.orm.attributes.instance_state(a2))
sess2.flush()
sess2.clear()
- assert sess2.query(User).get(u2.id).addresses[0].email_address == 'somenewaddress'
- # this use case is not supported; this is with a pending Address on the pre-merged
- # object, and we currently dont support 'dirty' objects being merged with dont_load=True.
- # in this case, the empty '_state.parents' collection would be an issue,
- # since the optimistic flag is False in _is_orphan() for pending instances.
- # so if we start supporting 'dirty' with dont_load=True, this test will need to pass
+ eq_(sess2.query(User).get(u2.id).addresses[0].email_address,
+ 'somenewaddress')
+
+ # this use case is not supported; this is with a pending Address on
+ # the pre-merged object, and we currently dont support 'dirty' objects
+ # being merged with dont_load=True. in this case, the empty
+ # '_state.parents' collection would be an issue, since the optimistic
+ # flag is False in _is_orphan() for pending instances. so if we start
+ # supporting 'dirty' with dont_load=True, this test will need to pass
sess = create_session()
u = sess.query(User).get(7)
u.addresses.append(Address())
u2 = sess2.merge(u, dont_load=True)
assert False
- # if dont_load is changed to support dirty objects, this code needs to pass
+ # if dont_load is changed to support dirty objects, this code
+ # needs to pass
a2 = u2.addresses[0]
a2.email_address='somenewaddress'
assert not sa.orm.object_mapper(a2)._is_orphan(
sa.orm.attributes.instance_state(a2))
sess2.flush()
sess2.clear()
- assert sess2.query(User).get(u2.id).addresses[0].email_address == 'somenewaddress'
+ eq_(sess2.query(User).get(u2.id).addresses[0].email_address,
+ 'somenewaddress')
except sa.exc.InvalidRequestError, e:
assert "dont_load=True option does not support" in str(e)
sess = create_session()
u1 = User(username='jack', fullname='jack')
- sess.save(u1)
+ sess.add(u1)
sess.flush()
assert sess.get(User, 'jack') is u1
sess = create_session()
u1 = User(username='jack', fullname='jack')
- sess.save(u1)
+ sess.add(u1)
sess.flush()
assert sess.get(User, 'jack') is u1
u1 = User(username='jack', fullname='jack')
u1.addresses.append(Address(email='jack1'))
u1.addresses.append(Address(email='jack2'))
- sess.save(u1)
+ sess.add(u1)
sess.flush()
assert sess.get(Address, 'jack1') is u1.addresses[0]
u1 = User(username='jack', fullname='jack')
a1.user = u1
a2.user = u1
- sess.save(a1)
- sess.save(a2)
+ sess.add(a1)
+ sess.add(a2)
sess.flush()
u1.username = 'ed'
u1 = User(username='jack', fullname='jack')
a1.user = u1
a2.user = u1
- sess.save(a1)
- sess.save(a2)
+ sess.add(a1)
+ sess.add(a2)
sess.flush()
u1.username = 'ed'
u1.items.append(i1)
u1.items.append(i2)
i2.users.append(u2)
- sess.save(u1)
- sess.save(u2)
+ sess.add(u1)
+ sess.add(u2)
sess.flush()
r = sess.query(Item).all()
n1.children.append(Node(name='n11'))
n1.children.append(Node(name='n12'))
n1.children.append(Node(name='n13'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
n1.name = 'new n1'
u1 = User(username='jack', fullname='jack')
u1.addresses.append(Address(email='jack1'))
u1.addresses.append(Address(email='jack2'))
- sess.save(u1)
+ sess.add(u1)
sess.flush()
a1 = u1.addresses[0]
u1.addresses.append(Address(email_address='ed@bar.com'))
u2 = pickle.loads(pickle.dumps(u1))
- sess.save(u2)
+ sess.add(u2)
sess.flush()
sess.clear()
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
- sess.save(u1)
+ sess.add(u1)
sess.flush()
sess.clear()
u1 = sess.query(User).get(u1.id)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='ed@bar.com'))
- sess.save(u1)
+ sess.add(u1)
sess.flush()
sess.clear()
eu = EmailUser(name="user1", email_address='foo@bar.com')
sess = create_session()
- sess.save(eu)
+ sess.add(eu)
sess.flush()
sess.clear()
d1 = D(name='d1', b_row=b, c_row=c)
d2 = D(name='d2', b_row=b, c_row=c)
d3 = D(name='d3', b_row=b, c_row=c)
- session.save(a)
- session.save(b)
+ session.add(a)
+ session.add(b)
session.flush()
@testing.resolve_artifact_names
e6 = Employee(u'emp6', c2, 2, e5)
e7 = Employee(u'emp7', c2, 3, e5)
- [sess.save(x) for x in [c1,c2]]
+ sess.add_all((c1, c2))
sess.flush()
sess.clear()
a1 = A()
a1.bs.append(B())
sess = create_session()
- sess.save(a1)
+ sess.add(a1)
sess.flush()
sess.delete(a1)
a1 = A()
b1.a = a1
sess = create_session()
- sess.save(b1)
+ sess.add(b1)
sess.flush()
b1.a = None
try:
c1.id = 5
c1.a = None
sess = create_session()
- sess.save(c1)
+ sess.add(c1)
# test that no error is raised.
sess.flush()
a1 = A()
b1.a = a1
sess = create_session()
- sess.save(b1)
+ sess.add(b1)
sess.flush()
sess.delete(b1)
sess.flush()
b1 = B()
a1.bs.append(b1)
sess = create_session()
- sess.save(a1)
+ sess.add(a1)
sess.flush()
sess.delete(a1)
b1 = B()
a1.bs.append(b1)
sess = create_session()
- sess.save(a1)
- sess.save(b1)
+ sess.add(a1)
+ sess.add(b1)
sess.flush()
sess.delete(a1)
a1 = A()
b1.a = a1
sess = create_session()
- sess.save(b1)
- sess.save(a1)
+ sess.add(b1)
+ sess.add(a1)
sess.flush()
sess.delete(b1)
sess.delete(a1)
con.policyNum = "99"
con.policyEffDate = datetime.date.today()
con.type = "TESTER"
- session.save(con)
+ session.add(con)
for i in range(0, 10):
li = LineItem()
li.id = i
con.lineItems.append(li)
- session.save(li)
+ session.add(li)
session.flush()
session.clear()
newcon = session.query(Container).first()
a1.bs.append(c1)
sess = create_session()
try:
- sess.save(a1)
+ sess.add(a1)
assert False
except AssertionError, err:
eq_(str(err),
a1.bs.append(b1)
a1.bs.append(c1)
sess = create_session()
- sess.save(a1)
- sess.save(b1)
- sess.save(c1)
+ sess.add(a1)
+ sess.add(b1)
+ sess.add(c1)
self.assertRaisesMessage(sa.orm.exc.FlushError,
"Attempting to flush an item", sess.flush)
a1.bs.append(b1)
a1.bs.append(c1)
sess = create_session()
- sess.save(a1)
- sess.save(b1)
- sess.save(c1)
+ sess.add(a1)
+ sess.add(b1)
+ sess.add(c1)
self.assertRaisesMessage(sa.orm.exc.FlushError,
"Attempting to flush an item", sess.flush)
d1 = D()
d1.a = b1
sess = create_session()
- sess.save(b1)
- sess.save(d1)
+ sess.add(b1)
+ sess.add(d1)
self.assertRaisesMessage(sa.orm.exc.FlushError,
"Attempting to flush an item", sess.flush)
d1.a = b1
sess = create_session()
self.assertRaisesMessage(AssertionError,
- "doesn't handle objects of type", sess.save, d1)
+ "doesn't handle objects of type", sess.add, d1)
class TypedAssociationTable(_base.MappedTest):
c3.data='c1data'
c3.t2 = c2b
sess = create_session()
- sess.save(c1)
- sess.save(c3)
+ sess.add(c1)
+ sess.add(c3)
sess.flush()
sess.clear()
mapper(Bar, bars)
sess = create_session()
- sess.save(Foo(id=4))
- sess.save(Foo(id=9))
- sess.save(Bar(id=1, fid=2))
- sess.save(Bar(id=2, fid=3))
- sess.save(Bar(id=3, fid=6))
- sess.save(Bar(id=4, fid=7))
+ sess.add_all((Foo(id=4),
+ Foo(id=9),
+ Bar(id=1, fid=2),
+ Bar(id=2, fid=3),
+ Bar(id=3, fid=6),
+ Bar(id=4, fid=7)))
sess.flush()
sess = create_session()
mapper(User, users)
s = create_session(bind=c)
- s.save(User(name='first'))
+ s.add(User(name='first'))
s.flush()
c.execute("select * from users")
mapper(User, users)
s = create_session(bind=c)
- s.save(User(name='first'))
+ s.add(User(name='first'))
s.flush()
c.execute("select * from users")
s.close()
s = create_session(bind=c)
s.begin()
tran = s.transaction
- s.save(User(name='first'))
+ s.add(User(name='first'))
s.flush()
c.execute("select * from users")
u = User(name='two')
- s.save(u)
+ s.add(u)
s.flush()
u = User(name='third')
- s.save(u)
+ s.add(u)
s.flush()
assert s.transaction is tran
tran.close()
sess = create_session(autocommit=False, bind=conn1)
u = User(name='x')
- sess.save(u)
+ sess.add(u)
sess.flush()
assert conn1.execute("select count(1) from users").scalar() == 1
assert conn2.execute("select count(1) from users").scalar() == 0
sess = create_session(bind=conn1, autocommit=False, autoflush=True)
u = User()
u.name='ed'
- sess.save(u)
+ sess.add(u)
u2 = sess.query(User).filter_by(name='ed').one()
assert u2 is u
eq_(conn1.execute("select count(1) from users").scalar(), 1)
sess = create_session(autoflush=True, autocommit=False)
u = User(name='ed', addresses=[Address(email_address='foo')])
- sess.save(u)
+ sess.add(u)
eq_(sess.query(Address).filter(Address.user==u).one(),
Address(email_address='foo'))
sess = create_session(autocommit=False, autoflush=True)
u = User()
u.name='ed'
- sess.save(u)
+ sess.add(u)
u2 = sess.query(User).filter_by(name='ed').one()
assert u2 is u
assert sess.execute("select count(1) from users", mapper=User).scalar() == 1
sess = create_session(bind=conn1, autocommit=False, autoflush=True)
u = User()
u.name='ed'
- sess.save(u)
+ sess.add(u)
sess.commit()
assert conn1.execute("select count(1) from users").scalar() == 1
assert testing.db.connect().execute("select count(1) from users").scalar() == 1
sess = create_session(bind=conn, autocommit=False, autoflush=True)
sess.begin(subtransactions=True)
u = User(name='ed')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.commit() # commit does nothing
trans.rollback() # rolls back
trans = conn.begin()
sess = create_session(bind=conn, autocommit=False, autoflush=True)
u1 = User(name='u1')
- sess.save(u1)
+ sess.add(u1)
sess.flush()
sess.begin_nested()
u2 = User(name='u2')
- sess.save(u2)
+ sess.add(u2)
sess.flush()
sess.rollback()
sess = create_session(autocommit=False, autoflush=True)
sess.begin(subtransactions=True)
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.commit() # commit does nothing
sess.rollback() # rolls back
sess.begin()
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.begin_nested() # nested transaction
u2 = User(name='u2')
- sess.save(u2)
+ sess.add(u2)
sess.flush()
sess.rollback()
mapper(User, users)
sess = create_session(autocommit=False)
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.begin_nested() # nested transaction
u2 = User(name='u2')
- sess.save(u2)
+ sess.add(u2)
sess.flush()
sess.rollback()
sess.begin_nested()
u1 = User(name='u1')
- sess.save(u1)
+ sess.add(u1)
sess.flush()
sess.rollback()
u2 = User(name='u2')
- sess.save(u2)
+ sess.add(u2)
sess.commit()
sess.begin_nested()
u3 = User(name='u3')
- sess.save(u3)
+ sess.add(u3)
sess.commit() # commit the nested transaction
sess.rollback()
sess.begin_nested()
transaction = sess.begin(subtransactions=True)
- sess.save(User(name='u1'))
+ sess.add(User(name='u1'))
transaction.commit()
sess.commit()
t1 = sess.begin()
t2 = sess.begin_nested()
- sess.save(User(name='u2'))
+ sess.add(User(name='u2'))
t2.commit()
assert sess.transaction is t1
sess.begin_nested()
- sess.save(User(name='u1'))
+ sess.add(User(name='u1'))
sess.flush()
sess.close()
- sess.save(User(name='u2'))
+ sess.add(User(name='u2'))
sess.commit()
sess.close()
sess.begin()
sess.begin(subtransactions=True)
- sess.save(User(name='u1'))
+ sess.add(User(name='u1'))
sess.flush()
sess.rollback()
sess.begin()
transaction = sess.transaction
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
assert transaction._connection_for_bind(testing.db) is transaction._connection_for_bind(c) is c
sess = create_session(bind=c, autocommit=False)
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.close()
assert not c.in_transaction()
sess = create_session(bind=c, autocommit=False)
u = User(name='u2')
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.commit()
assert not c.in_transaction()
trans = c.begin()
sess = create_session(bind=c, autocommit=True)
u = User(name='u3')
- sess.save(u)
+ sess.add(u)
sess.flush()
assert c.in_transaction()
trans.commit()
self.assertRaisesMessage(sa.exc.InvalidRequestError, "is not persisted", s.update, user)
self.assertRaisesMessage(sa.exc.InvalidRequestError, "is not persisted", s.delete, user)
- s.save(user)
+ s.add(user)
s.flush()
user = s.query(User).one()
s.expunge(user)
# save user
u = User(name='fred')
- s.save(u)
+ s.add(u)
s.flush()
s.clear()
s = create_session()
mapper(User, users)
- s.save(User(name='ed'))
+ s.add(User(name='ed'))
s.flush()
assert not s.dirty
mapper(User, users)
# save user
- s.save(User(name='u1'))
+ s.add(User(name='u1'))
s.flush()
user = s.query(User).one()
user = None
mapper(User, users)
for o in [User(name='u%s' % x) for x in xrange(10)]:
- s.save(o)
+ s.add(o)
# o is still live after this loop...
self.assert_(len(s.identity_map) == 0)
self.assert_(s.prune() == 1)
self.assert_(len(s.identity_map) == 0)
- s.save(User(name='x'))
+ s.add(User(name='x'))
self.assert_(s.prune() == 0)
self.assert_(len(s.identity_map) == 0)
s.flush()
s = create_session()
u = User(name='u1')
- s.save(u)
+ s.add(u)
a = Address(email_address='u1@e')
u.addresses.append(a)
assert u in s
u = User(name='u1')
a = Address(email_address='u1@e')
a.user = u
- s.save(a)
+ s.add(a)
assert u not in s
assert a in s
s.flush()
mapper(User, users)
s = create_session()
u = User(name='u1')
- s.save(u)
+ s.add(u)
s.flush()
key = s.identity_key(instance=u)
eq_(key, (User, (u.id,), None))
log.append('after_begin')
sess = create_session(extension = MyExt())
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
assert log == ['before_flush', 'after_begin', 'after_flush', 'before_commit', 'after_commit', 'after_flush_postexec']
log = []
sess = create_session(autocommit=False, extension=MyExt())
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
assert log == ['before_flush', 'after_begin', 'after_flush', 'after_flush_postexec']
sess2 = create_session()
u1 = User(name='u1')
- sess1.save(u1)
+ sess1.add(u1)
- self.assertRaisesMessage(sa.exc.InvalidRequestError, "already attached to session", sess2.save, u1)
+ self.assertRaisesMessage(sa.exc.InvalidRequestError, "already attached to session", sess2.add, u1)
u2 = pickle.loads(pickle.dumps(u1))
- sess2.save(u2)
+ sess2.add(u2)
@testing.resolve_artifact_names
def test_duplicate_update(self):
sess = Session()
u1 = User(name='u1')
- sess.save(u1)
+ sess.add(u1)
sess.flush()
assert u1.id is not None
sess = create_session()
class Foo(object):
def __init__(self):
- sess.save(self)
+ sess.add(self)
class Bar(Foo):
def __init__(self):
- sess.save(self)
+ sess.add(self)
Foo.__init__(self)
mapper(Foo, users)
mapper(Bar, users)
sess = create_session(bind=self.engine)
self.engine.begin()
u = User(name='ed')
- sess.save(u)
+ sess.add(u)
sess.flush()
self.engine.commit()
t1 = T(value=True, name="t1")
t2 = T(value=False, name="t2")
t3 = T(value=True, name="t3")
- sess.save(t1)
- sess.save(t2)
- sess.save(t3)
+ sess.add_all((t1, t2, t3))
sess.flush()
o1.t2s.append(T2(data='some t2', id=1))
o1.t2s.append(T2(data='some other t2', id=2))
- sess.save(o1)
+ sess.add(o1)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some t1')]
T2(data='fourth t2', id=4),
])
sess.delete(o1)
- sess.save(o2)
+ sess.add(o2)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some other t1')]
o1.t3s.append(T3(data='some t3', id=1))
o1.t3s.append(T3(data='some other t3', id=2))
- sess.save(o1)
+ sess.add(o1)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some t1')]
T3(data='fourth t3', id=4),
])
sess.delete(o1)
- sess.save(o2)
+ sess.add(o2)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some other t1')]
o1 = T2(data='some t2', id=1)
o1.t1 = T1(data='some t1', id=1)
- sess.save(o1)
+ sess.add(o1)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some t1')]
o2 = T2(data='some other t2', id=1, t1=T1(data='some other t1', id=2))
sess.delete(o1)
sess.delete(o1.t1)
- sess.save(o2)
+ sess.add(o2)
sess.flush()
assert list(sess.execute(t1.select(), mapper=T1)) == [(2, 'some other t1')]
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
