Commits

Mike Bayer committed 49ea0b2

- [feature] The Query.update() method is now
more lenient as to the table
being updated. Plain Table objects are better
supported now, and additional a joined-inheritance
subclass may be used with update(); the subclass
table will be the target of the update,
and if the parent table is referenced in the
WHERE clause, the compiler will call upon
UPDATE..FROM syntax as allowed by the dialect
to satisfy the WHERE clause. Target columns
must still be in the target table i.e.
does not support MySQL's multi-table update
feature (even though this is in Core).
PG's DELETE..USING is also not available
in Core yet.

  • Participants
  • Parent commits e4c31b4

Comments (0)

Files changed (6)

     None values in collections would be silently ignored.
     [ticket:2229]
 
+  - [feature] The Query.update() method is now
+    more lenient as to the table
+    being updated.  Plain Table objects are better
+    supported now, and additional a joined-inheritance
+    subclass may be used with update(); the subclass
+    table will be the target of the update,
+    and if the parent table is referenced in the
+    WHERE clause, the compiler will call upon
+    UPDATE..FROM syntax as allowed by the dialect
+    to satisfy the WHERE clause.  Target columns
+    must still be in the target table i.e.
+    does not support MySQL's multi-table update
+    feature (even though this is in Core).
+    PG's DELETE..USING is also not available
+    in Core yet.
+
   - [bug] Improvements to joined/subquery eager
     loading dealing with chains of subclass entities
     sharing a common base, with no specific "join depth"

lib/sqlalchemy/orm/persistence.py

         self.context = context = query._compile_context()
         if len(context.statement.froms) != 1 or \
                     not isinstance(context.statement.froms[0], schema.Table):
-            raise sa_exc.ArgumentError(
-                            "Only update via a single table query is "
-                            "currently supported")
-        self.primary_table = context.statement.froms[0]
+
+            self.primary_table = query._only_entity_zero(
+                    "This operation requires only one Table or "
+                    "entity be specified as the target."
+                ).mapper.local_table
+        else:
+            self.primary_table = context.statement.froms[0]
 
         session = query.session
 

test/dialect/test_postgresql.py

 # coding: utf-8
+
+from __future__ import with_statement
+
 from test.lib.testing import eq_, assert_raises, assert_raises_message, is_
 from test.lib import  engines
 import datetime

test/lib/fixtures.py

 
     def _teardown_each_tables(self):
         # no need to run deletes if tables are recreated on setup
-        if self.run_define_tables != 'each' and self.run_deletes:
+        if self.run_define_tables != 'each' and self.run_deletes == 'each':
             for table in reversed(self.metadata.sorted_tables):
                 try:
                     table.delete().execute().close()
         pass
 
 class DeclarativeMappedTest(MappedTest):
-    declarative_meta = None
-
     run_setup_classes = 'once'
     run_setup_mappers = 'once'
 
     @classmethod
-    def setup_class(cls):
-        if cls.declarative_meta is None:
-            cls.declarative_meta = sa.MetaData()
-
-        super(DeclarativeMappedTest, cls).setup_class()
-
-    @classmethod
-    def _teardown_once_class(cls):
-        if cls.declarative_meta.tables:
-            cls.declarative_meta.drop_all(testing.db)
-        super(DeclarativeMappedTest, cls)._teardown_once_class()
+    def _setup_once_tables(cls):
+        pass
 
     @classmethod
     def _with_register_classes(cls, fn):
                         cls, classname, bases, dict_)
         class DeclarativeBasic(object):
             __table_cls__ = schema.Table
-        _DeclBase = declarative_base(metadata=cls.declarative_meta,
+        _DeclBase = declarative_base(metadata=cls.metadata,
                             metaclass=FindFixtureDeclarative,
                             cls=DeclarativeBasic)
         cls.DeclarativeBasic = _DeclBase
         fn()
-        if cls.declarative_meta.tables:
-            cls.declarative_meta.create_all(testing.db)
+        if cls.metadata.tables:
+            cls.metadata.create_all(testing.db)

test/orm/test_of_type.py

     run_setup_classes = 'once'
     run_setup_mappers = 'once'
     run_inserts = 'once'
+    run_deletes = None
     __dialect__ = 'default'
 
     @classmethod

test/orm/test_update_delete.py

                 r"Can't call Query.delete\(\) when %s\(\) has been called" % mname,
                 q.delete)
 
-
     def test_delete(self):
         User = self.classes.User
 
 
         eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
 
+    def test_delete_against_metadata(self):
+        User = self.classes.User
+        users = self.tables.users
+
+        sess = Session()
+        sess.query(users).delete(synchronize_session=False)
+        eq_(sess.query(User).count(), 0)
+
     def test_delete_with_bindparams(self):
         User = self.classes.User
 
         eq_([john.age, jack.age, jill.age, jane.age], [15,27,19,27])
         eq_(sess.query(User.age).order_by(User.id).all(), zip([15,27,19,27]))
 
+    def test_update_against_metadata(self):
+        User, users = self.classes.User, self.tables.users
+
+        sess = Session()
+
+        sess.query(users).update({users.c.age: 29}, synchronize_session=False)
+        eq_(sess.query(User.age).order_by(User.id).all(), zip([29,29,29,29]))
+
     def test_update_with_bindparams(self):
         User = self.classes.User
 
         eq_(d1.cnt, 2)
         sess.close()
 
+class InheritTest(fixtures.DeclarativeMappedTest):
 
+    run_inserts = 'each'
+
+    run_deletes = 'each'
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class Person(Base):
+            __tablename__ = 'person'
+            id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
+            type = Column(String(50))
+            name = Column(String(50))
+
+        class Engineer(Person):
+            __tablename__ = 'engineer'
+            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
+            engineer_name = Column(String(50))
+
+        class Manager(Person):
+            __tablename__ = 'manager'
+            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
+            manager_name = Column(String(50))
+
+    @classmethod
+    def insert_data(cls):
+        Engineer, Person, Manager = cls.classes.Engineer, \
+                    cls.classes.Person, cls.classes.Manager
+        s = Session(testing.db)
+        s.add_all([
+            Engineer(name='e1', engineer_name='e1'),
+            Manager(name='m1', manager_name='m1'),
+            Engineer(name='e2', engineer_name='e2'),
+            Person(name='p1'),
+        ])
+        s.commit()
+
+    def test_illegal_metadata(self):
+        person = self.classes.Person.__table__
+        engineer = self.classes.Engineer.__table__
+
+        sess = Session()
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "This operation requires only one Table or entity be "
+            "specified as the target.",
+            sess.query(person.join(engineer)).update, {}
+        )
+
+    def test_update_subtable_only(self):
+        Engineer = self.classes.Engineer
+        s = Session(testing.db)
+        s.query(Engineer).update({'engineer_name': 'e5'})
+
+        eq_(
+            s.query(Engineer.engineer_name).all(),
+            [('e5', ), ('e5', )]
+        )
+
+    @testing.requires.update_from
+    def test_update_from(self):
+        Engineer = self.classes.Engineer
+        Person = self.classes.Person
+        s = Session(testing.db)
+        s.query(Engineer).filter(Engineer.id == Person.id).\
+            filter(Person.name == 'e2').update({'engineer_name': 'e5'})
+
+        eq_(
+            set(s.query(Person.name, Engineer.engineer_name)),
+            set([('e1', 'e1', ), ('e2', 'e5')])
+        )
+