1. Hajime Nakagami
  2. sqlalchemy

Commits

Mike Bayer  committed 1cb34ab Merge

merge default

  • Participants
  • Parent commits 89717aa, 2309db1
  • Branches rel_0_8

Comments (0)

Files changed (7)

File doc/build/changelog/changelog_08.rst

View file
  • Ignore whitespace
     :version: 0.8.0
 
     .. change::
+        :tags: feature, orm
+        :tickets: 2658
+
+      Added new helper function :func:`.was_deleted`, returns True
+      if the given object was the subject of a :meth:`.Session.delete`
+      operation.
+
+    .. change::
+        :tags: bug, orm
+        :tickets: 2658
+
+      An object that's deleted from a session will be de-associated with
+      that session fully after the transaction is committed, that is
+      the :func:`.object_session` function will return None.
+
+    .. change::
         :tags: bug, oracle
 
       The cx_oracle dialect will no longer run the bind parameter names

File doc/build/orm/mapper_config.rst

View file
  • Ignore whitespace
                 orders.c.customer_id
                 ]).group_by(orders.c.customer_id).alias()
 
-    customer_select = select([customers,subq]).\
-                where(customers.c.customer_id==subq.c.customer_id)
+    customer_select = select([customers, subq]).\
+                select_from(
+                    join(customers, subq,
+                            customers.c.id == subq.c.customer_id)
+                ).alias()
 
     class Customer(Base):
         __table__ = customer_select
 will only emit an INSERT into a table for which it has mapped the primary
 key.
 
+.. note::
+
+    The practice of mapping to arbitrary SELECT statements, especially
+    complex ones as above, is
+    almost never needed; it necessarily tends to produce complex queries
+    which are often less efficient than that which would be produced
+    by direct query construction.   The practice is to some degree
+    based on the very early history of SQLAlchemy where the :func:`.mapper`
+    construct was meant to represent the primary querying interface;
+    in modern usage, the :class:`.Query` object can be used to construct
+    virtually any SELECT statement, including complex composites, and should
+    be favored over the "map-to-selectable" approach.
+
 Multiple Mappers for One Class
 ==============================
 

File doc/build/orm/session.rst

View file
  • Ignore whitespace
 
 .. autofunction:: object_session
 
+.. autofunction:: was_deleted
+
 Attribute and State Management Utilities
 -----------------------------------------
 

File lib/sqlalchemy/orm/__init__.py

View file
  • Ignore whitespace
      object_mapper,
      outerjoin,
      polymorphic_union,
+     was_deleted,
      with_parent,
      with_polymorphic,
      )
     'undefer',
     'undefer_group',
     'validates',
+    'was_deleted',
     'with_polymorphic'
     )
 

File lib/sqlalchemy/orm/session.py

View file
  • Ignore whitespace
         if not self.nested and self.session.expire_on_commit:
             for s in self.session.identity_map.all_states():
                 s._expire(s.dict, self.session.identity_map._modified)
+            for s in self._deleted:
+                s.session_id = None
+            self._deleted.clear()
+
 
     def _connection_for_bind(self, bind):
         self._assert_is_active()

File lib/sqlalchemy/orm/util.py

View file
  • Ignore whitespace
 
 
 def has_identity(object):
+    """Return True if the given object has a database
+    identity.
+
+    This typically corresponds to the object being
+    in either the persistent or detached state.
+
+    .. seealso::
+
+        :func:`.was_deleted`
+
+    """
     state = attributes.instance_state(object)
     return state.has_identity
 
+def was_deleted(object):
+    """Return True if the given object was deleted
+    within a session flush.
+
+    .. versionadded:: 0.8.0
+
+    """
+
+    state = attributes.instance_state(object)
+    return state.deleted
 
 def instance_str(instance):
     """Return a string describing an instance."""

File test/orm/test_session.py

View file
  • Ignore whitespace
 from sqlalchemy.testing import eq_, assert_raises, \
-    assert_raises_message, assert_warnings
+    assert_raises_message
 from sqlalchemy.testing.util import gc_collect
 from sqlalchemy.testing import pickleable
 from sqlalchemy.util import pickle
 from sqlalchemy import Integer, String, Sequence
 from sqlalchemy.testing.schema import Table, Column
 from sqlalchemy.orm import mapper, relationship, backref, joinedload, \
-    exc as orm_exc, object_session
+    exc as orm_exc, object_session, was_deleted
 from sqlalchemy.util import pypy
 from sqlalchemy.testing import fixtures
 from test.orm import _fixtures
 
         # ensure tables are unbound
         m2 = sa.MetaData()
-        users_unbound =users.tometadata(m2)
+        users_unbound = users.tometadata(m2)
         addresses_unbound = addresses.tometadata(m2)
 
         mapper(Address, addresses_unbound)
         mapper(User, users_unbound, properties={
-            'addresses':relationship(Address,
+            'addresses': relationship(Address,
                                  backref=backref("user", cascade="all"),
                                  cascade="all")})
 
-        Session = sessionmaker(binds={User: self.metadata.bind,
+        sess = Session(binds={User: self.metadata.bind,
                                       Address: self.metadata.bind})
-        sess = Session()
 
         u1 = User(id=1, name='ed')
         sess.add(u1)
-        eq_(sess.query(User).filter(User.id==1).all(),
+        eq_(sess.query(User).filter(User.id == 1).all(),
             [User(id=1, name='ed')])
 
         # test expression binding
 
         # ensure tables are unbound
         m2 = sa.MetaData()
-        users_unbound =users.tometadata(m2)
+        users_unbound = users.tometadata(m2)
         addresses_unbound = addresses.tometadata(m2)
 
         mapper(Address, addresses_unbound)
         mapper(User, users_unbound, properties={
-            'addresses':relationship(Address,
+            'addresses': relationship(Address,
                                  backref=backref("user", cascade="all"),
                                  cascade="all")})
 
 
         u1 = User(id=1, name='ed')
         sess.add(u1)
-        eq_(sess.query(User).filter(User.id==1).all(),
+        eq_(sess.query(User).filter(User.id == 1).all(),
             [User(id=1, name='ed')])
 
         sess.execute(users_unbound.insert(), params=dict(id=2, name='jack'))
 
         # use :bindparam style
         eq_(sess.execute("select * from users where id=:id",
-                         {'id':7}).fetchall(),
+                         {'id': 7}).fetchall(),
             [(7, u'jack')])
 
 
         # use :bindparam style
         eq_(sess.scalar("select id from users where id=:id",
-                         {'id':7}),
+                         {'id': 7}),
             7)
 
     def test_parameter_execute(self):
                                 self.classes.User)
 
         mapper(User, users, properties={
-            'addresses':relationship(Address, backref="user")})
+            'addresses': relationship(Address, backref="user")})
         mapper(Address, addresses)
 
         sess = create_session(autoflush=True, autocommit=False)
         u = User(name='ed', addresses=[Address(email_address='foo')])
         sess.add(u)
-        eq_(sess.query(Address).filter(Address.user==u).one(),
+        eq_(sess.query(Address).filter(Address.user == u).one(),
             Address(email_address='foo'))
 
         # still works after "u" is garbage collected
         sess.commit()
         sess.close()
         u = sess.query(User).get(u.id)
-        q = sess.query(Address).filter(Address.user==u)
+        q = sess.query(Address).filter(Address.user == u)
         del u
         gc_collect()
         eq_(q.one(), Address(email_address='foo'))
 
         mapper(User, users)
         conn1 = testing.db.connect()
-        conn2 = testing.db.connect()
         sess = create_session(bind=conn1, autocommit=False,
                               autoflush=True)
         u = User()
 
         s = create_session()
         mapper(User, users, properties={
-            'addresses':relationship(Address, cascade="all, delete")
+            'addresses': relationship(Address, cascade="all, delete")
         })
         mapper(Address, addresses)
 
         go()
         eq_(canary, [False])
 
+    def test_deleted_expunged(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        sess = Session()
+        sess.add(User(name='x'))
+        sess.commit()
+
+        u1 = sess.query(User).first()
+        sess.delete(u1)
+
+        assert not was_deleted(u1)
+        sess.flush()
+
+        assert was_deleted(u1)
+        assert u1 not in sess
+        assert object_session(u1) is sess
+        sess.commit()
+
+        assert object_session(u1) is None
+
 class SessionStateWFixtureTest(_fixtures.FixtureTest):
 
     def test_autoflush_rollback(self):
 
         mapper(Address, addresses)
         mapper(User, users, properties={
-            'addresses':relationship(Address)})
+            'addresses': relationship(Address)})
 
         sess = create_session(autocommit=False, autoflush=True)
         u = sess.query(User).get(8)
 
         mapper(Address, addresses)
         mapper(User, users, properties={
-            'addresses':relationship(Address,
+            'addresses': relationship(Address,
                                  backref=backref("user", cascade="all"),
                                  cascade="all")})
 
 
         s = sessionmaker()()
         mapper(User, users, properties={
-            "addresses":relationship(Address, backref="user")
+            "addresses": relationship(Address, backref="user")
         })
         mapper(Address, addresses)
         s.add(User(name="ed", addresses=[Address(email_address="ed1")]))
         s.commit()
 
         user = s.query(User).options(joinedload(User.addresses)).one()
-        user.addresses[0].user # lazyload
+        user.addresses[0].user  # lazyload
         eq_(user, User(name="ed", addresses=[Address(email_address="ed1")]))
 
         del user
         assert len(s.identity_map) == 0
 
         user = s.query(User).options(joinedload(User.addresses)).one()
-        user.addresses[0].email_address='ed2'
-        user.addresses[0].user # lazyload
+        user.addresses[0].email_address = 'ed2'
+        user.addresses[0].user  # lazyload
         del user
         gc_collect()
         assert len(s.identity_map) == 2
 
         s = sessionmaker()()
         mapper(User, users, properties={
-            "address":relationship(Address, backref="user", uselist=False)
+            "address": relationship(Address, backref="user", uselist=False)
         })
         mapper(Address, addresses)
         s.add(User(name="ed", address=Address(email_address="ed1")))
         assert len(s.identity_map) == 0
 
         user = s.query(User).options(joinedload(User.address)).one()
-        user.address.email_address='ed2'
-        user.address.user # lazyload
+        user.address.email_address = 'ed2'
+        user.address.user  # lazyload
 
         del user
         gc_collect()
         User, Address = self.classes.User, self.classes.Address
         users, addresses = self.tables.users, self.tables.addresses
         mapper(User, users, properties={
-            "addresses":relationship(Address)
+            "addresses": relationship(Address)
         })
         mapper(Address, addresses)
         return User, Address
         # can't predict result here
         # deterministically, depending on if
         # 'name' or 'addresses' is tested first
-        mod  = s.is_modified(u)
+        mod = s.is_modified(u)
         addresses_loaded = 'addresses' in u.__dict__
         assert mod is not addresses_loaded
 
 
         s = sessionmaker()()
 
-        mapper(User, users, properties={'uname':sa.orm.synonym('name')})
+        mapper(User, users, properties={'uname': sa.orm.synonym('name')})
         u = User(uname='fred')
         assert s.is_modified(u)
         s.add(u)
 
     @classmethod
     def define_tables(cls, metadata):
-        global t1
-        t1 = Table('t1', metadata, Column('id', Integer,
+        Table('t1', metadata, Column('id', Integer,
                    primary_key=True, test_needs_autoincrement=True),
                    Column('data', String(50)))
 
     @classmethod
-    def setup_mappers(cls):
-        global T
-        class T(object):
+    def setup_classes(cls):
+        class T(cls.Basic):
             def __init__(self, data):
                 self.data = data
-        mapper(T, t1)
+        mapper(T, cls.tables.t1)
 
     def teardown(self):
         from sqlalchemy.orm.session import _sessions
         """
 
         all_states = sess.identity_map.all_states()
-        sess.identity_map.all_states = lambda : all_states
+        sess.identity_map.all_states = lambda: all_states
         for obj in objs:
             state = attributes.instance_state(obj)
             sess.identity_map.discard(state)
             state._dispose()
 
     def _test_session(self, **kwargs):
-        global sess
+        T = self.classes.T
         sess = create_session(**kwargs)
 
         data = o1, o2, o3, o4, o5 = [T('t1'), T('t2'), T('t3'), T('t4'