Class OnDuplicateClause

lib/sqlalchemy/dialects/mysql/dml.py:222–274

class OnDuplicateClause(SyntaxExtension, ClauseElement):
    __visit_name__ = "on_duplicate_key_update"

    _parameter_ordering: Optional[List[str]] = None

    update: Dict[str, ColumnElement[Any]]
    stringify_dialect = "mysql"

    _traverse_internals = [
        ("_parameter_ordering", InternalTraversal.dp_string_list),
        ("update", InternalTraversal.dp_dml_values),
    ]

    def __init__(
        self, inserted_alias: NamedFromClause, update: _UpdateArg
    ) -> None:
        self.inserted_alias = inserted_alias

        # auto-detect that parameters should be ordered.   This is copied from
        # Update._proces_colparams(), however we don't look for a special flag
        # in this case since we are not disambiguating from other use cases as
        # we are in Update.values().
        if isinstance(update, list) and (
            update and isinstance(update[0], tuple)
        ):
            self._parameter_ordering = [key for key, value in update]
            update = dict(update)

        if isinstance(update, dict):
            if not update:
                raise ValueError(
                    "update parameter dictionary must not be empty"
                )
        elif isinstance(update, ColumnCollection):
            update = dict(update)
        else:
            raise ValueError(
                "update parameter must be a non-empty dictionary "
                "or a ColumnCollection such as the `.c.` collection "
                "of a Table object"
            )

        self.update = {
            k: coercions.expect(
                roles.ExpressionElementRole, v, type_=NULLTYPE, is_crud=True
            )
            for k, v in update.items()
        }

    def apply_to_insert(self, insert_stmt: StandardInsert) -> None:
        insert_stmt.apply_syntax_extension_point(
            self.append_replacing_same_type, "post_values"
        )


_UpdateArg = Union[