.. changelog::
:version: 1.2.0b1
+ .. change:: 4009
+ :tags: feature, mysql
+ :tickets: 4009
+
+ Added support for MySQL's ON DUPLICATE KEY UPDATE
+ MySQL-specific :class:`.mysql.dml.Insert` object.
+ Pull request courtesy Michael Doronin.
+
+ .. seealso::
+
+ :ref:`change_4009`
+
.. change:: 4018
:tags: bug, sql
:tickets: 4018
Dialect Improvements and Changes - MySQL
========================================
+.. _change_4009:
+
+Support for INSERT..ON DUPLICATE KEY UPDATE
+-------------------------------------------
+
+The ``ON DUPLICATE KEY UPDATE`` clause of ``INSERT`` supported by MySQL
+is now supported using a MySQL-specific version of the
+:class:`.Insert` object, via :func:`sqlalchemy.dialects.mysql.dml.insert`.
+This :class:`.Insert` subclass adds a new method
+:meth:`.Insert.on_duplicate_key_update` that implements MySQL's syntax::
+
+ from sqlalchemy.dialect.mysql import insert
+
+ insert_stmt = insert(my_table). \\
+ values(id='some_id', data='some data to insert')
+
+ on_conflict_stmt = insert_stmt.on_duplicate_key_update(
+ data=stmt.values.data,
+ status='U'
+ )
+
+ conn.execute(do_update_stmt)
+
+The above will render::
+
+ INSERT INTO my_table (id, data)
+ VALUES (:id, :data)
+ ON DUPLICATE KEY UPDATE data=VALUES(data), status=:status_1
+
+.. seealso::
+
+ :ref:`mysql_insert_on_duplicate_key_update`
+
+:ticket:`4009`
+
+
Dialect Improvements and Changes - SQLite
=========================================
.. autoclass:: YEAR
:members: __init__
+MySQL DML Constructs
+-------------------------
+
+.. autofunction:: sqlalchemy.dialects.mysql.dml.insert
+
+.. autoclass:: sqlalchemy.dialects.mysql.dml.Insert
+ :members:
+
+
MySQL-Python
------------
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',
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'])
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
----------------
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))
--- /dev/null
+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
DateTime, Time, Date, Interval, NCHAR, CHAR, CLOB, TEXT, Boolean, \
BOOLEAN, LargeBinary, BLOB, SmallInteger, INT, func, cast
+from sqlalchemy.dialects.mysql import insert
from sqlalchemy.dialects.mysql import base as mysql
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
-from sqlalchemy.testing import mock
-from sqlalchemy import testing
from sqlalchemy.sql import table, column
-import re
+from sqlalchemy.sql.expression import literal_column
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
t1.outerjoin(t2, t1.c.x == t2.c.y, full=True),
"t1 FULL OUTER JOIN t2 ON t1.x = t2.y"
)
+
+
+class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = mysql.dialect()
+
+ def setup(self):
+ self.table = Table(
+ 'foos', MetaData(),
+ Column('id', Integer, primary_key=True),
+ Column('bar', String(10)),
+ Column('baz', String(10)),
+ )
+
+ def test_from_values(self):
+ stmt = insert(
+ self.table, [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}])
+ stmt = stmt.on_duplicate_key_update(
+ bar=stmt.values.bar, baz=stmt.values.baz)
+ expected_sql = (
+ 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) '
+ 'ON DUPLICATE KEY UPDATE bar = VALUES(bar), baz = VALUES(baz)'
+ )
+ self.assert_compile(stmt, expected_sql)
+
+ def test_from_literal(self):
+ stmt = insert(
+ self.table, [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}])
+ stmt = stmt.on_duplicate_key_update(bar=literal_column('bb'))
+ expected_sql = (
+ 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) '
+ 'ON DUPLICATE KEY UPDATE bar = bb'
+ )
+ self.assert_compile(stmt, expected_sql)
+
+ def test_python_values(self):
+ stmt = insert(
+ self.table, [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}])
+ stmt = stmt.on_duplicate_key_update(bar="foobar")
+ expected_sql = (
+ 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) '
+ 'ON DUPLICATE KEY UPDATE bar = %s'
+ )
+ self.assert_compile(stmt, expected_sql)
--- /dev/null
+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])
+
+
+
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
