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

3 files changed, 200 insertions, 0 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/__init__.py b/lib/sqlalchemy/dialects/mysql/__init__.py
index 2ff854210..f71127979 100644
--- a/lib/sqlalchemy/dialects/mysql/__init__.py
+++ b/lib/sqlalchemy/dialects/mysql/__init__.py
@@ -21,6 +21,8 @@ from .base import \
     TINYBLOB, TINYINT, TINYTEXT,\
     VARBINARY, VARCHAR, YEAR, dialect
 
+from .dml import insert, Insert
+
 __all__ = (
     'BIGINT', 'BINARY', 'BIT', 'BLOB', 'BOOLEAN', 'CHAR', 'DATE', 'DATETIME',
     'DECIMAL', 'DOUBLE', 'ENUM', 'DECIMAL', 'FLOAT', 'INTEGER', 'INTEGER',
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index c19253478..1781f9bb7 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -307,6 +307,8 @@ And of course any valid MySQL statement can be executed as a string as well.
 Some limited direct support for MySQL extensions to SQL is currently
 available.
 
+* INSERT..ON DUPLICATE KEY UPDATE:  See :ref:`mysql_insert_on_duplicate_key_update`
+
 * SELECT pragma::
 
     select(..., prefixes=['HIGH_PRIORITY', 'SQL_SMALL_RESULT'])
@@ -315,6 +317,87 @@ available.
 
     update(..., mysql_limit=10)
 
+.. _mysql_insert_on_duplicate_key_update:
+
+INSERT...ON DUPLICATE KEY UPDATE (Upsert)
+------------------------------------------
+
+MySQL allows "upserts" (update or insert)
+of rows into a table via the ``ON DUPLICATE KEY UPDATE`` clause of the
+``INSERT`` statement.  A candidate row will only be inserted if that row does
+not match an existing primary or unique key in the table; otherwise, an UPDATE will
+be performed.   The statement allows for separate specification of the
+values to INSERT versus the values for UPDATE.
+
+SQLAlchemy provides ``ON DUPLICATE KEY UPDATE`` support via the MySQL-specific
+:func:`.mysql.dml.insert()` function, which provides
+the generative method :meth:`~.mysql.dml.Insert.on_duplicate_key_update`::
+
+    from sqlalchemy.dialects.mysql import insert
+
+    insert_stmt = insert(my_table).values(
+        id='some_existing_id',
+        data='inserted value')
+
+    on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
+        data=insert_stmt.values.data,
+        status='U'
+    )
+
+    conn.execute(on_duplicate_key_stmt)
+
+Unlike Postgresql's "ON CONFLICT" phrase, the "ON DUPLICATE KEY UPDATE"
+phrase will always match on any primary key or unique key, and will always
+perform an UPDATE if there's a match; there are no options for it to raise
+an error or to skip performing an UPDATE.
+
+``ON DUPLICATE KEY UPDATE`` is used to perform an update of the already
+existing row, using any combination of new values as well as values
+from the proposed insertion.   These values are specified using
+keyword arguments passed to the
+:meth:`~.mysql.dml.Insert.on_duplicate_key_update`
+given column key values (usually the name of the column, unless it
+specifies :paramref:`.Column.key`) as keys and literal or SQL expressions
+as values::
+
+    on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
+        data="some data"
+        updated_at=func.current_timestamp()
+    )
+
+.. warning::
+
+    The :meth:`.Insert.on_duplicate_key_update` method does **not** take into
+    account Python-side default UPDATE values or generation functions, e.g.
+    e.g. those specified using :paramref:`.Column.onupdate`.
+    These values will not be exercised for an ON DUPLICATE KEY style of UPDATE,
+    unless they are manually specified explicitly in the parameters.
+
+In order to refer to the proposed insertion row, the special alias
+:attr:`~.mysql.dml.Insert.values` is available as an attribute on
+the :class:`.mysql.dml.Insert` object; this object is a
+:class:`.ColumnCollection` which contains all columns of the target
+table::
+
+    from sqlalchemy.dialects.mysql import insert
+
+    stmt = insert(my_table).values(
+        id='some_id',
+        data='inserted value',
+        author='jlh')
+    do_update_stmt = stmt.on_duplicate_key_update(
+        data="updated value",
+        author=stmt.values.author
+    )
+    conn.execute(do_update_stmt)
+
+When rendered, the "values" namespace will produce the expression
+``VALUES(<columnname>)``.
+
+.. versionadded:: 1.2 Added support for MySQL ON DUPLICATE KEY UPDATE clause
+
+
+
 rowcount Support
 ----------------
 
@@ -814,6 +897,42 @@ class MySQLCompiler(compiler.SQLCompiler):
             self.process(binary.left, **kw),
             self.process(binary.right, **kw))
 
+    def visit_on_duplicate_key_update(self, on_duplicate, **kw):
+        cols = self.statement.table.c
+
+        clauses = []
+        # traverse in table column order
+        for column in cols:
+            val = on_duplicate.update.get(column.key)
+            if val is None:
+                continue
+            elif elements._is_literal(val):
+                val = elements.BindParameter(None, val, type_=column.type)
+                value_text = self.process(val.self_group(), use_schema=False)
+            elif isinstance(val, elements.BindParameter) and val.type._isnull:
+                val = val._clone()
+                val.type = column.type
+                value_text = self.process(val.self_group(), use_schema=False)
+            elif isinstance(val, elements.ColumnClause) \
+                    and val.table is on_duplicate.values_alias:
+                value_text = 'VALUES(' + self.preparer.quote(column.name) + ')'
+            else:
+                value_text = self.process(val.self_group(), use_schema=False)
+            name_text = self.preparer.quote(column.name)
+            clauses.append("%s = %s" % (name_text, value_text))
+
+        non_matching = set(on_duplicate.update) - set(cols.keys())
+        if non_matching:
+            util.warn(
+                'Additional column names not matching '
+                "any column keys in table '%s': %s" % (
+                    self.statement.table.name,
+                    (', '.join("'%s'" % c for c in non_matching))
+                )
+            )
+
+        return 'ON DUPLICATE KEY UPDATE ' + ', '.join(clauses)
+
     def visit_concat_op_binary(self, binary, operator, **kw):
         return "concat(%s, %s)" % (self.process(binary.left),
                                    self.process(binary.right))
diff --git a/lib/sqlalchemy/dialects/mysql/dml.py b/lib/sqlalchemy/dialects/mysql/dml.py
new file mode 100644
index 000000000..743510317
--- /dev/null
+++ b/lib/sqlalchemy/dialects/mysql/dml.py
@@ -0,0 +1,79 @@
+from ...sql.elements import ClauseElement
+from ...sql.dml import Insert as StandardInsert
+from ...sql.expression import alias
+from ...util.langhelpers import public_factory
+from ...sql.base import _generative
+from ... import util
+
+__all__ = ('Insert', 'insert')
+
+
+class Insert(StandardInsert):
+    """MySQL-specific implementation of INSERT.
+
+    Adds methods for MySQL-specific syntaxes such as ON DUPLICATE KEY UPDATE.
+
+    .. versionadded:: 1.2
+
+    """
+
+    @property
+    def values(self):
+        """Provide the ``values`` namespace for an ON DUPLICATE KEY UPDATE statement
+
+        MySQL's ON DUPLICATE KEY UPDATE clause allows reference to the row
+        that would be inserted, via a special function called ``VALUES()``.
+        This attribute provides all columns in this row to be referenaceable
+        such that they will render within a ``VALUES()`` function inside the
+        ON DUPLICATE KEY UPDATE clause.
+
+        .. seealso::
+
+            :ref:`mysql_insert_on_duplicate_key_update` - example of how
+            to use :attr:`.Insert.values`
+
+        """
+        return self.values_alias.columns
+
+    @util.memoized_property
+    def values_alias(self):
+        return alias(self.table, name='values')
+
+    @_generative
+    def on_duplicate_key_update(self, **kw):
+        r"""
+        Specifies the ON DUPLICATE KEY UPDATE clause.
+
+        :param \**kw:  Column keys linked to UPDATE values.  The
+         values may be any SQL expression or supported literal Python
+         values.
+
+        .. warning:: This dictionary does **not** take into account
+           Python-specified default UPDATE values or generation functions,
+           e.g. those specified using :paramref:`.Column.onupdate`.
+           These values will not be exercised for an ON DUPLICATE KEY UPDATE
+           style of UPDATE, unless values are manually specified here.
+
+        .. versionadded:: 1.2
+
+        .. seealso::
+
+            :ref:`mysql_insert_on_duplicate_key_update`
+
+        """
+        values_alias = getattr(self, 'values_alias', None)
+        self._post_values_clause = OnDuplicateClause(values_alias, kw)
+        return self
+
+
+insert = public_factory(Insert, '.dialects.mysql.insert')
+
+
+class OnDuplicateClause(ClauseElement):
+    __visit_name__ = 'on_duplicate_key_update'
+
+    def __init__(self, values_alias, update):
+        self.values_alias = values_alias
+        if not update or not isinstance(update, dict):
+            raise ValueError('update parameter must be a non-empty dictionary')
+        self.update = update