Class OnConflictDoUpdate

lib/sqlalchemy/dialects/sqlite/dml.py:235–279

class OnConflictDoUpdate(OnConflictClause):
    __visit_name__ = "on_conflict_do_update"

    update_values_to_set: Dict[_DMLColumnElement, ColumnElement[Any]]
    update_whereclause: Optional[ColumnElement[Any]]

    _traverse_internals = OnConflictClause._traverse_internals + [
        ("update_values_to_set", InternalTraversal.dp_dml_values),
        ("update_whereclause", InternalTraversal.dp_clauseelement),
    ]

    def __init__(
        self,
        index_elements: _OnConflictIndexElementsT = None,
        index_where: _OnConflictIndexWhereT = None,
        set_: _OnConflictSetT = None,
        where: _OnConflictWhereT = None,
    ):
        super().__init__(
            index_elements=index_elements,
            index_where=index_where,
        )

        if isinstance(set_, dict):
            if not set_:
                raise ValueError("set parameter dictionary must not be empty")
        elif isinstance(set_, ColumnCollection):
            set_ = dict(set_)
        else:
            raise ValueError(
                "set parameter must be a non-empty dictionary "
                "or a ColumnCollection such as the `.c.` collection "
                "of a Table object"
            )
        self.update_values_to_set = {
            coercions.expect(roles.DMLColumnRole, k): coercions.expect(
                roles.ExpressionElementRole, v, type_=NULLTYPE, is_crud=True
            )
            for k, v in set_.items()
        }
        self.update_whereclause = (
            coercions.expect(roles.WhereHavingRole, where)
            if where is not None
            else None
        )