- [feature] Added "no_autoflush" context

manager to Session, used with with:
will temporarily disable autoflush.
This commit is contained in:
Mike Bayer 2012-02-25 14:34:02 -05:00
parent abe6c0f08a
commit deb7e76a42
5 changed files with 56 additions and 1 deletions

View File

@ -6,6 +6,10 @@ CHANGES
0.7.6
=====
- orm
- [feature] Added "no_autoflush" context
manager to Session, used with with:
will temporarily disable autoflush.
- [bug] Fixed bug whereby MappedCollection
would not get the appropriate collection
instrumentation if it were only used

View File

@ -978,6 +978,34 @@ class Session(object):
return self._query_cls(entities, self, **kwargs)
@property
@util.contextmanager
def no_autoflush(self):
"""Return a context manager that disables autoflush.
e.g.::
with session.no_autoflush:
some_object = SomeClass()
session.add(some_object)
# won't autoflush
some_object.related_thing = session.query(SomeRelated).first()
Operations that proceed within the ``with:`` block
will not be subject to flushes occurring upon query
access. This is useful when initializing a series
of objects which involve existing database queries,
where the uncompleted object should not yet be flushed.
New in 0.7.6.
"""
autoflush = self.autoflush
self.autoflush = False
yield self
self.autoflush = autoflush
def _autoflush(self):
if self.autoflush and not self._flushing:
self.flush()

View File

@ -7,7 +7,7 @@
from compat import callable, cmp, reduce, defaultdict, py25_dict, \
threading, py3k_warning, jython, pypy, win32, set_types, buffer, pickle, \
update_wrapper, partial, md5_hex, decode_slice, dottedgetter,\
parse_qsl, any
parse_qsl, any, contextmanager
from _collections import NamedTuple, ImmutableContainer, immutabledict, \
Properties, OrderedProperties, ImmutableProperties, OrderedDict, \

View File

@ -56,6 +56,12 @@ def buffer(x):
buffer = buffer
# end Py2K
try:
from contextlib import contextmanager
except ImportError:
def contextmanager(fn):
return fn
try:
from functools import update_wrapper
except ImportError:

View File

@ -222,6 +222,23 @@ class SessionTest(_fixtures.FixtureTest):
eq_(bind.connect().execute("select count(1) from users").scalar(), 1)
sess.close()
@testing.requires.python26
def test_with_no_autoflush(self):
User, users = self.classes.User, self.tables.users
mapper(User, users)
sess = Session()
u = User()
u.name = 'ed'
sess.add(u)
def go(obj):
assert u not in sess.query(User).all()
testing.run_as_contextmanager(sess.no_autoflush, go)
assert u in sess.new
assert u in sess.query(User).all()
assert u not in sess.new
def test_make_transient(self):
users, User = self.tables.users, self.classes.User