Implement MySQL's ON DUPLICATE KEY UPDATE

Added support for MySQL's ON DUPLICATE KEY UPDATE
MySQL-specific :class:`.mysql.dml.Insert` object.
Pull request courtesy Michael Doronin.

Co-authored-by: Mike Bayer <mike_mp@zzzcomputing.com>
Resolves: #4009
Change-Id: Ic71424f3c88af6082b48a910a2efb7fbfc0a7eb4
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/365

1 files changed, 59 insertions, 0 deletions

diff --git a/test/dialect/mysql/test_on_duplicate.py b/test/dialect/mysql/test_on_duplicate.py
new file mode 100644
index 000000000..2ff4a58cc
--- /dev/null
+++ b/test/dialect/mysql/test_on_duplicate.py
@@ -0,0 +1,59 @@
+from sqlalchemy.testing.assertions import eq_, assert_raises
+from sqlalchemy.testing import fixtures
+from sqlalchemy import testing
+from sqlalchemy.dialects.mysql import insert
+from sqlalchemy import Table, Column, Integer, String
+
+
+class OnDuplicateTest(fixtures.TablesTest):
+    __only_on__ = 'mysql',
+    __backend__ = True
+    run_define_tables = 'each'
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'foos', metadata,
+            Column('id', Integer, primary_key=True, autoincrement=True),
+            Column('bar', String(10)),
+            Column('baz', String(10)),
+        )
+
+    def test_bad_args(self):
+        assert_raises(
+            ValueError,
+            insert(self.tables.foos, values={}).on_duplicate_key_update
+        )
+
+    def test_on_duplicate_key_update(self):
+        foos = self.tables.foos
+        with testing.db.connect() as conn:
+            conn.execute(insert(foos, dict(id=1, bar='b', baz='bz')))
+            stmt = insert(foos, [dict(id=1, bar='ab'), dict(id=2, bar='b')])
+            stmt = stmt.on_duplicate_key_update(bar=stmt.values.bar)
+            result = conn.execute(stmt)
+            eq_(result.inserted_primary_key, [2])
+            eq_(
+                conn.execute(foos.select().where(foos.c.id == 1)).fetchall(),
+                [(1, 'ab', 'bz')]
+            )
+
+    def test_last_inserted_id(self):
+        foos = self.tables.foos
+        with testing.db.connect() as conn:
+            stmt = insert(foos, {"bar": "b", "baz": "bz"})
+            result = conn.execute(
+                stmt.on_duplicate_key_update(
+                    bar=stmt.values.bar, baz="newbz")
+            )
+            eq_(result.inserted_primary_key, [1])
+
+            stmt = insert(foos, {"id": 1, "bar": "b", "baz": "bz"})
+            result = conn.execute(
+                stmt.on_duplicate_key_update(
+                    bar=stmt.values.bar, baz="newbz")
+            )
+            eq_(result.inserted_primary_key, [1])
+
+
+