path: root/test/sql/test_update.py

import itertools
import random

from sqlalchemy import bindparam
from sqlalchemy import column
from sqlalchemy import exc
from sqlalchemy import exists
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import MetaData
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import update
from sqlalchemy import util
from sqlalchemy.dialects import mysql
from sqlalchemy.engine import default
from sqlalchemy.sql import operators
from sqlalchemy.sql.elements import BooleanClauseList
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table


class _UpdateFromTestBase:
    @classmethod
    def define_tables(cls, metadata):
        Table(
            "mytable",
            metadata,
            Column("myid", Integer),
            Column("name", String(30)),
            Column("description", String(50)),
        )
        Table(
            "myothertable",
            metadata,
            Column("otherid", Integer),
            Column("othername", String(30)),
        )
        Table(
            "users",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("name", String(30), nullable=False),
        )
        Table(
            "addresses",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("user_id", None, ForeignKey("users.id")),
            Column("name", String(30), nullable=False),
            Column("email_address", String(50), nullable=False),
        )
        Table(
            "dingalings",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("address_id", None, ForeignKey("addresses.id")),
            Column("data", String(30)),
        )
        Table(
            "update_w_default",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x", Integer),
            Column("ycol", Integer, key="y"),
            Column("data", String(30), onupdate=lambda: "hi"),
        )

    @classmethod
    def fixtures(cls):
        return dict(
            users=(
                ("id", "name"),
                (7, "jack"),
                (8, "ed"),
                (9, "fred"),
                (10, "chuck"),
            ),
            addresses=(
                ("id", "user_id", "name", "email_address"),
                (1, 7, "x", "jack@bean.com"),
                (2, 8, "x", "ed@wood.com"),
                (3, 8, "x", "ed@bettyboop.com"),
                (4, 8, "x", "ed@lala.com"),
                (5, 9, "x", "fred@fred.com"),
            ),
            dingalings=(
                ("id", "address_id", "data"),
                (1, 2, "ding 1/2"),
                (2, 5, "ding 2/5"),
            ),
        )


class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
    __dialect__ = "default_enhanced"

    @testing.variation("twotable", [True, False])
    @testing.variation("values", ["none", "blank"])
    def test_update_no_params(self, values, twotable):
        """test issue identified while doing #9721

        UPDATE with empty VALUES but multiple tables would raise a
        NoneType error; fixed this to emit an empty "SET" the way a single
        table UPDATE currently does.

        both cases should probably raise CompileError, however this could
        be backwards incompatible with current use cases (such as other test
        suites)

        """

        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        stmt = table1.update().where(table1.c.name == "jill")
        if twotable:
            stmt = stmt.where(table2.c.otherid == table1.c.myid)

        if values.blank:
            stmt = stmt.values()

        if twotable:
            if values.blank:
                self.assert_compile(
                    stmt,
                    "UPDATE mytable SET  FROM myothertable "
                    "WHERE mytable.name = :name_1 "
                    "AND myothertable.otherid = mytable.myid",
                )
            elif values.none:
                self.assert_compile(
                    stmt,
                    "UPDATE mytable SET myid=:myid, name=:name, "
                    "description=:description FROM myothertable "
                    "WHERE mytable.name = :name_1 "
                    "AND myothertable.otherid = mytable.myid",
                )
        elif values.blank:
            self.assert_compile(
                stmt,
                "UPDATE mytable SET  WHERE mytable.name = :name_1",
            )
        elif values.none:
            self.assert_compile(
                stmt,
                "UPDATE mytable SET myid=:myid, name=:name, "
                "description=:description WHERE mytable.name = :name_1",
            )

    def test_update_literal_binds(self):
        table1 = self.tables.mytable

        stmt = (
            table1.update().values(name="jack").where(table1.c.name == "jill")
        )

        self.assert_compile(
            stmt,
            "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'",
            literal_binds=True,
        )

    def test_update_custom_key_thing(self):
        table1 = self.tables.mytable

        class Thing:
            def __clause_element__(self):
                return table1.c.name

        stmt = (
            table1.update()
            .values({Thing(): "jack"})
            .where(table1.c.name == "jill")
        )

        self.assert_compile(
            stmt,
            "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'",
            literal_binds=True,
        )

    def test_update_ordered_custom_key_thing(self):
        table1 = self.tables.mytable

        class Thing:
            def __clause_element__(self):
                return table1.c.name

        stmt = (
            table1.update()
            .ordered_values((Thing(), "jack"))
            .where(table1.c.name == "jill")
        )

        self.assert_compile(
            stmt,
            "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'",
            literal_binds=True,
        )

    def test_update_broken_custom_key_thing(self):
        table1 = self.tables.mytable

        class Thing:
            def __clause_element__(self):
                return 5

        assert_raises_message(
            exc.ArgumentError,
            "SET/VALUES column expression or string key expected, got .*Thing",
            table1.update().values,
            {Thing(): "jack"},
        )

    def test_update_ordered_broken_custom_key_thing(self):
        table1 = self.tables.mytable

        class Thing:
            def __clause_element__(self):
                return 5

        assert_raises_message(
            exc.ArgumentError,
            "SET/VALUES column expression or string key expected, got .*Thing",
            table1.update().ordered_values,
            (Thing(), "jack"),
        )

    def test_correlated_update_one(self):
        table1 = self.tables.mytable

        # test against a straight text subquery
        u = update(table1).values(
            {
                table1.c.name: text(
                    "(select name from mytable where id=mytable.id)"
                )
            }
        )
        self.assert_compile(
            u,
            "UPDATE mytable SET name=(select name from mytable "
            "where id=mytable.id)",
        )

    def test_correlated_update_two(self):
        table1 = self.tables.mytable

        mt = table1.alias()
        u = update(table1).values(
            {
                table1.c.name: select(mt.c.name)
                .where(mt.c.myid == table1.c.myid)
                .scalar_subquery()
            }
        )
        self.assert_compile(
            u,
            "UPDATE mytable SET name=(SELECT mytable_1.name FROM "
            "mytable AS mytable_1 WHERE "
            "mytable_1.myid = mytable.myid)",
        )

    def test_correlated_update_three(self):
        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        # test against a regular constructed subquery
        s = (
            select(table2)
            .where(table2.c.otherid == table1.c.myid)
            .scalar_subquery()
        )
        u = (
            update(table1)
            .where(table1.c.name == "jack")
            .values({table1.c.name: s})
        )
        self.assert_compile(
            u,
            "UPDATE mytable SET name=(SELECT myothertable.otherid, "
            "myothertable.othername FROM myothertable WHERE "
            "myothertable.otherid = mytable.myid) "
            "WHERE mytable.name = :name_1",
        )

    def test_correlated_update_four(self):
        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        # test a non-correlated WHERE clause
        s = select(table2.c.othername).where(table2.c.otherid == 7)
        u = update(table1).where(table1.c.name == s.scalar_subquery())
        self.assert_compile(
            u,
            "UPDATE mytable SET myid=:myid, name=:name, "
            "description=:description WHERE mytable.name = "
            "(SELECT myothertable.othername FROM myothertable "
            "WHERE myothertable.otherid = :otherid_1)",
        )

    def test_correlated_update_five(self):
        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        # test one that is actually correlated...
        s = select(table2.c.othername).where(table2.c.otherid == table1.c.myid)
        u = table1.update().where(table1.c.name == s.scalar_subquery())
        self.assert_compile(
            u,
            "UPDATE mytable SET myid=:myid, name=:name, "
            "description=:description WHERE mytable.name = "
            "(SELECT myothertable.othername FROM myothertable "
            "WHERE myothertable.otherid = mytable.myid)",
        )

    def test_correlated_update_six(self):
        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        # test correlated FROM implicit in WHERE and SET clauses
        u = (
            table1.update()
            .values(name=table2.c.othername)
            .where(table2.c.otherid == table1.c.myid)
        )
        self.assert_compile(
            u,
            "UPDATE mytable SET name=myothertable.othername "
            "FROM myothertable WHERE myothertable.otherid = mytable.myid",
        )

    def test_correlated_update_seven(self):
        table1 = self.tables.mytable
        table2 = self.tables.myothertable

        u = (
            table1.update()
            .values(name="foo")
            .where(table2.c.otherid == table1.c.myid)
        )

        # this is the "default_enhanced" compiler.  there's no UPDATE FROM
        # in the base compiler.
        # See also test/dialect/mssql/test_compiler->test_update_from().
        self.assert_compile(
            u,
            "UPDATE mytable SET name=:name "
            "FROM myothertable WHERE myothertable.otherid = mytable.myid",
        )

    def test_binds_that_match_columns(self):
        """test bind params named after column names
        replace the normal SET/VALUES generation.

        See also test_compiler.py::CrudParamOverlapTest

        """

        t = table("foo", column("x"), column("y"))

        u = t.update().where(t.c.x == bindparam("x"))

        assert_raises(exc.CompileError, u.compile)

        self.assert_compile(u, "UPDATE foo SET  WHERE foo.x = :x", params={})

        assert_raises(exc.CompileError, u.values(x=7).compile)

        self.assert_compile(
            u.values(y=7), "UPDATE foo SET y=:y WHERE foo.x = :x"
        )

        assert_raises(
            exc.CompileError, u.values(x=7).compile, column_keys=["x", "y"]
        )
        assert_raises(exc.CompileError, u.compile, column_keys=["x", "y"])

        self.assert_compile(
            u.values(x=3 + bindparam("x")),
            "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x",
        )

        self.assert_compile(
            u.values(x=3 + bindparam("x")),
            "UPDATE foo SET x=(:param_1 + :x) WHERE foo.x = :x",
            params={"x": 1},
        )

        self.assert_compile(
            u.values(x=3 + bindparam("x")),
            "UPDATE foo SET x=(:param_1 + :x), y=:y WHERE foo.x = :x",
            params={"x": 1, "y": 2},
        )

    def test_labels_no_collision(self):

        t = table("foo", column("id"), column("foo_id"))

        self.assert_compile(
            t.update().where(t.c.id == 5),
            "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :id_1",
        )

        self.assert_compile(
            t.update().where(t.c.id == bindparam(key=t.c.id._label)),
            "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :foo_id_1",
        )

    def test_labels_no_collision_index(self):
        """test for [ticket:4911]"""

        t = Table(
            "foo",
            MetaData(),
            Column("id", Integer, index=True),
            Column("foo_id", Integer),
        )

        self.assert_compile(
            t.update().where(t.c.id == 5),
            "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :id_1",
        )

        self.assert_compile(
            t.update().where(t.c.id == bindparam(key=t.c.id._label)),
            "UPDATE foo SET id=:id, foo_id=:foo_id WHERE foo.id = :foo_id_1",
        )

    def test_inline_defaults(self):
        m = MetaData()
        foo = Table("foo", m, Column("id", Integer))

        t = Table(
            "test",
            m,
            Column("col1", Integer, onupdate=func.foo(1)),
            Column(
                "col2",
                Integer,
                onupdate=select(func.coalesce(func.max(foo.c.id))),
            ),
            Column("col3", String(30)),
        )

        self.assert_compile(
            t.update().values({"col3": "foo"}),
            "UPDATE test SET col1=foo(:foo_1), col2=(SELECT "
            "coalesce(max(foo.id)) AS coalesce_1 FROM foo), "
            "col3=:col3",
        )

        self.assert_compile(
            t.update().inline().values({"col3": "foo"}),
            "UPDATE test SET col1=foo(:foo_1), col2=(SELECT "
            "coalesce(max(foo.id)) AS coalesce_1 FROM foo), "
            "col3=:col3",
        )

    def test_update_1(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1).where(table1.c.myid == 7),
            "UPDATE mytable SET name=:name WHERE mytable.myid = :myid_1",
            params={table1.c.name: "fred"},
        )

    def test_update_2(self):
        table1 = self.tables.mytable

        self.assert_compile(
            table1.update()
            .where(table1.c.myid == 7)
            .values({table1.c.myid: 5}),
            "UPDATE mytable SET myid=:myid WHERE mytable.myid = :myid_1",
            checkparams={"myid": 5, "myid_1": 7},
        )

    def test_update_3(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1).where(table1.c.myid == 7),
            "UPDATE mytable SET name=:name WHERE mytable.myid = :myid_1",
            params={"name": "fred"},
        )

    def test_update_4(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1).values({table1.c.name: table1.c.myid}),
            "UPDATE mytable SET name=mytable.myid",
        )

    def test_update_5(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1)
            .where(table1.c.name == bindparam("crit"))
            .values(
                {table1.c.name: "hi"},
            ),
            "UPDATE mytable SET name=:name WHERE mytable.name = :crit",
            params={"crit": "notthere"},
            checkparams={"crit": "notthere", "name": "hi"},
        )

    def test_update_6(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1)
            .where(table1.c.myid == 12)
            .values(
                {table1.c.name: table1.c.myid},
            ),
            "UPDATE mytable "
            "SET name=mytable.myid, description=:description "
            "WHERE mytable.myid = :myid_1",
            params={"description": "test"},
            checkparams={"description": "test", "myid_1": 12},
        )

    def test_update_7(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1)
            .where(table1.c.myid == 12)
            .values({table1.c.myid: 9}),
            "UPDATE mytable "
            "SET myid=:myid, description=:description "
            "WHERE mytable.myid = :myid_1",
            params={"myid_1": 12, "myid": 9, "description": "test"},
        )

    def test_update_8(self):
        table1 = self.tables.mytable

        self.assert_compile(
            update(table1).where(table1.c.myid == 12),
            "UPDATE mytable SET myid=:myid WHERE mytable.myid = :myid_1",
            params={"myid": 18},
            checkparams={"myid": 18, "myid_1": 12},
        )

    def test_update_9(self):
        table1 = self.tables.mytable

        s = (
            table1.update()
            .where(table1.c.myid == 12)
            .values({table1.c.name: "lala"})
        )
        c = s.compile(column_keys=["id", "name"])
        eq_(str(s), str(c))

    def test_update_10(self):
        table1 = self.tables.mytable

        v1 = {table1.c.name: table1.c.myid}
        v2 = {table1.c.name: table1.c.name + "foo"}
        self.assert_compile(
            update(table1).where(table1.c.myid == 12).values(v1).values(v2),
            "UPDATE mytable "
            "SET "
            "name=(mytable.name || :name_1), "
            "description=:description "
            "WHERE mytable.myid = :myid_1",
            params={"description": "test"},
        )

    def test_update_11(self):
        table1 = self.tables.mytable

        values = {
            table1.c.name: table1.c.name + "lala",
            table1.c.myid: func.do_stuff(table1.c.myid, literal("hoho")),
        }

        self.assert_compile(
            update(table1)
            .where(
                (table1.c.myid == func.hoho(4))
                & (
                    table1.c.name
                    == literal("foo") + table1.c.name + literal("lala")
                )
            )
            .values(values),
            "UPDATE mytable "
            "SET "
            "myid=do_stuff(mytable.myid, :param_1), "
            "name=(mytable.name || :name_1) "
            "WHERE "
            "mytable.myid = hoho(:hoho_1) AND "
            "mytable.name = :param_2 || mytable.name || :param_3",
        )

    def test_unconsumed_names_kwargs(self):
        t = table("t", column("x"), column("y"))

        assert_raises_message(
            exc.CompileError,
            "Unconsumed column names: z",
            t.update().values(x=5, z=5).compile,
        )

    def test_unconsumed_names_values_dict(self):
        t = table("t", column("x"), column("y"))
        t2 = table("t2", column("q"), column("z"))

        assert_raises_message(
            exc.CompileError,
            "Unconsumed column names: j",
            t.update()
            .values(x=5, j=7)
            .values({t2.c.z: 5})
            .where(t.c.x == t2.c.q)
            .compile,
        )

    def test_unconsumed_names_kwargs_w_keys(self):
        t = table("t", column("x"), column("y"))

        assert_raises_message(
            exc.CompileError,
            "Unconsumed column names: j",
            t.update().values(x=5, j=7).compile,
            column_keys=["j"],
        )

    def test_update_ordered_parameters_newstyle_1(self):
        table1 = self.tables.mytable

        # Confirm that we can pass values as list value pairs
        # note these are ordered *differently* from table.c
        values = [
            (table1.c.name, table1.c.name + "lala"),
            (table1.c.myid, func.do_stuff(table1.c.myid, literal("hoho"))),
        ]
        self.assert_compile(
            update(table1)
            .where(
                (table1.c.myid == func.hoho(4))
                & (
                    table1.c.name
                    == literal("foo") + table1.c.name + literal("lala")
                )
            )
            .ordered_values(*values),
            "UPDATE mytable "
            "SET "
            "name=(mytable.name || :name_1), "
            "myid=do_stuff(mytable.myid, :param_1) "
            "WHERE "
            "mytable.myid = hoho(:hoho_1) AND "
            "mytable.name = :param_2 || mytable.name || :param_3",
        )

    def test_update_ordered_parameters_newstyle_2(self):
        table1 = self.tables.mytable

        # Confirm that we can pass values as list value pairs
        # note these are ordered *differently* from table.c
        values = [
            (table1.c.name, table1.c.name + "lala"),
            ("description", "some desc"),
            (table1.c.myid, func.do_stuff(table1.c.myid, literal("hoho"))),
        ]
        self.assert_compile(
            update(table1)
            .where(
                (table1.c.myid == func.hoho(4))
                & (
                    table1.c.name
                    == literal("foo") + table1.c.name + literal("lala")
                ),
            )
            .ordered_values(*values),
            "UPDATE mytable "
            "SET "
            "name=(mytable.name || :name_1), "
            "description=:description, "
            "myid=do_stuff(mytable.myid, :param_1) "
            "WHERE "
            "mytable.myid = hoho(:hoho_1) AND "
            "mytable.name = :param_2 || mytable.name || :param_3",
        )

    def test_update_ordered_parameters_multiple(self):
        table1 = self.tables.mytable

        stmt = update(table1)

        stmt = stmt.ordered_values(("name", "somename"))

        assert_raises_message(
            exc.ArgumentError,
            "This statement already has ordered values present",
            stmt.ordered_values,
            ("myid", 10),
        )

    def test_update_ordered_then_nonordered(self):
        table1 = self.tables.mytable

        stmt = table1.update().ordered_values(("myid", 1), ("name", "d1"))

        assert_raises_message(
            exc.InvalidRequestError,
            "This statement already has ordered values present",
            stmt.values,
            {"myid": 2, "name": "d2"},
        )

    def test_update_no_multiple_parameters_allowed(self):
        table1 = self.tables.mytable

        stmt = table1.update().values(
            [{"myid": 1, "name": "n1"}, {"myid": 2, "name": "n2"}]
        )

        assert_raises_message(
            exc.InvalidRequestError,
            "UPDATE construct does not support multiple parameter sets.",
            stmt.compile,
        )

    def test_update_ordereddict(self):
        table1 = self.tables.mytable

        # Confirm that ordered dicts are treated as normal dicts,
        # columns sorted in table order
        values = util.OrderedDict(
            (
                (table1.c.name, table1.c.name + "lala"),
                (table1.c.myid, func.do_stuff(table1.c.myid, literal("hoho"))),
            )
        )

        self.assert_compile(
            update(table1)
            .where(
                (table1.c.myid == func.hoho(4))
                & (
                    table1.c.name
                    == literal("foo") + table1.c.name + literal("lala")
                ),
            )
            .values(values),
            "UPDATE mytable "
            "SET "
            "myid=do_stuff(mytable.myid, :param_1), "
            "name=(mytable.name || :name_1) "
            "WHERE "
            "mytable.myid = hoho(:hoho_1) AND "
            "mytable.name = :param_2 || mytable.name || :param_3",
        )

    def test_where_empty(self):
        table1 = self.tables.mytable
        self.assert_compile(
            table1.update().where(
                BooleanClauseList._construct_raw(operators.and_)
            ),
            "UPDATE mytable SET myid=:myid, name=:name, "
            "description=:description",
        )
        self.assert_compile(
            table1.update().where(
                BooleanClauseList._construct_raw(operators.or_)
            ),
            "UPDATE mytable SET myid=:myid, name=:name, "
            "description=:description",
        )

    def test_prefix_with(self):
        table1 = self.tables.mytable

        stmt = (
            table1.update()
            .prefix_with("A", "B", dialect="mysql")
            .prefix_with("C", "D")
        )

        self.assert_compile(
            stmt,
            "UPDATE C D mytable SET myid=:myid, name=:name, "
            "description=:description",
        )

        self.assert_compile(
            stmt,
            "UPDATE A B C D mytable SET myid=%s, name=%s, description=%s",
            dialect=mysql.dialect(),
        )

    def test_update_to_expression_one(self):
        """test update from an expression.

        this logic is triggered currently by a left side that doesn't
        have a key.  The current supported use case is updating the index
        of a PostgreSQL ARRAY type.

        """
        table1 = self.tables.mytable
        expr = func.foo(table1.c.myid)
        eq_(expr.key, None)
        self.assert_compile(
            table1.update().values({expr: "bar"}),
            "UPDATE mytable SET foo(myid)=:param_1",
        )

    @testing.fixture
    def randomized_param_order_update(self):
        from sqlalchemy.sql.dml import UpdateDMLState

        super_process_ordered_values = UpdateDMLState._process_ordered_values

        # this fixture is needed for Python 3.6 and above to work around
        # dictionaries being insert-ordered.  in python 2.7 the previous
        # logic fails pretty easily without this fixture.
        def _process_ordered_values(self, statement):
            super_process_ordered_values(self, statement)

            tuples = list(self._dict_parameters.items())
            random.shuffle(tuples)
            self._dict_parameters = dict(tuples)

        dialect = default.StrCompileDialect()
        dialect.paramstyle = "qmark"
        dialect.positional = True

        with mock.patch.object(
            UpdateDMLState, "_process_ordered_values", _process_ordered_values
        ):
            yield

    def random_update_order_parameters():
        from sqlalchemy import ARRAY

        t = table(
            "foo",
            column("data1", ARRAY(Integer)),
            column("data2", ARRAY(Integer)),
            column("data3", ARRAY(Integer)),
            column("data4", ARRAY(Integer)),
        )

        idx_to_value = [
            (t.c.data1, 5, 7),
            (t.c.data2, 10, 18),
            (t.c.data3, 8, 4),
            (t.c.data4, 12, 14),
        ]

        def combinations():
            while True:
                random.shuffle(idx_to_value)
                yield list(idx_to_value)

        return testing.combinations(
            *[
                (t, combination)
                for i, combination in zip(range(10), combinations())
            ],
            argnames="t, idx_to_value",
        )

    @random_update_order_parameters()
    def test_update_to_expression_two(
        self, randomized_param_order_update, t, idx_to_value
    ):
        """test update from an expression.

        this logic is triggered currently by a left side that doesn't
        have a key.  The current supported use case is updating the index
        of a PostgreSQL ARRAY type.

        """

        dialect = default.StrCompileDialect()
        dialect.paramstyle = "qmark"
        dialect.positional = True

        stmt = t.update().ordered_values(
            *[(col[idx], val) for col, idx, val in idx_to_value]
        )

        self.assert_compile(
            stmt,
            "UPDATE foo SET %s"
            % (
                ", ".join(
                    "%s[?]=?" % col.key for col, idx, val in idx_to_value
                )
            ),
            dialect=dialect,
            checkpositional=tuple(
                itertools.chain.from_iterable(
                    (idx, val) for col, idx, val in idx_to_value
                )
            ),
        )

    def test_update_to_expression_three(self):
        # this test is from test_defaults but exercises a particular
        # parameter ordering issue
        metadata = MetaData()

        q = Table(
            "q",
            metadata,
            Column("x", Integer, default=2),
            Column("y", Integer, onupdate=5),
            Column("z", Integer),
        )

        p = Table(
            "p",
            metadata,
            Column("s", Integer),
            Column("t", Integer),
            Column("u", Integer, onupdate=1),
        )

        cte = (
            q.update().where(q.c.z == 1).values(x=7).returning(q.c.z).cte("c")
        )
        stmt = select(p.c.s, cte.c.z).where(p.c.s == cte.c.z)

        dialect = default.StrCompileDialect()
        dialect.paramstyle = "qmark"
        dialect.positional = True

        self.assert_compile(
            stmt,
            "WITH c AS (UPDATE q SET x=?, y=? WHERE q.z = ? RETURNING q.z) "
            "SELECT p.s, c.z FROM p, c WHERE p.s = c.z",
            checkpositional=(7, None, 1),
            dialect=dialect,
        )

    @testing.variation("paramstyle", ["qmark", "format", "numeric"])
    def test_update_bound_ordering(self, paramstyle):
        """test that bound parameters between the UPDATE and FROM clauses
        order correctly in different SQL compilation scenarios.

        """
        table1 = self.tables.mytable
        table2 = self.tables.myothertable
        sel = select(table2).where(table2.c.otherid == 5).alias()
        upd = (
            table1.update()
            .where(table1.c.name == sel.c.othername)
            .values(name="foo")
        )

        if paramstyle.qmark:

            dialect = default.StrCompileDialect(paramstyle="qmark")
            self.assert_compile(
                upd,
                "UPDATE mytable SET name=? FROM (SELECT "
                "myothertable.otherid AS otherid, "
                "myothertable.othername AS othername "
                "FROM myothertable "
                "WHERE myothertable.otherid = ?) AS anon_1 "
                "WHERE mytable.name = anon_1.othername",
                checkpositional=("foo", 5),
                dialect=dialect,
            )
        elif paramstyle.format:
            self.assert_compile(
                upd,
                "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
                "myothertable.othername AS othername "
                "FROM myothertable "
                "WHERE myothertable.otherid = %s) AS anon_1 "
                "SET mytable.name=%s "
                "WHERE mytable.name = anon_1.othername",
                checkpositional=(5, "foo"),
                dialect=mysql.dialect(),
            )
        elif paramstyle.numeric:
            dialect = default.StrCompileDialect(paramstyle="numeric")
            self.assert_compile(
                upd,
                "UPDATE mytable SET name=:1 FROM (SELECT "
                "myothertable.otherid AS otherid, "
                "myothertable.othername AS othername "
                "FROM myothertable "
                "WHERE myothertable.otherid = :2) AS anon_1 "
                "WHERE mytable.name = anon_1.othername",
                checkpositional=("foo", 5),
                dialect=dialect,
            )
        else:
            paramstyle.fail()


class UpdateFromCompileTest(
    _UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL
):
    __dialect__ = "default_enhanced"

    run_create_tables = run_inserts = run_deletes = None

    def test_alias_one(self):
        table1 = self.tables.mytable
        talias1 = table1.alias("t1")

        # this case is nonsensical.  the UPDATE is entirely
        # against the alias, but we name the table-bound column
        # in values.   The behavior here isn't really defined
        self.assert_compile(
            update(talias1)
            .where(talias1.c.myid == 7)
            .values({table1.c.name: "fred"}),
            "UPDATE mytable AS t1 "
            "SET name=:name "
            "WHERE t1.myid = :myid_1",
        )

    def test_alias_two(self):
        table1 = self.tables.mytable
        talias1 = table1.alias("t1")

        # Here, compared to
        # test_alias_one(), here we actually have UPDATE..FROM,
        # which is causing the "table1.c.name" param to be handled
        # as an "extra table", hence we see the full table name rendered.
        self.assert_compile(
            update(talias1)
            .where(table1.c.myid == 7)
            .values({table1.c.name: "fred"}),
            "UPDATE mytable AS t1 "
            "SET name=:mytable_name "
            "FROM mytable "
            "WHERE mytable.myid = :myid_1",
            checkparams={"mytable_name": "fred", "myid_1": 7},
        )

    def test_alias_two_mysql(self):
        table1 = self.tables.mytable
        talias1 = table1.alias("t1")

        self.assert_compile(
            update(talias1)
            .where(table1.c.myid == 7)
            .values({table1.c.name: "fred"}),
            "UPDATE mytable AS t1, mytable SET mytable.name=%s "
            "WHERE mytable.myid = %s",
            checkparams={"mytable_name": "fred", "myid_1": 7},
            dialect="mysql",
        )

    def test_update_from_multitable_same_name_mysql(self):
        users, addresses = self.tables.users, self.tables.addresses

        self.assert_compile(
            users.update()
            .values(name="newname")
            .values({addresses.c.name: "new address"})
            .where(users.c.id == addresses.c.user_id),
            "UPDATE users, addresses SET addresses.name=%s, "
            "users.name=%s WHERE users.id = addresses.user_id",
            checkparams={"addresses_name": "new address", "name": "newname"},
            dialect="mysql",
        )

    def test_update_from_join_unsupported_cases(self):
        """
        found_during_typing

        It's unclear how to cleanly guard against this case without producing
        false positives, particularly due to the support for UPDATE
        of a CTE.  I'm also not sure of the nature of the failure and why
        it happens this way.

        """
        users, addresses = self.tables.users, self.tables.addresses

        j = users.join(addresses)

        with expect_raises_message(
            exc.CompileError,
            r"Encountered unsupported case when compiling an INSERT or UPDATE "
            r"statement.  If this is a multi-table "
            r"UPDATE statement, please provide string-named arguments to the "
            r"values\(\) method with distinct names; support for multi-table "
            r"UPDATE statements that "
            r"target multiple tables for UPDATE is very limited",
        ):
            update(j).where(addresses.c.email_address == "e1").values(
                {users.c.id: 10, addresses.c.email_address: "asdf"}
            ).compile(dialect=mysql.dialect())

        with expect_raises_message(
            exc.CompileError,
            r"Encountered unsupported case when compiling an INSERT or UPDATE "
            r"statement.  If this is a multi-table "
            r"UPDATE statement, please provide string-named arguments to the "
            r"values\(\) method with distinct names; support for multi-table "
            r"UPDATE statements that "
            r"target multiple tables for UPDATE is very limited",
        ):
            update(j).where(addresses.c.email_address == "e1").compile(
                dialect=mysql.dialect()
            )

    def test_update_from_join_mysql_whereclause(self):
        users, addresses = self.tables.users, self.tables.addresses

        j = users.join(addresses)
        self.assert_compile(
            update(j)
            .values(name="newname")
            .where(addresses.c.email_address == "e1"),
            ""
            "UPDATE users "
            "INNER JOIN addresses ON users.id = addresses.user_id "
            "SET users.name=%s "
            "WHERE "
            "addresses.email_address = %s",
            checkparams={"email_address_1": "e1", "name": "newname"},
            dialect=mysql.dialect(),
        )

    def test_update_from_join_mysql_no_whereclause_one(self):
        users, addresses = self.tables.users, self.tables.addresses

        j = users.join(addresses)
        self.assert_compile(
            update(j).values(name="newname"),
            ""
            "UPDATE users "
            "INNER JOIN addresses ON users.id = addresses.user_id "
            "SET users.name=%s",
            checkparams={"name": "newname"},
            dialect=mysql.dialect(),
        )

    def test_update_from_join_mysql_no_whereclause_two(self):
        users, addresses = self.tables.users, self.tables.addresses

        j = users.join(addresses)
        self.assert_compile(
            update(j).values({users.c.name: addresses.c.email_address}),
            ""
            "UPDATE users "
            "INNER JOIN addresses ON users.id = addresses.user_id "
            "SET users.name=addresses.email_address",
            checkparams={},
            dialect=mysql.dialect(),
        )

    def test_update_from_join_mysql_no_whereclause_three(self):
        users, addresses, dingalings = (
            self.tables.users,
            self.tables.addresses,
            self.tables.dingalings,
        )

        j = users.join(addresses).join(dingalings)
        self.assert_compile(
            update(j).values({users.c.name: dingalings.c.id}),
            ""
            "UPDATE users "
            "INNER JOIN addresses ON users.id = addresses.user_id "
            "INNER JOIN dingalings ON addresses.id = dingalings.address_id "
            "SET users.name=dingalings.id",
            checkparams={},
            dialect=mysql.dialect(),
        )

    def test_update_from_join_mysql_no_whereclause_four(self):
        users, addresses, dingalings = (
            self.tables.users,
            self.tables.addresses,
            self.tables.dingalings,
        )

        j = users.join(addresses).join(dingalings)

        self.assert_compile(
            update(j).values(name="foo"),
            ""
            "UPDATE users "
            "INNER JOIN addresses ON users.id = addresses.user_id "
            "INNER JOIN dingalings ON addresses.id = dingalings.address_id "
            "SET users.name=%s",
            checkparams={"name": "foo"},
            dialect=mysql.dialect(),
        )

    def test_render_table(self):
        users, addresses = self.tables.users, self.tables.addresses

        self.assert_compile(
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(addresses.c.email_address == "e1"),
            "UPDATE users "
            "SET name=:name FROM addresses "
            "WHERE "
            "users.id = addresses.user_id AND "
            "addresses.email_address = :email_address_1",
            checkparams={"email_address_1": "e1", "name": "newname"},
        )

    def test_render_multi_table(self):
        users = self.tables.users
        addresses = self.tables.addresses
        dingalings = self.tables.dingalings

        checkparams = {"email_address_1": "e1", "id_1": 2, "name": "newname"}

        self.assert_compile(
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(addresses.c.email_address == "e1")
            .where(addresses.c.id == dingalings.c.address_id)
            .where(dingalings.c.id == 2),
            "UPDATE users "
            "SET name=:name "
            "FROM addresses, dingalings "
            "WHERE "
            "users.id = addresses.user_id AND "
            "addresses.email_address = :email_address_1 AND "
            "addresses.id = dingalings.address_id AND "
            "dingalings.id = :id_1",
            checkparams=checkparams,
        )

    def test_render_table_mysql(self):
        users, addresses = self.tables.users, self.tables.addresses

        self.assert_compile(
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(addresses.c.email_address == "e1"),
            "UPDATE users, addresses "
            "SET users.name=%s "
            "WHERE "
            "users.id = addresses.user_id AND "
            "addresses.email_address = %s",
            checkparams={"email_address_1": "e1", "name": "newname"},
            dialect=mysql.dialect(),
        )

    def test_render_subquery(self):
        users, addresses = self.tables.users, self.tables.addresses

        checkparams = {"email_address_1": "e1", "id_1": 7, "name": "newname"}

        subq = (
            select(
                addresses.c.id, addresses.c.user_id, addresses.c.email_address
            )
            .where(addresses.c.id == 7)
            .alias()
        )
        self.assert_compile(
            users.update()
            .values(name="newname")
            .where(users.c.id == subq.c.user_id)
            .where(subq.c.email_address == "e1"),
            "UPDATE users "
            "SET name=:name FROM ("
            "SELECT "
            "addresses.id AS id, "
            "addresses.user_id AS user_id, "
            "addresses.email_address AS email_address "
            "FROM addresses "
            "WHERE addresses.id = :id_1"
            ") AS anon_1 "
            "WHERE users.id = anon_1.user_id "
            "AND anon_1.email_address = :email_address_1",
            checkparams=checkparams,
        )

    def test_correlation_to_extra(self):
        users, addresses = self.tables.users, self.tables.addresses

        stmt = (
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(
                ~exists()
                .where(addresses.c.user_id == users.c.id)
                .where(addresses.c.email_address == "foo")
                .correlate(addresses)
            )
        )

        self.assert_compile(
            stmt,
            "UPDATE users SET name=:name FROM addresses WHERE "
            "users.id = addresses.user_id AND NOT "
            "(EXISTS (SELECT * FROM users WHERE addresses.user_id = users.id "
            "AND addresses.email_address = :email_address_1))",
        )

    def test_dont_correlate_to_extra(self):
        users, addresses = self.tables.users, self.tables.addresses

        stmt = (
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(
                ~exists()
                .where(addresses.c.user_id == users.c.id)
                .where(addresses.c.email_address == "foo")
                .correlate()
            )
        )

        self.assert_compile(
            stmt,
            "UPDATE users SET name=:name FROM addresses WHERE "
            "users.id = addresses.user_id AND NOT "
            "(EXISTS (SELECT * FROM addresses, users "
            "WHERE addresses.user_id = users.id "
            "AND addresses.email_address = :email_address_1))",
        )

    def test_autocorrelate_error(self):
        users, addresses = self.tables.users, self.tables.addresses

        stmt = (
            users.update()
            .values(name="newname")
            .where(users.c.id == addresses.c.user_id)
            .where(
                ~exists()
                .where(addresses.c.user_id == users.c.id)
                .where(addresses.c.email_address == "foo")
            )
        )

        assert_raises_message(
            exc.InvalidRequestError,
            ".*returned no FROM clauses due to auto-correlation.*",
            stmt.compile,
            dialect=default.StrCompileDialect(),
        )


class UpdateFromRoundTripTest(_UpdateFromTestBase, fixtures.TablesTest):
    __backend__ = True

    @testing.requires.update_from
    def test_exec_two_table(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        connection.execute(
            addresses.update()
            .values(email_address=users.c.name)
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "x", "ed"),
            (3, 8, "x", "ed"),
            (4, 8, "x", "ed"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

    @testing.requires.update_from
    def test_exec_two_table_plus_alias(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        a1 = addresses.alias()
        connection.execute(
            addresses.update()
            .values(email_address=users.c.name)
            .where(users.c.id == a1.c.user_id)
            .where(users.c.name == "ed")
            .where(a1.c.id == addresses.c.id)
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "x", "ed"),
            (3, 8, "x", "ed"),
            (4, 8, "x", "ed"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

    @testing.requires.update_from
    def test_exec_three_table(self, connection):
        users = self.tables.users
        addresses = self.tables.addresses
        dingalings = self.tables.dingalings

        connection.execute(
            addresses.update()
            .values(email_address=users.c.name)
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
            .where(addresses.c.id == dingalings.c.address_id)
            .where(dingalings.c.id == 1)
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "x", "ed"),
            (3, 8, "x", "ed@bettyboop.com"),
            (4, 8, "x", "ed@lala.com"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

    @testing.requires.multi_table_update
    def test_exec_multitable(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        values = {addresses.c.email_address: "updated", users.c.name: "ed2"}

        connection.execute(
            addresses.update()
            .values(values)
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "x", "updated"),
            (3, 8, "x", "updated"),
            (4, 8, "x", "updated"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

        expected = [(7, "jack"), (8, "ed2"), (9, "fred"), (10, "chuck")]
        self._assert_users(connection, users, expected)

    @testing.requires.multi_table_update
    def test_exec_join_multitable(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        values = {addresses.c.email_address: "updated", users.c.name: "ed2"}

        connection.execute(
            update(users.join(addresses))
            .values(values)
            .where(users.c.name == "ed")
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "x", "updated"),
            (3, 8, "x", "updated"),
            (4, 8, "x", "updated"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

        expected = [(7, "jack"), (8, "ed2"), (9, "fred"), (10, "chuck")]
        self._assert_users(connection, users, expected)

    @testing.requires.multi_table_update
    def test_exec_multitable_same_name(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        values = {addresses.c.name: "ad_ed2", users.c.name: "ed2"}

        connection.execute(
            addresses.update()
            .values(values)
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
        )

        expected = [
            (1, 7, "x", "jack@bean.com"),
            (2, 8, "ad_ed2", "ed@wood.com"),
            (3, 8, "ad_ed2", "ed@bettyboop.com"),
            (4, 8, "ad_ed2", "ed@lala.com"),
            (5, 9, "x", "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

        expected = [(7, "jack"), (8, "ed2"), (9, "fred"), (10, "chuck")]
        self._assert_users(connection, users, expected)

    def _assert_addresses(self, connection, addresses, expected):
        stmt = addresses.select().order_by(addresses.c.id)
        eq_(connection.execute(stmt).fetchall(), expected)

    def _assert_users(self, connection, users, expected):
        stmt = users.select().order_by(users.c.id)
        eq_(connection.execute(stmt).fetchall(), expected)


class UpdateFromMultiTableUpdateDefaultsTest(
    _UpdateFromTestBase, fixtures.TablesTest
):
    __backend__ = True

    @classmethod
    def define_tables(cls, metadata):
        Table(
            "users",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("name", String(30), nullable=False),
            Column("some_update", String(30), onupdate="im the update"),
        )

        Table(
            "addresses",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("user_id", None, ForeignKey("users.id")),
            Column("email_address", String(50), nullable=False),
        )

        Table(
            "foobar",
            metadata,
            Column(
                "id", Integer, primary_key=True, test_needs_autoincrement=True
            ),
            Column("user_id", None, ForeignKey("users.id")),
            Column("data", String(30)),
            Column("some_update", String(30), onupdate="im the other update"),
        )

    @classmethod
    def fixtures(cls):
        return dict(
            users=(
                ("id", "name", "some_update"),
                (8, "ed", "value"),
                (9, "fred", "value"),
            ),
            addresses=(
                ("id", "user_id", "email_address"),
                (2, 8, "ed@wood.com"),
                (3, 8, "ed@bettyboop.com"),
                (4, 9, "fred@fred.com"),
            ),
            foobar=(
                ("id", "user_id", "data"),
                (2, 8, "d1"),
                (3, 8, "d2"),
                (4, 9, "d3"),
            ),
        )

    @testing.requires.multi_table_update
    def test_defaults_second_table(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        values = {addresses.c.email_address: "updated", users.c.name: "ed2"}

        ret = connection.execute(
            addresses.update()
            .values(values)
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
        )

        eq_(set(ret.prefetch_cols()), {users.c.some_update})

        expected = [
            (2, 8, "updated"),
            (3, 8, "updated"),
            (4, 9, "fred@fred.com"),
        ]
        self._assert_addresses(connection, addresses, expected)

        expected = [(8, "ed2", "im the update"), (9, "fred", "value")]
        self._assert_users(connection, users, expected)

    @testing.requires.multi_table_update
    def test_defaults_second_table_same_name(self, connection):
        users, foobar = self.tables.users, self.tables.foobar

        values = {foobar.c.data: foobar.c.data + "a", users.c.name: "ed2"}

        ret = connection.execute(
            users.update()
            .values(values)
            .where(users.c.id == foobar.c.user_id)
            .where(users.c.name == "ed")
        )

        eq_(
            set(ret.prefetch_cols()),
            {users.c.some_update, foobar.c.some_update},
        )

        expected = [
            (2, 8, "d1a", "im the other update"),
            (3, 8, "d2a", "im the other update"),
            (4, 9, "d3", None),
        ]
        self._assert_foobar(connection, foobar, expected)

        expected = [(8, "ed2", "im the update"), (9, "fred", "value")]
        self._assert_users(connection, users, expected)

    @testing.requires.multi_table_update
    def test_no_defaults_second_table(self, connection):
        users, addresses = self.tables.users, self.tables.addresses

        ret = connection.execute(
            addresses.update()
            .values({"email_address": users.c.name})
            .where(users.c.id == addresses.c.user_id)
            .where(users.c.name == "ed")
        )

        eq_(ret.prefetch_cols(), [])

        expected = [(2, 8, "ed"), (3, 8, "ed"), (4, 9, "fred@fred.com")]
        self._assert_addresses(connection, addresses, expected)

        # users table not actually updated, so no onupdate
        expected = [(8, "ed", "value"), (9, "fred", "value")]
        self._assert_users(connection, users, expected)

    def _assert_foobar(self, connection, foobar, expected):
        stmt = foobar.select().order_by(foobar.c.id)
        eq_(connection.execute(stmt).fetchall(), expected)

    def _assert_addresses(self, connection, addresses, expected):
        stmt = addresses.select().order_by(addresses.c.id)
        eq_(connection.execute(stmt).fetchall(), expected)

    def _assert_users(self, connection, users, expected):
        stmt = users.select().order_by(users.c.id)
        eq_(connection.execute(stmt).fetchall(), expected)