- Removed an internal join cache which could potentially leak
memory when issuing query.join() repeatedly to ad-hoc
selectables.
-
+
+ - The "clear()", "save()", "update()", "save_or_update()"
+ Session methods have been deprecated, replaced by
+ "expunge_all()" and "add()". "expunge_all()" has also
+ been added to ScopedSession.
+
- Modernized the "no mapped table" exception and added a more
explicit __table__/__tablename__ exception to declarative.
>>> db.loans.count()
1
>>> _ = db.loans.insert(book_id=1, user_name='Bhargan Basepair')
- >>> db.clear()
+ >>> db.expunge_all()
>>> db.flush()
>>> db.loans.count()
1
Session.flush()
def clear(self):
- Session.clear()
+ Session.expunge_all()
+
+ def expunge_all(self):
+ Session.expunge_all()
def map(self, selectable, **kwargs):
try:
public_methods = (
'__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
'clear', 'close', 'commit', 'connection', 'delete', 'execute', 'expire',
- 'expire_all', 'expunge', 'flush', 'get_bind', 'is_modified',
+ 'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind', 'is_modified',
'merge', 'query', 'refresh', 'rollback', 'save',
'save_or_update', 'scalar', 'update')
not use any connection resources until they are first needed.
"""
- self.clear()
+ self.expunge_all()
if self.transaction is not None:
for transaction in self.transaction._iterate_parents():
transaction.close()
+ @classmethod
def close_all(cls):
"""Close *all* sessions in memory."""
for sess in _sessions.values():
sess.close()
- close_all = classmethod(close_all)
def expunge_all(self):
"""Remove all object instances from this ``Session``.
self.identity_map = self._identity_cls()
self._new = {}
self._deleted = {}
- clear = expunge_all
- # TODO: deprecate
- #clear = util.deprecated()(expunge_all)
+ clear = util.deprecated("Use session.expunge_all()")(expunge_all)
# TODO: need much more test coverage for bind_mapper() and similar !
# TODO: + crystalize + document resolution order vis. bind_mapper/bind_table
self.identity_map.discard(state)
self._deleted.pop(state, None)
- @util.pending_deprecation('0.5.x', "Use session.add()")
+ @util.deprecated("Use session.add()")
def save(self, instance):
"""Add a transient (unsaved) instance to this ``Session``.
state = _state_for_unsaved_instance(instance, create=True)
self._save_impl(state)
- @util.pending_deprecation('0.5.x', "Use session.add()")
+ @util.deprecated("Use session.add()")
def update(self, instance):
"""Bring a detached (saved) instance into this ``Session``.
self._cascade_save_or_update(state)
save_or_update = (
- util.pending_deprecation('0.5.x', "Use session.add()")(add))
+ util.deprecated("Use session.add()")(add))
def _cascade_save_or_update(self, state):
for state, mapper in _cascade_unknown_state_iterator('save-update', state, halt_on=lambda c:c in self):
merged = mapper.class_manager.new_instance()
merged_state = attributes.instance_state(merged)
new_instance = True
- self.save(merged)
+ self.add(merged)
_recursive[instance] = merged
def roundtrip(self, obj):
if obj not in self.session:
- self.session.save(obj)
+ self.session.add(obj)
self.session.flush()
id, type_ = obj.id, type(obj)
- self.session.clear()
+ self.session.expunge_all()
return self.session.query(type_).get(id)
def _test_sequence_ops(self):
def roundtrip(obj):
if obj not in session:
- session.save(obj)
+ session.add(obj)
session.flush()
id, type_ = obj.id, type(obj)
- session.clear()
+ session.expunge_all()
return session.query(type_).get(id)
p = Parent('p')
self.metadata.drop_all()
def roundtrip(self, obj):
- self.session.save(obj)
+ self.session.add(obj)
self.session.flush()
id, type_ = obj.id, type(obj)
- self.session.clear()
+ self.session.expunge_all()
return self.session.query(type_).get(id)
def test_lazy_list(self):
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
u1 = User(name='ed', addresses=[Address(email='abc'), Address(email='def'), Address(email='xyz')])
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(User).filter(User.name == 'ed').one(),
User(name='ed', addresses=[Address(email='xyz'), Address(email='def'), Address(email='abc')])
)
u1 = User(name='ed', addresses=[Address(email='abc'), Address(email='xyz'), Address(email='def')])
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(User).filter(User.name == 'ed').one(),
User(name='ed', addresses=[Address(email='abc'), Address(email='def'), Address(email='xyz')])
)
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).options(eagerload(User.addresses)).all(), [User(name='u1', addresses=[
Address(email='one'),
Address(email='two'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).filter(User.name == 'u1').one()
a = u.addresses
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(),
[User(name='u1', address_count=2, addresses=[
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(),
[User(name='u1', a='a', b='b')])
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(),
[User(name='u1', adr_count=2, addresses=[
sess = create_session()
sess.add(User(name='u1'))
sess.flush()
- sess.clear()
+ sess.expunge_all()
u1 = sess.query(User).filter(User.name == 'u1').one()
assert 'name' not in u1.__dict__
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(),
[User(name='u1', address_count=2, addresses=[
sess.add(c1)
sess.add(c2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_((sess.query(Company).
filter(Company.employees.of_type(Engineer).
# assert that the "id" column is available without a second load.
# this would be the symptom of the previous step not being correct.
- sess.clear()
+ sess.expunge_all()
def go():
assert sess.query(Manager).filter(Manager.name=='dogbert').one().id
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
assert sess.query(Person).filter(Manager.name=='dogbert').one().id
self.assert_sql_count(testing.db, go, 1)
sess.add(c1)
sess.add(c2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_((sess.query(Person).
filter(Engineer.primary_language == 'cobol').first()),
sess.add(c1)
sess.add(c2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_((sess.query(Person).
filter(Engineer.primary_language == 'cobol').first()),
sess.add(c1)
sess.add(c2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_((sess.query(Person).with_polymorphic(Engineer).
filter(Engineer.primary_language == 'cobol').first()),
e3 =Engineer(name="vlad", primary_language=cobol)
sess.add_all([e1, e2, e3])
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_((sess.query(Person).
filter(Engineer.primary_language.has(Language.name=='cobol')).first()),
sess.add_all([e1, e2, m1, e3])
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(
sess.query(Person).order_by(Person.name).all(),
[
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', addresses=[
Address(email='one'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(nom='u1', addresses=[
Address(email='one'),
sess = create_session()
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).all(), [User(name='u1', handles=[
IMHandle(network='blabber', handle='foo'),
eq_(u1.uc_name, 'SOMEUSER', u1.uc_name)
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
rt = sess.query(User).filter(User.uc_name == 'SOMEUSER').one()
eq_(rt, u1)
- sess.clear()
+ sess.expunge_all()
rt = sess.query(User).filter(User.uc_name.startswith('SOMEUSE')).one()
eq_(rt, u1)
self.assert_(s1.bullets[3].position == 4)
session = create_session()
- session.save(s1)
+ session.add(s1)
session.flush()
id = s1.id
- session.clear()
+ session.expunge_all()
del s1
srt = session.query(Slide).get(id)
s1.bullets._reorder()
self.assert_(s1.bullets[4].position == 5)
session = create_session()
- session.save(s1)
+ session.add(s1)
session.flush()
id = s1.id
- session.clear()
+ session.expunge_all()
del s1
srt = session.query(Slide).get(id)
srt.bullets._raw_append(Bullet('raw2'))
srt.bullets[-1].position = 6
session.flush()
- session.clear()
+ session.expunge_all()
srt = session.query(Slide).get(id)
titles = ['s1/b1','s1/b2','s1/b100','s1/b4', 'raw', 'raw2']
self.assert_(s1.bullets[5].position == 5)
session = create_session()
- session.save(s1)
+ session.add(s1)
session.flush()
id = s1.id
- session.clear()
+ session.expunge_all()
del s1
srt = session.query(Slide).get(id)
self.assert_(s1.bullets[li] == b[bi])
session = create_session()
- session.save(s1)
+ session.add(s1)
session.flush()
id = s1.id
- session.clear()
+ session.expunge_all()
del s1
srt = session.query(Slide).get(id)
self.assert_(s1.bullets[2].position == 2)
session = create_session()
- session.save(s1)
+ session.add(s1)
session.flush()
new_bullet = Bullet('new 2')
id = s1.id
session.flush()
- session.clear()
+ session.expunge_all()
srt = session.query(Slide).get(id)
sess.add_all((item1, item2))
sess.flush()
saved = repr([item1, item2])
- sess.clear()
+ sess.expunge_all()
l = sess.query(Item).all()
loaded = repr(l)
eq_(saved, loaded)
item1.keywords.append(KeywordAssociation(red_keyword, 'new_red_assoc'))
sess.flush()
saved = repr([item1])
- sess.clear()
+ sess.expunge_all()
l = sess.query(Item).all()
loaded = repr(l)
eq_(saved, loaded)
sess.flush()
saved = repr([item1, item2])
- sess.clear()
+ sess.expunge_all()
l = sess.query(Item).all()
loaded = repr(l)
eq_(saved, loaded)
session = create_session()
session.add(p)
session.flush()
- session.clear()
+ session.expunge_all()
obj = session.query(Left).filter_by(data='l1').one()
d2.comments = [Comment('uid2', 'comment')]
sess.add_all((d, d2))
sess.flush()
- sess.clear()
+ sess.expunge_all()
# this eager load sets up an AliasedClauses for the "comment"
# relationship, then stores it in clauses_by_lead_mapper[mapper for
# Derived]
d = sess.query(Derived).get('uid1')
- sess.clear()
+ sess.expunge_all()
assert len([c for c in d.comments]) == 1
# this eager load sets up an AliasedClauses for the "comment"
# for DerivedII]. the bug was that the previous AliasedClause create
# prevented this population from occurring.
d2 = sess.query(DerivedII).get('uid2')
- sess.clear()
+ sess.expunge_all()
# object is not in the session; therefore the lazy load cant trigger
# here, eager load had to succeed
sess = create_session()
sess.add(d)
sess.flush()
- sess.clear()
+ sess.expunge_all()
x = sess.query(Design).get(1)
x.inheritedParts
company_id = c1.company_id
invoice_id = i1.invoice_id
- session.clear()
+ session.expunge_all()
c = session.query(Company).get(company_id)
- session.clear()
+ session.expunge_all()
i = session.query(Invoice).get(invoice_id)
eq_(c, i.company)
company_id = c1.company_id
- session.clear()
+ session.expunge_all()
a = session.query(Company).get(company_id)
session.flush()
invoice_id = i1.invoice_id
- session.clear()
+ session.expunge_all()
c = session.query(Company).get(company_id)
- session.clear()
+ session.expunge_all()
i = session.query(Invoice).get(invoice_id)
eq_(c, i.company)
session.add(acc1)
session.flush()
- session.clear()
+ session.expunge_all()
def go():
# load just the first Account. eager loading will actually load
sess = create_session(bind=bind)
assert sess.bind is bind
f = Foo()
- sess.save(f)
+ sess.add(f)
sess.flush()
assert sess.query(Foo).get(f.id) is f
finally:
Order(description='someotherorder')])
sess.add(u)
sess.flush()
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(u.id)
eq_(u, User(name='jack',
u.orders=[Order(description="order 3"), Order(description="order 4")]
sess.flush()
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(u.id)
eq_(u, User(name='jack',
Address(email_address="address2")])
sess.add(u)
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert addresses.count().scalar() == 2
assert users.count().scalar() == 1
u2 = User(name='newuser', orders=[o])
sess.add(u2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert users.count().scalar() == 1
assert orders.count().scalar() == 1
eq_(sess.query(User).all(),
assert u1 not in sess
assert o1 in sess
- sess.clear()
+ sess.expunge_all()
o1 = Order()
u1 = User(orders=[o1])
assert o1 not in sess
assert u1 in sess
- sess.clear()
+ sess.expunge_all()
u1 = User()
o1 = Order()
assert i1 in sess
assert k1 not in sess
- sess.clear()
+ sess.expunge_all()
i1 = Item()
k1 = Keyword()
jack = sess.query(User).filter_by(name="jack").one()
p = jack.pref
e = jack.pref.extra[0]
- sess.clear()
+ sess.expunge_all()
jack.pref = None
- sess.update(jack)
- sess.update(p)
- sess.update(e)
+ sess.add(jack)
+ sess.add(p)
+ sess.add(e)
assert p in sess
assert e in sess
sess.flush()
u.addresses.remove(a1)
assert a1 in s
s.flush()
- s.clear()
+ s.expunge_all()
eq_(s.query(Address).all(), [Address(email_address='ad1')])
b1 = Business(description='business1', address=Address(street='address2'))
session.add_all((h1,b1))
session.flush()
- session.clear()
+ session.expunge_all()
eq_(session.query(Home).get(h1.id), Home(description='home1', address=Address(street='address1')))
eq_(session.query(Business).get(b1.id), Business(description='business1', address=Address(street='address2')))
sess.add(a1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(A).get(a1.id),
A(name='a1', bs=[B(name='b1'), B(name='b2'), B(name='b3')]))
a1.bs[1].foo='b3modified'
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(A).get(a1.id),
A(name='a1', bs=[B(name='b1'), B(name='b2'), B(name='b3')]))
session.add(p)
session.flush()
pid = p.id
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
Child('foo', 'newvalue'))
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
self.assert_(len(list(collections.collection_adapter(p.children))) == 2)
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
self.assert_(len(list(collections.collection_adapter(p.children))) == 2)
self.assert_(len(list(collections.collection_adapter(p.children))) == 1)
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
self.assert_(len(list(collections.collection_adapter(p.children))) == 1)
del p.children['bar']
self.assert_(len(list(collections.collection_adapter(p.children))) == 0)
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
self.assert_(len(list(collections.collection_adapter(p.children))) == 0)
session.add(p)
session.flush()
pid = p.id
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
Child('foo', '1', 'newvalue'))
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Parent).get(pid)
sess = create_session()
sess.add(f)
sess.flush()
- sess.clear()
+ sess.expunge_all()
f = sess.query(Foo).get(f.col1)
assert len(list(f.bars)) == 2
f.bars.clear()
sess = create_session()
sess.add(f)
sess.flush()
- sess.clear()
+ sess.expunge_all()
f = sess.query(Foo).get(f.col1)
assert len(list(f.bars)) == 2
f.bars.clear()
sess = create_session()
sess.add(f)
sess.flush()
- sess.clear()
+ sess.expunge_all()
f = sess.query(Foo).get(f.col1)
assert len(list(f.bars)) == 2
col.append_with_event(Bar('b'))
f.bars['a'] = Bar('a')
sess.flush()
- sess.clear()
+ sess.expunge_all()
f = sess.query(Foo).get(f.col1)
assert len(list(f.bars)) == 2
sess = create_session()
sess.add(p1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
p2 = sess.query(Parent).get(p1.col1)
o = list(p2.children)
sess = create_session()
sess.add(c1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
c1 = sess.query(C1).get(c1.c1)
c2 = C1()
c2.parent = c1
s = create_session()
s.add(t1)
s.flush()
- s.clear()
+ s.expunge_all()
t = s.query(TT).filter_by(id=t1.id).one()
eq_(t.children[0].parent_uuid, t1.uuid)
t1.children.append(t2)
s.add(t1)
s.flush()
- s.clear()
+ s.expunge_all()
t = s.query(TT).filter_by(id=t2.id).one()
eq_(t.uuid, t2.uuid)
c1.child1_data = "qwerty"
session.add(c1)
session.flush()
- session.clear()
+ session.expunge_all()
c1 = session.query(Child1).filter_by(child1_data="qwerty").one()
c2 = Child2()
# to fire off anyway
session.add(f2)
session.flush()
- session.clear()
+ session.expunge_all()
f1 = session.query(A).get(f1.id)
f2 = session.query(A).get(f2.id)
from sqlalchemy.orm import attributes
self.assertEquals(attributes.get_history(attributes.instance_state(u1), 'addresses'), ([], [Address(email_address='lala@hoho.com')], []))
- sess.clear()
+ sess.expunge_all()
# test the test fixture a little bit
assert User(name='jack', addresses=[Address(email_address='wrong')]) != sess.query(User).first()
sess.delete(u.addresses[3])
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(u.id)
sess.delete(u)
sess.delete(u.addresses[3])
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(u.id)
sess.delete(u)
address.user = user
session.add(user)
session.flush()
- session.clear()
+ session.expunge_all()
def query1():
session = create_session(testing.db)
sess.query(User).all()
m.add_property("addresses", relation(mapper(Address, addresses)))
- sess.clear()
+ sess.expunge_all()
def go():
eq_(
[User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])],
sess.query(Address).filter(Address.id.in_([1, 4, 5])),
sess.query(Address).filter(Address.id.in_([1, 4, 5])).limit(3)
]:
- sess.clear()
+ sess.expunge_all()
eq_(q.all(),
[Address(id=1, user=User(id=7)),
Address(id=4, user=User(id=8)),
sess.query(User).filter(User.id==7),
sess.query(User).filter(User.id==7).limit(1)
]:
- sess.clear()
+ sess.expunge_all()
eq_(q.all(),
[User(id=7, addresses=[Address(id=1)])]
)
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(7)
def go():
assert u.addresses[0].user_id==7
'dingalings':relation(Dingaling, lazy=False)})
mapper(Dingaling, dingalings, properties={
'address_id':deferred(dingalings.c.address_id)})
- sess.clear()
+ sess.expunge_all()
def go():
u = sess.query(User).get(8)
eq_(User(id=8,
self.assertEquals(o1.address, None)
self.assert_sql_count(testing.db, go, 2)
- sess.clear()
+ sess.expunge_all()
def go():
o1 = sess.query(Order).filter(Order.id==5).one()
self.assertEquals(o1.address, None)
eq_(ret, self._assert_result())
self.assert_sql_count(testing.db, go, 6)
- sess.clear()
+ sess.expunge_all()
def go():
ret = sess.query(User, oalias).options(eagerload('addresses'), eagerload(oalias.items)).join(('orders', oalias)).order_by(User.id, oalias.id).all()
eq_(ret, self._assert_result())
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
def go():
d = sess.query(Node).filter_by(data='n1').first()
assert Node(data='n1', children=[
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# eager load with join depth 1. when eager load of 'n1' hits the
# children of 'n12', no columns are present, eager loader degrades to
n1 = Node(data='n1')
n1.append(Node(data='n11'))
n1.append(Node(data='n12'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(
)
self.assert_sql_count(testing.db, go, 4)
- sess.clear()
+ sess.expunge_all()
def go():
assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data')).order_by(Node.id).first()
self.assert_sql_count(testing.db, go, 3)
- sess.clear()
+ sess.expunge_all()
def go():
assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data'), undefer('children.data')).first()
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
def go():
d = sess.query(Node).filter_by(data='n1').options(eagerload('children.children')).first()
assert Node(data='n1', children=[
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
def go():
d = sess.query(Node).filter_by(data='n1').first()
assert Node(data='n1', children=[
w1 = Widget(name=u'w1')
w2 = Widget(name=u'w2')
w1.children.append(w2)
- sess.save(w1)
+ sess.add(w1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert [Widget(name='w1', children=[Widget(name='w2')])] == sess.query(Widget).filter(Widget.name==u'w1').all()
})
session = create_session()
- session.save(User(name='joe', tags=[Tag(score1=5.0, score2=3.0), Tag(score1=55.0, score2=1.0)]))
- session.save(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), Tag(score1=50.0, score2=1.0), Tag(score1=15.0, score2=2.0)]))
+ session.add(User(name='joe', tags=[Tag(score1=5.0, score2=3.0), Tag(score1=55.0, score2=1.0)]))
+ session.add(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), Tag(score1=50.0, score2=1.0), Tag(score1=15.0, score2=2.0)]))
session.flush()
- session.clear()
+ session.expunge_all()
for user in session.query(User).all():
eq_(user.query_score, user.prop_score)
mapper(User, users)
s = create_session()
u = s.query(User).get(7)
- s.clear()
+ s.expunge_all()
self.assertRaisesMessage(sa.exc.InvalidRequestError, r"is not persistent within this Session", s.expire, u)
u.name = 'somenewname'
self.assert_sql_count(testing.db, go, 0)
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(User).get(7).name == 'somenewname'
@testing.resolve_artifact_names
sa.orm.clear_mappers()
mapper(Order, orders)
- sess.clear()
+ sess.expunge_all()
# same tests, using deferred at the options level
o = sess.query(Order).options(sa.orm.defer('description')).get(3)
mapper(User, users)
s = create_session()
u = s.query(User).get(7)
- s.clear()
+ s.expunge_all()
self.assertRaisesMessage(sa.exc.InvalidRequestError, r"is not persistent within this Session", lambda: s.refresh(u))
@testing.resolve_artifact_names
s.add(u)
s.flush()
- s.clear()
+ s.expunge_all()
u = s.query(User).filter(User.name=='Justin').one()
s.expire(u)
#print "APPENDING", parent.__class__.__name__ , "TO", child.__class__.__name__
- sess.save(parent_obj)
+ sess.add(parent_obj)
parent_obj.collection.append(child_obj)
if direction == ONETOMANY:
child2 = child_class('child2')
parent_obj.collection.append(child2)
- sess.save(child2)
+ sess.add(child2)
elif direction == MANYTOONE:
parent2 = parent_class('parent2')
parent2.collection.append(child_obj)
- sess.save(parent2)
- sess.save(somea)
- sess.save(someb)
- sess.save(somec)
+ sess.add(parent2)
+ sess.add(somea)
+ sess.add(someb)
+ sess.add(somec)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# assert result via direct get() of parent object
result = sess.query(parent_class).get(parent_obj.id)
assert result2.id == parent2.id
assert result2.collection[0].id == child_obj.id
- sess.clear()
+ sess.expunge_all()
# assert result via polymorphic load of parent object
result = sess.query(A).filter_by(id=parent_obj.id).one()
sess = create_session()
for x in (a1, b1, b2, b3, c1, c2, c3):
- sess.save(x)
+ sess.add(x)
sess.flush()
- sess.clear()
+ sess.expunge_all()
#for obj in sess.query(A).all():
# print obj
b1 = Blub("blub #1")
b2 = Blub("blub #2")
f = Foo("foo #1")
- sess.save(b1)
- sess.save(b2)
- sess.save(f)
+ sess.add(b1)
+ sess.add(b2)
+ sess.add(f)
b1.parent_foo = f
b2.parent_foo = f
sess.flush()
compare = ','.join([repr(b1), repr(b2), repr(b1.parent_foo), repr(b2.parent_foo)])
- sess.clear()
+ sess.expunge_all()
l = sess.query(Blub).all()
result = ','.join([repr(l[0]), repr(l[1]), repr(l[0].parent_foo), repr(l[1].parent_foo)])
print compare
mapper(Bar, inherits=Foo, polymorphic_identity=0)
sess = create_session()
f1 = Bar()
- sess.save(f1)
+ sess.add(f1)
sess.flush()
assert f1.type == 0
- sess.clear()
+ sess.expunge_all()
assert isinstance(sess.query(Foo).one(), Bar)
class PolymorphicSynonymTest(ORMTest):
sess = create_session()
at1 = T1(info='at1')
at2 = T2(info='at2', data='t2 data')
- sess.save(at1)
- sess.save(at2)
+ sess.add(at1)
+ sess.add(at2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(T2).filter(T2.info=='at2').one(), at2)
self.assertEquals(at2.info, "THE INFO IS:at2")
t4_1 = T4(data='t4')
t3_1.t4s.append(t4_1)
- sess.save(t1_1)
+ sess.add(t1_1)
assert t4_1 in sess.new
f = Foo()
b = Bar()
bl = Blub()
- sess.save(f)
- sess.save(b)
- sess.save(bl)
+ sess.add(f)
+ sess.add(b)
+ sess.add(bl)
sess.flush()
if polymorphic:
admin_mapper = mapper(Admin, admins, inherits=user_mapper)
sess = create_session()
adminrole = Role()
- sess.save(adminrole)
+ sess.add(adminrole)
sess.flush()
# create an Admin, and append a Role. the dependency processors
a = Admin()
a.roles.append(adminrole)
a.password = 'admin'
- sess.save(a)
+ sess.add(a)
sess.flush()
assert user_roles.count().scalar() == 1
adminrole = Role('admin')
sess = create_session()
- sess.save(adminrole)
+ sess.add(adminrole)
sess.flush()
# create admin user
a = Admin(email='tim', password='admin')
a.roles.append(adminrole)
- sess.save(a)
+ sess.add(a)
sess.flush()
a.password = 'sadmin'
b1 = Base(value='b1')
s1 = Sub(value='sub1', subdata='some subdata')
- sess.save(b1)
- sess.save(s1)
+ sess.add(b1)
+ sess.add(s1)
sess.flush()
b1 = Base(value='b1')
s1 = Sub(value='sub1', subdata='some subdata')
s2 = Sub(value='sub2', subdata='some other subdata')
- sess.save(b1)
- sess.save(s1)
- sess.save(s2)
+ sess.add(b1)
+ sess.add(s1)
+ sess.add(s2)
sess.flush()
session = create_session()
a = A(data1='a1')
- session.save(a)
+ session.add(a)
b = B(data1='b1', data2='b2')
- session.save(b)
+ session.add(b)
c = C(data1='c1', data2='c2', data3='c3')
- session.save(c)
+ session.add(c)
session.flush()
- session.clear()
+ session.expunge_all()
assert len(session.query(A).all()) == 3
assert len(session.query(B).all()) == 2
sess.add_all([s1, b1])
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(Base).get(b1.base_id).subdata == "this is base"
assert sess.query(Sub).get(s1.base_id).subdata == "this is sub"
sess.add_all([s1, b1])
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(Base).get(b1.base_id).data == "this is base"
assert sess.query(Sub).get(s1.base_id).data == "this is base"
s1 = Sub()
s1.data = 's1data'
s1.sub = 's1sub'
- sess.save(s1)
+ sess.add(s1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# load s1 via Base. s1.id won't populate since it's relative to
# the "sub" table. The optimized load kicks in and tries to
concrete=True, polymorphic_identity='engineer')
session = create_session()
- session.save(Manager('Tom', 'knows how to manage things'))
- session.save(Engineer('Kurt', 'knows how to hack'))
+ session.add(Manager('Tom', 'knows how to manage things'))
+ session.add(Engineer('Kurt', 'knows how to hack'))
session.flush()
- session.clear()
+ session.expunge_all()
assert set([repr(x) for x in session.query(Employee)]) == set(["Engineer Kurt knows how to hack", "Manager Tom knows how to manage things"])
assert set([repr(x) for x in session.query(Manager)]) == set(["Manager Tom knows how to manage things"])
self.assertEquals(hacker.nickname, "Badass")
self.assert_sql_count(testing.db, go, 0)
- session.clear()
+ session.expunge_all()
assert repr(session.query(Employee).filter(Employee.name=='Tom').one()) == "Manager Tom knows how to manage things"
assert repr(session.query(Manager).filter(Manager.name=='Tom').one()) == "Manager Tom knows how to manage things"
self.assertEquals(hacker.nickname, "Badass")
self.assert_sql_count(testing.db, go, 0)
- session.clear()
+ session.expunge_all()
# check that we aren't getting a cartesian product in the raw SQL.
# this requires that Engineer's polymorphic discriminator is not rendered
c = Company()
c.employees.append(Manager('Tom', 'knows how to manage things'))
c.employees.append(Engineer('Kurt', 'knows how to hack'))
- session.save(c)
+ session.add(c)
session.flush()
- session.clear()
+ session.expunge_all()
def go():
c2 = session.query(Company).get(c.id)
assert set([repr(x) for x in c2.employees]) == set(["Engineer Kurt knows how to hack", "Manager Tom knows how to manage things"])
self.assert_sql_count(testing.db, go, 2)
- session.clear()
+ session.expunge_all()
def go():
c2 = session.query(Company).options(eagerload(Company.employees)).get(c.id)
assert set([repr(x) for x in c2.employees]) == set(["Engineer Kurt knows how to hack", "Manager Tom knows how to manage things"])
def _set_name(self, name):
session = create_session()
s = session.query(LocationName).filter(LocationName.name==name).first()
- session.clear()
+ session.expunge_all()
if s is not None:
self._name = s
page = ClassifiedPage(magazine=magazine,page_no=1)
page2 = MagazinePage(magazine=magazine,page_no=2)
page3 = ClassifiedPage(magazine=magazine,page_no=3)
- session.save(pub)
+ session.add(pub)
session.flush()
print [x for x in session]
- session.clear()
+ session.expunge_all()
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Publication).filter(Publication.name=="Test").one()
print p.issues[0].locations[0].magazine.pages
g = Group(name="group1")
g.users.append(User(name="user1", password="pw", email="foo@bar.com", login_id="lg1"))
sess = create_session()
- sess.save(g)
+ sess.add(g)
sess.flush()
# TODO: put an assertion
print class_mapper(Bar).primary_key
b = Bar('somedata')
sess = create_session()
- sess.save(b)
+ sess.add(b)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# test that "bar.bid" does not need to be referenced in a get
# (ticket 185)
sess = create_session()
b = Bar('barfoo')
- sess.save(b)
+ sess.add(b)
sess.flush()
f1 = Foo('subfoo1')
b.foos.append(f2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
l = sess.query(Bar).all()
print l[0]
sess = create_session()
b = Bar('bar #1')
- sess.save(b)
+ sess.add(b)
b.foos.append(Foo("foo #1"))
b.foos.append(Foo("foo #2"))
sess.flush()
compare = repr(b) + repr(sorted([repr(o) for o in b.foos]))
- sess.clear()
+ sess.expunge_all()
l = sess.query(Bar).all()
print repr(l[0]) + repr(l[0].foos)
found = repr(l[0]) + repr(sorted([repr(o) for o in l[0].foos]))
b2 = Bar("bar #2")
bl1 = Blub("blub #1")
for o in (f1, b1, b2, bl1):
- sess.save(o)
+ sess.add(o)
bl1.foos.append(f1)
bl1.bars.append(b2)
sess.flush()
compare = repr(bl1)
blubid = bl1.id
- sess.clear()
+ sess.expunge_all()
l = sess.query(Blub).all()
print l
self.assert_(repr(l[0]) == compare)
- sess.clear()
+ sess.expunge_all()
x = sess.query(Blub).filter_by(id=blubid).one()
print x
self.assert_(repr(x) == compare)
obj = newobj
# save to DB
- sess.save(t)
+ sess.add(t)
sess.flush()
# string version of the saved list
# clear and query forwards
- sess.clear()
+ sess.expunge_all()
node = sess.query(Table1).filter(Table1.id==t.id).first()
assertlist = []
while (node):
forwards = repr(assertlist)
# clear and query backwards
- sess.clear()
+ sess.expunge_all()
node = sess.query(Table1).filter(Table1.id==obj.id).first()
assertlist = []
while (node):
c.employees.append(Person(status='HHH', name='joesmith'))
c.employees.append(Engineer(status='CGG', engineer_name='engineer2', primary_language='python', name='wally'))
c.employees.append(Manager(status='ABA', manager_name='manager2', name='jsmith'))
- session.save(c)
+ session.add(c)
session.flush()
- session.clear()
+ session.expunge_all()
self.assertEquals(session.query(Company).get(c.company_id), c)
class RelationToSubclassTest(PolymorphTest):
c = Company(name='company1')
c.managers.append(Manager(status='AAB', manager_name='manager1', name='pointy haired boss'))
- sess.save(c)
+ sess.add(c)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company).filter_by(company_id=c.company_id).one(), c)
assert c.managers[0].company is c
session = create_session()
c = Company(name='company1')
c.employees = employees
- session.save(c)
+ session.add(c)
session.flush()
- session.clear()
+ session.expunge_all()
self.assertEquals(session.query(Person).get(dilbert.person_id), dilbert)
- session.clear()
+ session.expunge_all()
self.assertEquals(session.query(Person).filter(Person.person_id==dilbert.person_id).one(), dilbert)
- session.clear()
+ session.expunge_all()
def go():
cc = session.query(Company).get(c.company_id)
dilbert.engineer_name = 'hes dibert!'
session.flush()
- session.clear()
+ session.expunge_all()
def go():
session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
self.assert_sql_count(testing.db, go, 1)
- session.clear()
+ session.expunge_all()
dilbert = session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
def go():
# assert that only primary table is queried for already-present-in-session
# test standalone orphans
daboss = Boss(status='BBB', manager_name='boss', golf_swing='fore', **{person_attribute_name:'daboss'})
- session.save(daboss)
+ session.add(daboss)
self.assertRaises(orm_exc.FlushError, session.flush)
c = session.query(Company).first()
daboss.company = c
manager_list = [e for e in c.employees if isinstance(e, Manager)]
session.flush()
- session.clear()
+ session.expunge_all()
self.assertEquals(session.query(Manager).order_by(Manager.person_id).all(), manager_list)
c = session.query(Company).first()
p = Person(name='some person')
m = Manager(name='some manager')
p.manager = m
- session.save(p)
+ session.add(p)
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Person).get(p.person_id)
m = session.query(Manager).get(m.person_id)
p = Person(name='some person')
m = Manager(name='some manager')
m.employee = p
- session.save(m)
+ session.add(m)
session.flush()
- session.clear()
+ session.expunge_all()
p = session.query(Person).get(p.person_id)
m = session.query(Manager).get(m.person_id)
m.colleague = p
if usedata:
m.data = Data('ms data')
- sess.save(m)
+ sess.add(m)
sess.flush()
- sess.clear()
+ sess.expunge_all()
p = sess.query(Person).get(p.person_id)
m = sess.query(Manager).get(m.person_id)
print p
p.data = Data('ps data')
m.data = Data('ms data')
- sess.save(m)
- sess.save(p)
+ sess.add(m)
+ sess.add(p)
sess.flush()
- sess.clear()
+ sess.expunge_all()
p = sess.query(Person).get(p.person_id)
p2 = sess.query(Person).get(p2.person_id)
p3 = sess.query(Person).get(p3.person_id)
# creating 5 managers named from M1 to E5
for i in range(1,5):
- session.save(Manager(name="M%d" % i,longer_status="YYYYYYYYY"))
+ session.add(Manager(name="M%d" % i,longer_status="YYYYYYYYY"))
# creating 5 engineers named from E1 to E5
for i in range(1,5):
- session.save(Engineer(name="E%d" % i,status="X"))
+ session.add(Engineer(name="E%d" % i,status="X"))
session.flush()
manager3 = session.query(Manager).filter(Manager.name=="M3").first()
car1 = Car(employee=engineer4)
- session.save(car1)
+ session.add(car1)
car2 = Car(employee=manager3)
- session.save(car2)
+ session.add(car2)
session.flush()
- session.clear()
+ session.expunge_all()
def go():
testcar = session.query(Car).options(eagerload('employee')).get(car1.car_id)
assert str(usingGet) == "Engineer E4, status X"
assert str(usingProperty) == "Engineer E4, status X"
- session.clear()
+ session.expunge_all()
print "-----------------------------------------------------------------"
# and now for the lightning round, eager !
assert str(testcar.employee) == "Engineer E4, status X"
self.assert_sql_count(testing.db, go, 1)
- session.clear()
+ session.expunge_all()
s = session.query(Car)
c = s.join("employee").filter(Person.name=="E4")[0]
assert c.car_id==car1.car_id
car1 = Car()
car2 = Car()
car2.manager = Manager()
- sess.save(car1)
- sess.save(car2)
+ sess.add(car1)
+ sess.add(car2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
carlist = sess.query(Car).all()
assert carlist[0].manager is None
m = Manager(name='manager1')
m2 =Manager(name='manager2')
m.colleague = m2
- sess.save(m)
+ sess.add(m)
sess.flush()
- sess.clear()
+ sess.expunge_all()
m = sess.query(Manager).get(m.person_id)
m2 = sess.query(Manager).get(m2.person_id)
assert m.colleague is m2
car=Car()
else:
car=Offraod_Car()
- session.save(Manager(name="M%d" % i,category="YYYYYYYYY",car=car))
- session.save(Engineer(name="E%d" % i,field="X",car=car))
+ session.add(Manager(name="M%d" % i,category="YYYYYYYYY",car=car))
+ session.add(Engineer(name="E%d" % i,field="X",car=car))
session.flush()
- session.clear()
+ session.expunge_all()
r = session.query(Person).all()
for p in r:
active = Status(name="active")
dead = Status(name="dead")
- session.save(active)
- session.save(dead)
+ session.add(active)
+ session.add(dead)
session.flush()
# TODO: we haven't created assertions for all the data combinations created here
st=active
else:
st=dead
- session.save(Manager(name="M%d" % i,category="YYYYYYYYY",status=st))
- session.save(Engineer(name="E%d" % i,field="X",status=st))
+ session.add(Manager(name="M%d" % i,category="YYYYYYYYY",status=st))
+ session.add(Engineer(name="E%d" % i,field="X",status=st))
session.flush()
# create 2 cars for E4, one active and one dead
car1 = Car(employee=engineer4,status=active)
car2 = Car(employee=engineer4,status=dead)
- session.save(car1)
- session.save(car2)
+ session.add(car1)
+ session.add(car2)
session.flush()
# this particular adapt used to cause a recursion overflow;
c = Manager().set( name= 'head', machine= 'fast', duties= 'many')
session = create_session()
- session.save(a)
- session.save(b)
- session.save(c)
+ session.add(a)
+ session.add(b)
+ session.add(c)
session.flush()
assert set(session.query(Employee).all()) == set([a,b,c])
assert set(session.query( Engineer).all()) == set([b,c])
ot1 = T1()
ot2 = T2()
sess = create_session()
- sess.save(ot1)
- sess.save(ot2)
+ sess.add(ot1)
+ sess.add(ot2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# query using get(), using only one value. this requires the select_table mapper
# has the same single-col primary key.
ot1 = T1()
ot2 = T2()
sess = create_session()
- sess.save(ot1)
- sess.save(ot2)
+ sess.add(ot1)
+ sess.add(ot2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# query using get(), using only one value. this requires the select_table mapper
# has the same single-col primary key.
session = create_session()
bob = Employee()
- session.save(bob)
+ session.add(bob)
tag = Tag('crazy')
bob.tags.append(tag)
bob.tags.append(tag)
session.flush()
- session.clear()
+ session.expunge_all()
# query from Employee with limit, query needs to apply eager limiting subquery
instance = session.query(Employee).filter_by(id=1).limit(1).first()
assert len(instance.tags) == 2
c = C(cdata='c1', adata='a1', b=B(data='c'))
d = D(cdata='c2', adata='a2', ddata='d2', b=B(data='d'))
sess = create_session()
- sess.save(c)
- sess.save(d)
+ sess.add(c)
+ sess.add(d)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(A).all(), [C(cdata='c1', adata='a1'), D(cdata='c2', adata='a2', ddata='d2')])
if __name__ == "__main__":
d1 = Detail(name='d1')
a1.specification.append(SpecLine(slave=d1))
- session.save(a1)
+ session.add(a1)
orig = repr(a1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Product).filter_by(name='a1').one()
new = repr(a1)
s = SpecLine(slave=Product(name='p1'))
s2 = SpecLine(slave=Detail(name='d1'))
- session.save(s)
- session.save(s2)
+ session.add(s)
+ session.add(s2)
orig = repr([s, s2])
session.flush()
- session.clear()
+ session.expunge_all()
new = repr(session.query(SpecLine).all())
print orig
print new
a1.specification.append(SpecLine(slave=Detail(name='d1')))
a1.documents.append(Document('doc1'))
a1.documents.append(RasterDocument('doc2'))
- session.save(a1)
+ session.add(a1)
orig = repr(a1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Product).filter_by(name='a1').one()
new = repr(a1)
a1 = Assembly(name='a1')
a1.documents.append(RasterDocument('doc2'))
- session.save(a1)
+ session.add(a1)
orig = repr(a1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Product).filter_by(name='a1').one()
new = repr(a1)
del a1.documents[0]
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Product).filter_by(name='a1').one()
assert len(session.query(Document).all()) == 0
a1.specification.append(SpecLine(slave=Detail(name='d1')))
a1.documents.append(Document('doc1'))
a1.documents.append(RasterDocument('doc2'))
- session.save(a1)
+ session.add(a1)
orig = repr(a1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Product).filter_by(name='a1').one()
new = repr(a1)
c2.employees = [e3]
sess = create_session()
- sess.save(c1)
- sess.save(c2)
+ sess.add(c1)
+ sess.add(c2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
all_employees = [e1, e2, b1, m1, e3]
c1_employees = [e1, e2, b1, m1]
sess = create_session()
for aliased in (True, False):
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Person).with_polymorphic(Manager).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%review%')).all(), [b1, m1])
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Person).with_polymorphic([Manager, Engineer]).join('paperwork', aliased=aliased).filter(Paperwork.description.like('%#2%')).all(), [e1, m1])
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Person).with_polymorphic([Manager, Engineer]).join('paperwork', aliased=aliased).filter(Person.name.like('%dog%')).filter(Paperwork.description.like('%#2%')).all(), [m1])
def test_join_to_polymorphic(self):
self.assertEquals(sess.query(Person).with_polymorphic(Engineer).filter(Engineer.primary_language=='java').all(), emps_without_relations[0:1])
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(Person).with_polymorphic('*').all(), emps_without_relations)
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(Person).with_polymorphic(Engineer).all(), emps_without_relations)
self.assert_sql_count(testing.db, go, 3)
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(Person).with_polymorphic(Engineer, people.outerjoin(engineers)).all(), emps_without_relations)
self.assert_sql_count(testing.db, go, 3)
- sess.clear()
+ sess.expunge_all()
def go():
# limit the polymorphic join down to just "Person", overriding select_table
self.assertEquals(sess.query(Person).with_polymorphic(Person).all(), emps_without_relations)
p1 = Person(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1)
sess = create_session()
- sess.save(p1)
- sess.save(e1)
+ sess.add(p1)
+ sess.add(e1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Engineer).filter(Engineer.reports_to.has(Person.name=='dogbert')).first(), Engineer(name='dilbert'))
p1 = Person(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=p1)
sess = create_session()
- sess.save(p1)
- sess.save(e1)
+ sess.add(p1)
+ sess.add(e1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(
sess.query(Engineer).join('reports_to', aliased=True).filter(Person.name=='dogbert').first(),
m1 = Manager(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1)
sess = create_session()
- sess.save(m1)
- sess.save(e1)
+ sess.add(m1)
+ sess.add(e1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Engineer).filter(Engineer.reports_to.has(Manager.name=='dogbert')).first(), Engineer(name='dilbert'))
m1 = Manager(name='dogbert')
e1 = Engineer(name='dilbert', primary_language='java', reports_to=m1)
sess = create_session()
- sess.save(m1)
- sess.save(e1)
+ sess.add(m1)
+ sess.add(e1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(
sess.query(Engineer).join('reports_to', aliased=True).filter(Manager.name=='dogbert').first(),
org2 = Organization(name='org2', engineers=[e3, e4])
sess = create_session()
- sess.save(org1)
- sess.save(org2)
+ sess.add(org1)
+ sess.add(org2)
sess.flush()
def test_not_contains(self):
e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed', company=c1)
sess.add_all([c1, c2, m1, m2, e1, e2])
sess.commit()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(
sess.query(Company).filter(Company.employees.of_type(JuniorEngineer).any()).all(),
[
self.assertEquals(c1.engineers, [e2])
self.assertEquals(c2.engineers, [e1])
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company).order_by(Company.name).all(),
[
Company(name='c1', engineers=[JuniorEngineer(name='Ed')]),
)
# eager load join should limit to only "Engineer"
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company).options(eagerload('engineers')).order_by(Company.name).all(),
[
Company(name='c1', engineers=[JuniorEngineer(name='Ed')]),
)
# join() to Company.engineers, Employee as the requested entity
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company, Employee).join(Company.engineers).order_by(Company.name).all(),
[
(Company(name='c1'), JuniorEngineer(name='Ed')),
# join() to Company.engineers, Engineer as the requested entity.
# this actually applies the IN criterion twice which is less than ideal.
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company, Engineer).join(Company.engineers).order_by(Company.name).all(),
[
(Company(name='c1'), JuniorEngineer(name='Ed')),
)
# join() to Company.engineers without any Employee/Engineer entity
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company).join(Company.engineers).filter(Engineer.name.in_(['Tom', 'Kurt'])).all(),
[
Company(name='c2')
# section to "inheritance" laying out all the various behaviors Query has.
@testing.fails_on_everything_except()
def go():
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Company).\
filter(Company.company_id==Engineer.company_id).filter(Engineer.name.in_(['Tom', 'Kurt'])).all(),
[
mapper(Manager, inherits=Employee,polymorphic_identity='manager')
sess = create_session()
- sess.save(Person(name='p1'))
- sess.save(Employee(name='e1', employee_data='ed1'))
- sess.save(Manager(name='m1', employee_data='ed2', manager_data='md1'))
+ sess.add(Person(name='p1'))
+ sess.add(Employee(name='e1', employee_data='ed1'))
+ sess.add(Manager(name='m1', employee_data='ed2', manager_data='md1'))
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Person).order_by(Person.person_id).all(), [
Person(name='p1'),
Employee(name='e1', employee_data='ed1'),
Manager(name='m1', employee_data='ed2', manager_data='md1')
])
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Employee).order_by(Person.person_id).all(), [
Employee(name='e1', employee_data='ed1'),
Manager(name='m1', employee_data='ed2', manager_data='md1')
])
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(Manager).order_by(Person.person_id).all(), [
Manager(name='m1', employee_data='ed2', manager_data='md1')
])
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(Person).with_polymorphic('*').order_by(Person.person_id).all(), [
ad1 = Address(email_address='somenewaddress', id=12)
sess.add(ad1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
ad2 = sess.query(Address).get(1)
ad3 = sess.query(Address).get(ad1.id)
p3.places.append(p4)
sess.flush()
- sess.clear()
+ sess.expunge_all()
l = sess.query(Place).order_by(place.c.place_id).all()
(p1, p2, p3, p4, p5, p6, p7) = l
assert p1.places == [p2,p3,p5]
sess.add(tran)
sess.flush()
- sess.clear()
+ sess.expunge_all()
r = sess.query(Transition).all()
self.assert_unordered_result(r, Transition,
{'name': 'transition1',
self.assert_(len(c1.students) == 1)
sess.add(s1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
s = sess.query(Student).filter_by(name='Student1').one()
c = sess.query(Course).filter_by(name='Course3').one()
self.assert_(len(s.courses) == 3)
u.name = 'ed'
u3 = User()
u3.name = 'some user'
- sess.save(u3)
+ sess.add(u3)
sess.flush()
sess.rollback()
m = mapper(User, users)
m.add_property('name', synonym('_name', map_column=True))
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).filter_by(name='jack').one()
eq_(u._name, 'jack')
eq_(u.name, 'jack')
eq_(u.uc_name, "SOME USER NAME")
sess.flush()
- sess.clear()
+ sess.expunge_all()
q = sess.query(User)
u2 = q.filter(User.name=='some user name').one()
eq_(len(u.addresses), 3)
self.sql_count_(0, go)
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).filter_by(id=8).one()
eq_(u.id, 8)
eq_(l, self.static.user_address_result)
self.sql_count_(1, go)
- sess.clear()
+ sess.expunge_all()
# then select just from users. run it into instances.
# then assert the data, which will launch 3 more lazy loads
eq_(l, self.static.user_all_result)
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
# then select just from users. run it into instances.
# then assert the data, which will launch 6 more lazy loads
eq_(u1.name, 'ed modified')
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).filter_by(name='ed modified').one(), User(name='ed'))
u1.addresses.append(Address(id=15, email_address='foo@bar.com'))
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(
sess.query(User).filter_by(name='edward').one(),
User(name='edward', addresses=[Address(email_address='foo@bar.com')])
sess = create_session()
o = Order()
- sess.save(o)
+ sess.add(o)
o.id = 7
def go():
o.description = "some description"
sess = create_session()
o = Order()
- sess.save(o)
+ sess.add(o)
o.id = 7
def go():
o.description = "some description"
sess = create_session()
o = Order()
- sess.save(o)
+ sess.add(o)
def go():
o.description = "some description"
self.sql_count_(0, go)
("SELECT orders.user_id AS orders_user_id "
"FROM orders WHERE orders.id = :param_1",
{'param_1':1})])
- sess.clear()
+ sess.expunge_all()
q2 = q.options(sa.orm.undefer('user_id'))
self.sql_eq_(q2.all, [
self.sql_count_(1, go)
eq_(item.description, 'item 4')
- sess.clear()
+ sess.expunge_all()
l = q.options(sa.orm.undefer('orders.items.description')).all()
item = l[0].orders[1].items[1]
def go():
def test_query_twice_with_clear(self):
session = create_session()
result = session.query(Thing).first()
- session.clear()
+ session.expunge_all()
thing = session.query(Thing).options(sa.orm.undefer("name")).first()
self._test(thing)
def test_eagerload_with_clear(self):
session = create_session()
human = session.query(Human).options(sa.orm.eagerload("thing")).first()
- session.clear()
+ session.expunge_all()
thing = session.query(Thing).options(sa.orm.undefer("name")).first()
self._test(thing)
def test_join_with_clear(self):
session = create_session()
result = session.query(Human).add_entity(Thing).join("thing").first()
- session.clear()
+ session.expunge_all()
thing = session.query(Thing).options(sa.orm.undefer("name")).first()
self._test(thing)
g.version_id=1
g.edges.append(Edge(Point(3, 4), Point(5, 6)))
g.edges.append(Edge(Point(14, 5), Point(2, 7)))
- sess.save(g)
+ sess.add(g)
sess.flush()
- sess.clear()
+ sess.expunge_all()
g2 = sess.query(Graph).get([g.id, g.version_id])
for e1, e2 in zip(g.edges, g2.edges):
eq_(e1.start, e2.start)
g2.edges[1].end = Point(18, 4)
sess.flush()
- sess.clear()
+ sess.expunge_all()
e = sess.query(Edge).get(g2.edges[1].id)
eq_(e.end, Point(18, 4))
e.end.x = 19
e.end.y = 5
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(Edge).get(g2.edges[1].id).end, Point(19, 5))
g.edges[1].end = Point(19, 5)
- sess.clear()
+ sess.expunge_all()
def go():
g2 = (sess.query(Graph).
options(sa.orm.eagerload('edges'))).get([g.id, g.version_id])
sess = create_session()
g = Graph(Version(1, 1))
- sess.save(g)
+ sess.add(g)
sess.flush()
- sess.clear()
+ sess.expunge_all()
g2 = sess.query(Graph).get([1, 1])
eq_(g.version, g2.version)
- sess.clear()
+ sess.expunge_all()
g2 = sess.query(Graph).get(Version(1, 1))
eq_(g.version, g2.version)
@testing.fails_on_everything_except("sqlite")
def go():
g = Graph(Version(2, None))
- sess.save(g)
+ sess.add(g)
sess.flush()
- sess.clear()
+ sess.expunge_all()
g2 = sess.query(Graph).filter_by(version=Version(2, None)).one()
eq_(g.version, g2.version)
go()
sess = create_session()
f1 = Foobar()
f1.foob = FBComposite(None, 5, None, None)
- sess.save(f1)
+ sess.add(f1)
sess.flush()
assert f1.foob == FBComposite(2, 5, 15, None)
sess = create_session()
f1 = Foobar()
f1.foob = FBComposite(None, 5, None, None)
- sess.save(f1)
+ sess.add(f1)
sess.flush()
assert f1.foob == FBComposite(2, 5, 15, None)
e = Edge(None, None)
g.edges.append(e)
- sess.save(g)
+ sess.add(g)
sess.flush()
- sess.clear()
+ sess.expunge_all()
g2 = sess.query(Graph).get([1, 1])
assert g2.edges[-1].start.x is None
mapper(User, users, extension=Ext())
sess = create_session()
u = User(name='u1')
- sess.save(u)
+ sess.add(u)
sess.flush()
u = sess.query(User).populate_existing().get(u.id)
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(u.id)
u.name = 'u1 changed'
sess.flush()
sess = create_session()
am = AdminUser(name='au1', email_address='au1@e1')
- sess.save(am)
+ sess.add(am)
sess.flush()
am = sess.query(AdminUser).populate_existing().get(am.id)
- sess.clear()
+ sess.expunge_all()
am = sess.query(AdminUser).get(am.id)
am.name = 'au1 changed'
sess.flush()
sess = create_session()
i1 = Item(description="i1")
k1 = Keyword(name="k1")
- sess.save(i1)
- sess.save(k1)
+ sess.add(i1)
+ sess.add(k1)
sess.flush()
eq_(methods,
['instrument_class', 'instrument_class', 'init_instance',
sess = create_session()
am = AdminUser(name="au1", email_address="au1@e1")
- sess.save(am)
+ sess.add(am)
sess.flush()
am = sess.query(AdminUser).populate_existing().get(am.id)
- sess.clear()
+ sess.expunge_all()
am = sess.query(AdminUser).get(am.id)
am.name = 'au1 changed'
sess.flush()
u1.name = 'ed'
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(User).first()
s = create_session()
for i in range(3):
h1 = H1()
- s.save(h1)
+ s.add(h1)
h1.h2s.append(H2())
h1.h3s.extend([H3(), H3()])
m = Map(state='AK', mapper=c)
sess = create_session()
- sess.save(c)
+ sess.add(c)
sess.flush()
- sess.clear()
+ sess.expunge_all()
for C, M in ((Cartographer, Map),
(sa.orm.aliased(Cartographer), sa.orm.aliased(Map))):
assert u2 in sess
eq_(u2, User(id=7, name='fred'))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).first(), User(id=7, name='fred'))
@testing.resolve_artifact_names
assert merged_users[0] is not u
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).one(),
User(id=7, name='fred', addresses=OrderedSet([
u = User(id=7, name='fred')
sess.add(u)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(on_load.called, 0)
assert u2 is not _u2
eq_(on_load.called, 1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).first(), User(id=7, name='fred jones'))
eq_(on_load.called, 2)
sess = create_session()
sess.add(u)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(on_load.called, 0)
]))
)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).one(),
User(id=7, name='fred', addresses=OrderedSet([
Address(id=3, email_address='fred3'),
sess = create_session()
sess.add(u)
sess.flush()
- sess.clear()
+ sess.expunge_all()
u.name='fred jones'
u.addresses.add(Address(id=3, email_address='fred3'))
u = sess.merge(u)
eq_(on_load.called, 4)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(User).first(),
User(id=7, name='fred jones', addresses=OrderedSet([
Address(email_address='hoho@bar.com')]))
sess.flush()
- sess.clear()
+ sess.expunge_all()
u2 = sess.query(User).get(7)
eq_(u2, User(id=7, name='fred', addresses=[
sess3.flush()
# assert modified/merged data was saved
- sess.clear()
+ sess.expunge_all()
u = sess.query(User).get(7)
eq_(u, User(id=7, name='fred2', addresses=[
Address(email_address='foo@bar.com'),
assert not sa.orm.object_mapper(a2)._is_orphan(
sa.orm.attributes.instance_state(a2))
sess2.flush()
- sess2.clear()
+ sess2.expunge_all()
eq_(sess2.query(User).get(u2.id).addresses[0].email_address,
'somenewaddress')
assert not sa.orm.object_mapper(a2)._is_orphan(
sa.orm.attributes.instance_state(a2))
sess2.flush()
- sess2.clear()
+ sess2.expunge_all()
eq_(sess2.query(User).get(u2.id).addresses[0].email_address,
'somenewaddress')
except sa.exc.InvalidRequestError, e:
sess = create_session()
u = User()
u.name = 'ed'
- sess.save(u)
+ sess.add(u)
sess.flush()
sess.expunge(u)
sess.merge(u)
assert sess.query(User).get('jack') is None
- sess.clear()
+ sess.expunge_all()
u1 = sess.query(User).get('ed')
self.assertEquals(User(username='ed', fullname='jack'), u1)
sess.expire(u1)
self.assertRaises(sa.orm.exc.ObjectDeletedError, getattr, u1, 'username')
- sess.clear()
+ sess.expunge_all()
assert sess.query(User).get('jack') is None
assert sess.query(User).get('ed').fullname == 'jack'
sess.expire(u1)
u1.username = 'ed'
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(User).get('ed').fullname == 'jack'
sess.flush()
assert u1.addresses[0].username == 'ed'
- sess.clear()
+ sess.expunge_all()
self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
u1 = sess.query(User).get('ed')
self.assert_sql_count(testing.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses
else:
self.assert_sql_count(testing.db, go, 1) # test passive_updates=True; update user
- sess.clear()
+ sess.expunge_all()
assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.query(User).get('jack')
u1 = sess.query(User).get('jack')
u1.addresses = []
u1.username = 'fred'
sess.flush()
- sess.clear()
+ sess.expunge_all()
assert sess.query(Address).get('jack1').username is None
u1 = sess.query(User).get('fred')
self.assertEquals(User(username='fred', fullname='jack'), u1)
self.assert_sql_count(testing.db, go, 0)
assert a1.username == a2.username == 'ed'
- sess.clear()
+ sess.expunge_all()
self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
@testing.fails_on('sqlite', 'FIXME: unknown')
else:
self.assert_sql_count(testing.db, go, 3)
self.assertEquals([Address(username='ed'), Address(username='ed')], [ad1, ad2])
- sess.clear()
+ sess.expunge_all()
self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
u1 = sess.query(User).get('ed')
self.assert_sql_count(testing.db, go, 1)
else:
self.assert_sql_count(testing.db, go, 3)
- sess.clear()
+ sess.expunge_all()
self.assertEquals([Address(username='fred'), Address(username='fred')], sess.query(Address).all())
sess.flush()
self.assert_sql_count(testing.db, go, 0)
- sess.clear()
+ sess.expunge_all()
r = sess.query(Item).all()
self.assertEquals(Item(itemname='item1'), r[0])
self.assertEquals(['jack'], [u.username for u in r[0].users])
self.assertEquals(Item(itemname='item2'), r[1])
self.assertEquals(['ed', 'jack'], sorted([u.username for u in r[1].users]))
- sess.clear()
+ sess.expunge_all()
u2 = sess.query(User).get(u2.username)
u2.username='wendy'
sess.flush()
assert u1.addresses[0].username == 'ed'
self.assertEquals(sa.select([addresses.c.username]).execute().fetchall(), [('ed',), ('ed',)])
- sess.clear()
+ sess.expunge_all()
self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
u1 = sess.query(User).get(u1.id)
self.assert_sql_count(testing.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses
else:
self.assert_sql_count(testing.db, go, 1) # test passive_updates=True; update user
- sess.clear()
+ sess.expunge_all()
assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.query(User).get(u1.id)
- sess.clear()
+ sess.expunge_all()
u1 = sess.query(User).get(u1.id)
u1.addresses = []
u1.username = 'fred'
sess.flush()
- sess.clear()
+ sess.expunge_all()
a1 = sess.query(Address).get(a1.id)
self.assertEquals(a1.username, None)
p.jack = None
assert j.port is None
- ctx.clear()
+ ctx.expunge_all()
j = ctx.query(Jack).get(jid)
p = ctx.query(Port).get(pid)
sess.add(u2)
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(u1, sess.query(User).get(u2.id))
u1.addresses.append(Address(email_address='ed@bar.com'))
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
u1 = sess.query(User).get(u1.id)
assert 'name' not in u1.__dict__
assert 'addresses' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
- sess2.update(u2)
+ sess2.add(u2)
self.assertEquals(u2.name, 'ed')
self.assertEquals(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
u1.addresses.append(Address(email_address='ed@bar.com'))
sess.add(u1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
u1 = sess.query(User).options(sa.orm.defer('name'), sa.orm.defer('addresses.email_address')).get(u1.id)
assert 'name' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
- sess2.update(u2)
+ sess2.add(u2)
self.assertEquals(u2.name, 'ed')
assert 'addresses' not in u2.__dict__
ad = u2.addresses[0]
sess = create_session()
sess.add(eu)
sess.flush()
- sess.clear()
+ sess.expunge_all()
eu = sess.query(User).first()
eu2 = pickle.loads(pickle.dumps(eu))
sess2 = create_session()
- sess2.update(eu2)
+ sess2.add(eu2)
assert 'email_address' not in eu2.__dict__
self.assertEquals(eu2.email_address, 'foo@bar.com')
u = s.query(User).get(7)
u2 = s.query(User).get(7)
assert u is u2
- s.clear()
+ s.expunge_all()
u2 = s.query(User).get(7)
assert u is not u2
u = s.query(User).populate_existing().get(7)
u2 = s.query(User).populate_existing().get(7)
assert u is u2
- s.clear()
+ s.expunge_all()
u2 = s.query(User).populate_existing().get(7)
assert u is not u2
assert [User(id=7), User(id=9), User(id=8)] == q.all()
- sess.clear()
+ sess.expunge_all()
# test that it works on embedded eagerload/LIMIT subquery
q = sess.query(User).join('addresses').distinct().options(eagerload('addresses')).order_by(desc(Address.email_address)).limit(2)
l = q.order_by(User.id, AdAlias.id).all()
self.assertEquals(l, expected)
- sess.clear()
+ sess.expunge_all()
q = sess.query(User).add_entity(AdAlias)
l = q.select_from(outerjoin(User, AdAlias)).filter(AdAlias.email_address=='ed@bettyboop.com').all()
assert fixtures.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
l = q.options(contains_alias('ulist'), contains_eager('addresses')).from_statement(query).all()
def go():
assert fixtures.user_address_result == q.all()
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
adalias = addresses.alias()
q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias))
def go():
self.assertEquals(fixtures.user_address_result, q.order_by(User.id).all())
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
selectquery = users.outerjoin(addresses).select(users.c.id<10, use_labels=True, order_by=[users.c.id, addresses.c.id])
q = sess.query(User)
assert fixtures.user_address_result[0:3] == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
l = list(q.options(contains_eager(User.addresses)).instances(selectquery.execute()))
assert fixtures.user_address_result[0:3] == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
l = q.options(contains_eager('addresses')).from_statement(selectquery).all()
l = list(q.options(contains_eager('addresses', alias="adalias")).instances(selectquery.execute()))
assert fixtures.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
# expression.Alias object
def go():
assert fixtures.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
# Aliased object
adalias = aliased(Address)
l = q.options(contains_eager('addresses', alias=adalias)).outerjoin((adalias, User.addresses)).order_by(User.id, adalias.id)
assert fixtures.user_address_result == l.all()
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
oalias = orders.alias('o1')
ialias = items.alias('i1')
assert fixtures.user_order_result == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
# test using Alias with more than one level deep
def go():
l = list(q.options(contains_eager('orders', alias=oalias), contains_eager('orders.items', alias=ialias)).instances(query.execute()))
assert fixtures.user_order_result == l
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
# test using Aliased with more than one level deep
oalias = aliased(Order)
outerjoin((oalias, User.orders), (ialias, oalias.items)).order_by(User.id, oalias.id, ialias.id)
assert fixtures.user_order_result == l.all()
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def test_mixed_eager_contains_with_limit(self):
sess = create_session()
Order(address_id=None,user_id=7,description=u'order 5',isopen=0,id=5)
])])
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
# same as above, except Order is aliased, so two adapters are applied by the
selectquery = users.outerjoin(addresses).select(use_labels=True, order_by=[users.c.id, addresses.c.id])
self.assertEquals(list(sess.query(User, Address).instances(selectquery.execute())), expected)
- sess.clear()
+ sess.expunge_all()
for address_entity in (Address, aliased(Address)):
q = sess.query(User).add_entity(address_entity).outerjoin(('addresses', address_entity)).order_by(User.id, address_entity.id)
self.assertEquals(q.all(), expected)
- sess.clear()
+ sess.expunge_all()
q = sess.query(User).add_entity(address_entity)
q = q.join(('addresses', address_entity)).filter_by(email_address='ed@bettyboop.com')
self.assertEquals(q.all(), [(user8, address3)])
- sess.clear()
+ sess.expunge_all()
q = sess.query(User, address_entity).join(('addresses', address_entity)).filter_by(email_address='ed@bettyboop.com')
self.assertEquals(q.all(), [(user8, address3)])
- sess.clear()
+ sess.expunge_all()
q = sess.query(User, address_entity).join(('addresses', address_entity)).options(eagerload('addresses')).filter_by(email_address='ed@bettyboop.com')
self.assertEquals(list(util.OrderedSet(q.all())), [(user8, address3)])
- sess.clear()
+ sess.expunge_all()
def test_aliased_multi_mappers(self):
sess = create_session()
l = q.order_by(User.id, adalias.c.id).all()
assert l == expected
- sess.clear()
+ sess.expunge_all()
q = sess.query(User).add_entity(Address, alias=adalias)
l = q.select_from(users.outerjoin(adalias)).filter(adalias.c.email_address=='ed@bettyboop.com').all()
for add_col in (User.name, users.c.name):
assert sess.query(User).add_column(add_col).all() == expected
- sess.clear()
+ sess.expunge_all()
self.assertRaises(sa_exc.InvalidRequestError, sess.query(User).add_column, object())
q = sess.query(User)
q = q.group_by([c for c in users.c]).order_by(User.id).outerjoin('addresses').add_column(func.count(Address.id).label('count'))
self.assertEquals(q.all(), expected)
- sess.clear()
+ sess.expunge_all()
adalias = aliased(Address)
q = sess.query(User)
q = q.group_by([c for c in users.c]).order_by(User.id).outerjoin(('addresses', adalias)).add_column(func.count(adalias.id).label('count'))
self.assertEquals(q.all(), expected)
- sess.clear()
+ sess.expunge_all()
s = select([users, func.count(addresses.c.id).label('count')]).select_from(users.outerjoin(addresses)).group_by(*[c for c in users.c]).order_by(User.id)
q = sess.query(User)
l = q.add_column("count").add_column("concat").from_statement(s).all()
assert l == expected
- sess.clear()
+ sess.expunge_all()
# test with select_from()
q = create_session().query(User).add_column(func.count(addresses.c.id))\
.group_by([c for c in users.c]).order_by(users.c.id)
assert q.all() == expected
- sess.clear()
+ sess.expunge_all()
q = create_session().query(User).add_column(func.count(addresses.c.id))\
.add_column(("Name:" + users.c.name)).outerjoin('addresses')\
.group_by([c for c in users.c]).order_by(users.c.id)
assert q.all() == expected
- sess.clear()
+ sess.expunge_all()
q = create_session().query(User).add_column(func.count(adalias.c.id))\
.add_column(("Name:" + users.c.name)).outerjoin(('addresses', adalias))\
.group_by([c for c in users.c]).order_by(users.c.id)
assert q.all() == expected
- sess.clear()
+ sess.expunge_all()
class ImmediateTest(_fixtures.FixtureTest):
])
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
sel2 = orders.select(orders.c.id.in_([1,2,3]))
self.assertEquals(sess.query(Order).select_from(sel2).join(['items', 'keywords']).filter(Keyword.name == 'red').order_by(Order.id).all(), [
Order(description=u'order 1',id=1),
]
)
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(User).options(eagerload('addresses')).select_from(sel).filter(User.id==8).order_by(User.id).all(),
[User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)])]
)
self.assert_sql_count(testing.db, go, 1)
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(User).options(eagerload('addresses')).select_from(sel).order_by(User.id)[1], User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)]))
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
- sess.save(n1)
+ sess.add(n1)
sess.flush()
sess.close()
n2.children = [n3, n6, n7]
n3.children = [n5, n4]
- sess.save(n1)
- sess.save(n2)
- sess.save(n3)
- sess.save(n4)
+ sess.add(n1)
+ sess.add(n2)
+ sess.add(n3)
+ sess.add(n4)
sess.flush()
sess.close()
# run the eager version twice to test caching of aliased clauses
for x in range(2):
- sess.clear()
+ sess.expunge_all()
def go():
self.assertEquals(sess.query(Address).options(eagerload('user')).all(), address_result)
self.assert_sql_count(testing.db, go, 1)
sess.add_all((c1, c2))
sess.flush()
- sess.clear()
+ sess.expunge_all()
test_c1 = sess.query(Company).get(c1.company_id)
test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
sess.add_all((c1, c2))
sess.flush()
- sess.clear()
+ sess.expunge_all()
test_c1 = sess.query(Company).get(c1.company_id)
test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
s.flush()
- s.clear()
+ s.expunge_all()
j = s.query(Job).filter_by(jobno=u'somejob').one()
oldp = list(j.pages)
j.pages = []
s.flush()
- s.clear()
+ s.expunge_all()
j = s.query(Job).filter_by(jobno=u'somejob2').one()
j.pages[1].current_version = 12
s.delete(j)
sess.flush()
assert a1 not in sess
assert b1 not in sess
- sess.clear()
+ sess.expunge_all()
sa.orm.clear_mappers()
@testing.resolve_artifact_names
sess.flush()
assert a1 not in sess
assert b1 not in sess
- sess.clear()
+ sess.expunge_all()
sa.orm.clear_mappers()
@testing.resolve_artifact_names
sess.flush()
assert a1 not in sess
assert b1 not in sess
- sess.clear()
+ sess.expunge_all()
@testing.resolve_artifact_names
def test_delete_manual_BtoA(self):
con.lineItems.append(li)
session.add(li)
session.flush()
- session.clear()
+ session.expunge_all()
newcon = session.query(Container).first()
assert con.policyNum == newcon.policyNum
assert len(newcon.lineItems) == 10
t1.foo.append(TagInstance(data='not_iplc_case'))
sess.add(t1)
sess.flush()
- sess.clear()
+ sess.expunge_all()
# relation works
eq_(sess.query(Tag).all(), [Tag(data='some tag', foo=[TagInstance(data='iplc_case')])])
sess.add(c1)
sess.add(c3)
sess.flush()
- sess.clear()
+ sess.expunge_all()
c1 = sess.query(C1).get(c1.id)
assert set([x.id for x in c1.t2s]) == set([c2a.id, c2b.id])
sess.add_all((c1, c3))
sess.flush()
- sess.clear()
+ sess.expunge_all()
c1 = sess.query(C1).get(c1.t1id)
assert set([x.t2id for x in c1.t2s]) == set([c2a.t2id, c2b.t2id])
sess.add_all((f1, f2))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(Foo).filter_by(id=f1.id).one(),
Foo(bars=[Bar(data='b1'), Bar(data='b2')]))
eq_(sess.query(Foo).filter_by(id=f2.id).one(),
sess.add_all((b1, b2, b3, b4))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(Foo).filter_by(id=f1.id).one(),
Foo(bars=[Bar(data='b1'), Bar(data='b2'), Bar(data='b4')]))
eq_(sess.query(Foo).filter_by(id=f2.id).one(),
sess = create_session()
sess.add(T2(data='t2', t1=T1(data='t1'), t3s=[T3(data='t3')]))
sess.flush()
- sess.clear()
+ sess.expunge_all()
a = sess.query(T1).first()
eq_(a.t3s, [T3(data='t3')])
b3 = T2(data='b3', t1id='Number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T1).first(),
T1(id='number1', data='a1', t2s=[
b3 = T2(data='b3', t1id='Number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
[T2(data='b1', t1=[T1(id='number1', data='a1')]),
b3 = T2(data='b2', t1id='number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T1).first(),
T1(id='NuMbeR1', data='a1', t2s=[
b3 = T2(data='b3', t1id='number2')
sess.add_all((a1, a2, b1, b2, b3))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
[T2(data='b1', t1=[T1(id='NuMbeR1', data='a1')]),
sso = SomeOtherObject()
s.options.append(sso)
Session.flush()
- Session.clear()
+ Session.expunge_all()
@testing.resolve_artifact_names
def test_query(self):
b = BaseClass(data='b1')
s = SubClass(data='s1', somedata='somedata')
Session.commit()
- Session.clear()
+ Session.expunge_all()
eq_(expunge_list([BaseClass(data='b1'),
SubClass(data='s1', somedata='somedata')]),
sess = create_session(bind=testing.db)
sess.add(Subset(data=1))
sess.flush()
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(Subset).all(), [Subset(data=1)])
eq_(sess.query(Subset).filter(Subset.data==1).one(), Subset(data=1))
assert c.scalar("select count(1) from users") == 1
+ @testing.uses_deprecated()
@engines.close_open_connections
@testing.resolve_artifact_names
def test_save_update_delete(self):
# modify outside of session, assert changes remain/get saved
user.name = "fred"
- s.update(user)
+ s.add(user)
assert user in s
assert user in s.dirty
s.flush()
- s.clear()
+ s.expunge_all()
assert s.query(User).count() == 1
user = s.query(User).one()
assert user.name == 'fred'
# ensure its not dirty if no changes occur
- s.clear()
+ s.expunge_all()
assert user not in s
- s.update(user)
+ s.add(user)
assert user in s
assert user not in s.dirty
u = User(name='fred')
s.add(u)
s.flush()
- s.clear()
+ s.expunge_all()
user = s.query(User).one()
assert user not in s.dirty
assert a not in s
s.flush()
print "\n".join([repr(x.__dict__) for x in s])
- s.clear()
+ s.expunge_all()
assert s.query(User).one().id == u.id
assert s.query(Address).first() is None
assert u not in s
assert a in s
s.flush()
- s.clear()
+ s.expunge_all()
assert s.query(Address).one().id == a.id
assert s.query(User).first() is None
sess.add(User(name='u2'))
sess.flush()
- sess.clear()
+ sess.expunge_all()
self.assertEquals(sess.query(User).order_by(User.name).all(),
[
User(name='u1 modified'),
assert u2 is not None and u2 is not u1
assert u2 in sess
- self.assertRaises(Exception, lambda: sess.update(u1))
+ self.assertRaises(Exception, lambda: sess.add(u1))
sess.expunge(u2)
assert u2 not in sess
u1.name = "John"
u2.name = "Doe"
- sess.update(u1)
+ sess.add(u1)
assert u1 in sess
assert Session.object_session(u1) is sess
sess.flush()
- sess.clear()
+ sess.expunge_all()
u3 = sess.query(User).get(u1.id)
assert u3 is not u1 and u3 is not u2 and u3.name == u1.name
self._test_session().flush()
def test_clear(self):
- self._test_session().clear()
+ self._test_session().expunge_all()
def test_close(self):
self._test_session().close()
return mapper(cls, Table('t', sa.MetaData(),
Column('id', Integer, primary_key=True)))
+ @testing.uses_deprecated()
def _test_instance_guards(self, user_arg):
watchdog = set()
sess = create_session()
for c in [tokyo, newyork, toronto, london, dublin, brasilia, quito]:
- sess.save(c)
+ sess.add(c)
sess.commit()
tokyo.city # reload 'city' attribute on tokyo
- sess.clear()
+ sess.expunge_all()
eq_(db2.execute(weather_locations.select()).fetchall(), [(1, 'Asia', 'Tokyo')])
eq_(db1.execute(weather_locations.select()).fetchall(), [(2, 'North America', 'New York'), (3, 'North America', 'Toronto')])
self.assert_(u.addresses == [a])
session.commit()
- session.clear()
+ session.expunge_all()
u = session.query(m).one()
assert u.addresses[0].user == u
session = create_session()
session.add(a1)
session.flush()
- session.clear()
+ session.expunge_all()
new_a1 = session.query(A).filter(t1.c.a == a1.a).one()
assert new_a1.a == a1.a
assert new_a1.t2s[0].d == b1.d
- session.clear()
+ session.expunge_all()
new_a1 = (session.query(A).options(sa.orm.eagerload('t2s')).
filter(t1.c.a == a1.a)).one()
assert new_a1.a == a1.a
assert new_a1.t2s[0].d == b1.d
- session.clear()
+ session.expunge_all()
new_a1 = session.query(A).filter(A.a == a1.a).one()
assert new_a1.a == a1.a
assert new_a1.t2s[0].d == b1.d
- session.clear()
+ session.expunge_all()
@testing.fails_on('mssql', 'pyodbc returns a non unicode encoding of the results description.')
@testing.resolve_artifact_names
session = create_session()
session.add_all((a1, b1))
session.flush()
- session.clear()
+ session.expunge_all()
eq_([A(b=5), B(e=7)], session.query(A).all())
session = create_session()
session.add(f1)
session.flush()
- session.clear()
+ session.expunge_all()
f2 = session.query(Foo).filter_by(id=f1.id).one()
assert 'data' in sa.orm.attributes.instance_state(f2).unmodified
assert f2 in session.dirty
assert 'data' not in sa.orm.attributes.instance_state(f2).unmodified
session.flush()
- session.clear()
+ session.expunge_all()
f3 = session.query(Foo).filter_by(id=f1.id).one()
ne_(f3.data,f1.data)
session = create_session(autocommit=False)
session.add(f1)
session.commit()
- session.clear()
+ session.expunge_all()
f1 = session.query(Foo).get(f1.id)
f1.val = u'hi'
f1.data[0]['personne']['VenSoir']= False
self.sql_count_(1, session.commit)
- session.clear()
+ session.expunge_all()
f = session.query(Foo).get(f1.id)
eq_(f.data,
[ {
session = create_session()
session.add(e)
session.flush()
- session.clear()
+ session.expunge_all()
e2 = session.query(Entry).get((e.multi_id, 2))
self.assert_(e is not e2)
assert (u.counter == 2) is True
self.sql_count_(1, go)
- session.clear()
+ session.expunge_all()
u = session.query(User).get(u.id)
eq_(u.name, 'test2')
eq_(u.counter, 2)
session.add(mc)
session.flush()
- session.clear()
+ session.expunge_all()
assert myothertable.count().scalar() == 4
mc = session.query(MyClass).get(mc.id)
mc.children.append(MyOtherClass())
session.add(mc)
session.flush()
- session.clear()
+ session.expunge_all()
assert myothertable.count().scalar() == 4
mc = session.query(MyClass).get(mc.id)
mc.children.append(MyOtherClass())
session.add(mc)
session.flush()
- session.clear()
+ session.expunge_all()
assert myothertable.count().scalar() == 1
self.assert_(h5.foober == 'im the new foober')
self.sql_count_(0, go)
- session.clear()
+ session.expunge_all()
(h1, h2, h3, h4, h5) = session.query(Hoho).order_by(Hoho.id).all()
session = create_session()
session.add(h1)
session.flush()
- session.clear()
+ session.expunge_all()
eq_(session.query(Hoho).get(h1.id),
Hoho(hoho=hohoval,
h1 = session.query(Hoho).get(h1.id)
h1.secondaries.append(Secondary(data='s2'))
session.flush()
- session.clear()
+ session.expunge_all()
eq_(session.query(Hoho).get(h1.id),
Hoho(hoho=hohoval,
session.delete(u1)
session.flush()
- session.clear()
+ session.expunge_all()
u2 = session.query(User).get(u2.id)
eq_(len(u2.addresses), 1)
u2.addresses.append(a)
session.flush()
- session.clear()
+ session.expunge_all()
u2 = session.query(User).get(u2.id)
eq_(len(u2.addresses), 1)
assert u is nu
# clear out the identity map, so next get forces a SELECT
- session.clear()
+ session.expunge_all()
# check it again, identity should be different but ids the same
nu = session.query(m).get(u.id)
# change first users name and save
session = create_session()
- session.update(u)
+ session.add(u)
u.name = 'modifiedname'
assert u in session.dirty
session.flush()
session = create_session()
session.add(u)
session.flush()
- session.clear()
+ session.expunge_all()
u = session.query(SUser).first()
eq_(u.syn_name, 'User:some name:User')
session = create_session()
session.add(u)
session.flush()
- session.clear()
+ session.expunge_all()
u = session.query(User).one()
u.name = 'newname'
session = create_session()
session.add(au)
session.flush()
- session.clear()
+ session.expunge_all()
rt = session.query(AddressUser).one()
eq_(au.user_id, rt.user_id)
eq_(list(session.execute(orders.select(), mapper=Order)),
[(42, None, None, 'foo', None)])
- session.clear()
+ session.expunge_all()
# assert that a set operation doesn't trigger a load operation
o = session.query(Order).filter(Order.description == 'foo').one()
eq_(list(session.execute(orders.select(), mapper=Order)),
[(42, None, None, 'hoho', None)])
- session.clear()
+ session.expunge_all()
# test assigning None to an unloaded deferred also works
o = session.query(Order).filter(Order.description == 'hoho').one()
session = create_session()
session.add(u)
session.flush()
- session.clear()
+ session.expunge_all()
u = session.query(User).get(u.id)
u.name = ''
session = create_session()
session.add(u)
session.flush()
- session.clear()
+ session.expunge_all()
id = m.primary_key_from_instance(u)
address_rows = addresses.select(addresses.c.id.in_([u.id])).execute().fetchall()
eq_(address_rows[0].values(), [u.id, u.foo_id, 'lala@hey.com'])
- session.clear()
+ session.expunge_all()
u = session.query(User).get(id)
assert u.name == 'imnew'
session = create_session()
session.add(u)
session.flush()
- session.clear()
+ session.expunge_all()
u = session.query(User).get(u.id)
session.delete(u)
# test insert ordering is maintained
assert names == ['user1', 'user2', 'user4', 'user5', 'user3']
- session.clear()
+ session.expunge_all()
sa.orm.clear_mappers()
session = create_session()
session.add(a1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
u1 = session.query(User).get(u1.id)
a1.user = None
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
u1 = session.query(User).get(u1.id)
assert a1.user is None
session = create_session()
session.add_all((a1, a2))
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
a2 = session.query(Address).get(a2.id)
a1.user = None
a2.user = u1
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
a2 = session.query(Address).get(a2.id)
session = create_session()
session.add_all((a1, u1, u2))
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
u1 = session.query(User).get(u1.id)
a1.user = u2
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
u1 = session.query(User).get(u1.id)
u2 = session.query(User).get(u2.id)
session = create_session()
session.add(u1)
session.flush()
- session.clear()
+ session.expunge_all()
a1 = session.query(Address).get(a1.id)
a1.user = None
session.flush()
- session.clear()
+ session.expunge_all()
assert session.query(Address).get(a1.id).user is None
assert session.query(User).get(u1.id).addresses == []
item.keywords.append(k2)
session.flush()
- session.clear()
+ session.expunge_all()
item = session.query(Item).get(item.id)
assert item.keywords == [k1, k2]
for clear in (False, True):
if clear:
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T).all(), [T(value=True, name="t1"), T(value=False, name="t2"), T(value=True, name="t3")])
if clear:
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T).filter(T.value==True).all(), [T(value=True, name="t1"),T(value=True, name="t3")])
if clear:
- sess.clear()
+ sess.expunge_all()
eq_(sess.query(T).filter(T.value==False).all(), [T(value=False, name="t2")])
t2 = sess.query(T).get(t2.id)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
