column("name", String(128)),
column("description", String(128)),
)
- u = update(table1, values=dict(name="foo")).returning(
- table1.c.myid, table1.c.name
+ u = (
+ update(table1)
+ .values(dict(name="foo"))
+ .returning(table1.c.myid, table1.c.name)
)
self.assert_compile(
u,
"UPDATE mytable SET name=:name OUTPUT "
"inserted.myid, inserted.name",
)
- u = update(table1, values=dict(name="foo")).returning(table1)
+ u = update(table1).values(dict(name="foo")).returning(table1)
self.assert_compile(
u,
"UPDATE mytable SET name=:name OUTPUT "
"inserted.description",
)
u = (
- update(table1, values=dict(name="foo"))
+ update(table1)
+ .values(dict(name="foo"))
.returning(table1)
.where(table1.c.name == "bar")
)
"inserted.description WHERE mytable.name = "
":name_1",
)
- u = update(table1, values=dict(name="foo")).returning(
- func.length(table1.c.name)
+ u = (
+ update(table1)
+ .values(dict(name="foo"))
+ .returning(func.length(table1.c.name))
)
self.assert_compile(
u,
column("name", String(128)),
column("description", String(128)),
)
- i = insert(table1, values=dict(name="foo")).returning(
- table1.c.myid, table1.c.name
+ i = (
+ insert(table1)
+ .values(dict(name="foo"))
+ .returning(table1.c.myid, table1.c.name)
)
self.assert_compile(
i,
"inserted.myid, inserted.name VALUES "
"(:name)",
)
- i = insert(table1, values=dict(name="foo")).returning(table1)
+ i = insert(table1).values(dict(name="foo")).returning(table1)
self.assert_compile(
i,
"INSERT INTO mytable (name) OUTPUT "
"inserted.myid, inserted.name, "
"inserted.description VALUES (:name)",
)
- i = insert(table1, values=dict(name="foo")).returning(
- func.length(table1.c.name)
+ i = (
+ insert(table1)
+ .values(dict(name="foo"))
+ .returning(func.length(table1.c.name))
)
self.assert_compile(
i,
def test_bad_args(self):
assert_raises(
ValueError,
- insert(self.tables.foos, values={}).on_duplicate_key_update,
+ insert(self.tables.foos).values({}).on_duplicate_key_update,
)
assert_raises(
exc.ArgumentError,
- insert(self.tables.foos, values={}).on_duplicate_key_update,
+ insert(self.tables.foos).values({}).on_duplicate_key_update,
{"id": 1, "bar": "b"},
id=1,
bar="b",
)
assert_raises(
exc.ArgumentError,
- insert(self.tables.foos, values={}).on_duplicate_key_update,
+ insert(self.tables.foos).values({}).on_duplicate_key_update,
{"id": 1, "bar": "b"},
{"id": 2, "bar": "baz"},
)
def test_on_duplicate_key_update_multirow(self, connection):
foos = self.tables.foos
conn = connection
- conn.execute(insert(foos, dict(id=1, bar="b", baz="bz")))
+ conn.execute(insert(foos).values(dict(id=1, bar="b", baz="bz")))
stmt = insert(foos).values([dict(id=1, bar="ab"), dict(id=2, bar="b")])
stmt = stmt.on_duplicate_key_update(bar=stmt.inserted.bar)
def test_on_duplicate_key_update_singlerow(self, connection):
foos = self.tables.foos
conn = connection
- conn.execute(insert(foos, dict(id=1, bar="b", baz="bz")))
+ conn.execute(insert(foos).values(dict(id=1, bar="b", baz="bz")))
stmt = insert(foos).values(dict(id=2, bar="b"))
stmt = stmt.on_duplicate_key_update(bar=stmt.inserted.bar)
def test_on_duplicate_key_update_null_multirow(self, connection):
foos = self.tables.foos
conn = connection
- conn.execute(insert(foos, dict(id=1, bar="b", baz="bz")))
+ conn.execute(insert(foos).values(dict(id=1, bar="b", baz="bz")))
stmt = insert(foos).values([dict(id=1, bar="ab"), dict(id=2, bar="b")])
stmt = stmt.on_duplicate_key_update(updated_once=None)
result = conn.execute(stmt)
def test_on_duplicate_key_update_expression_multirow(self, connection):
foos = self.tables.foos
conn = connection
- conn.execute(insert(foos, dict(id=1, bar="b", baz="bz")))
+ conn.execute(insert(foos).values(dict(id=1, bar="b", baz="bz")))
stmt = insert(foos).values([dict(id=1, bar="ab"), dict(id=2, bar="b")])
stmt = stmt.on_duplicate_key_update(
bar=func.concat(stmt.inserted.bar, "_foo")
foos = self.tables.foos
conn = connection
conn.execute(
- insert(
- foos,
+ insert(foos).values(
[
dict(id=1, bar="b", baz="bz"),
dict(id=2, bar="b", baz="bz2"),
column("name", String(128)),
column("description", String(128)),
)
- u = update(table1, values=dict(name="foo")).returning(
- table1.c.myid, table1.c.name
+ u = (
+ update(table1)
+ .values(dict(name="foo"))
+ .returning(table1.c.myid, table1.c.name)
)
self.assert_compile(
u,
"RETURNING mytable.myid, mytable.name",
dialect=dialect,
)
- u = update(table1, values=dict(name="foo")).returning(table1)
+ u = update(table1).values(dict(name="foo")).returning(table1)
self.assert_compile(
u,
"UPDATE mytable SET name=%(name)s "
"mytable.description",
dialect=dialect,
)
- u = update(table1, values=dict(name="foo")).returning(
- func.length(table1.c.name)
+ u = (
+ update(table1)
+ .values(dict(name="foo"))
+ .returning(func.length(table1.c.name))
)
self.assert_compile(
u,
column("description", String(128)),
)
- i = insert(table1, values=dict(name="foo")).returning(
- table1.c.myid, table1.c.name
+ i = (
+ insert(table1)
+ .values(dict(name="foo"))
+ .returning(table1.c.myid, table1.c.name)
)
self.assert_compile(
i,
"mytable.name",
dialect=dialect,
)
- i = insert(table1, values=dict(name="foo")).returning(table1)
+ i = insert(table1).values(dict(name="foo")).returning(table1)
self.assert_compile(
i,
"INSERT INTO mytable (name) VALUES "
"mytable.name, mytable.description",
dialect=dialect,
)
- i = insert(table1, values=dict(name="foo")).returning(
- func.length(table1.c.name)
+ i = (
+ insert(table1)
+ .values(dict(name="foo"))
+ .returning(func.length(table1.c.name))
)
self.assert_compile(
i,
)
self.assert_compile(stmt, expected)
- stmt = update(tbl1, values=dict(id=1))
+ stmt = update(tbl1).values(dict(id=1))
stmt = stmt.with_hint("ONLY", dialect_name="postgresql")
expected = "UPDATE ONLY testtbl1 SET id=%(id)s"
self.assert_compile(stmt, expected)
def test_do_nothing_no_target(self):
- i = insert(
- self.table1, values=dict(name="foo")
- ).on_conflict_do_nothing()
+ i = (
+ insert(self.table1)
+ .values(dict(name="foo"))
+ .on_conflict_do_nothing()
+ )
self.assert_compile(
i,
"INSERT INTO mytable (name) VALUES "
def test_do_nothing_index_elements_target(self):
- i = insert(
- self.table1, values=dict(name="foo")
- ).on_conflict_do_nothing(index_elements=["myid"])
+ i = (
+ insert(self.table1)
+ .values(dict(name="foo"))
+ .on_conflict_do_nothing(index_elements=["myid"])
+ )
self.assert_compile(
i,
"INSERT INTO mytable (name) VALUES "
)
def test_do_update_str_index_elements_target_two(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
index_elements=["myid"], set_=dict(name=i.excluded.name)
)
)
def test_do_update_col_index_elements_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
index_elements=[self.table1.c.myid],
set_=dict(name=i.excluded.name),
)
def test_do_update_unnamed_pk_constraint_target(self):
- i = insert(self.table_with_metadata, values=dict(myid=1, name="foo"))
+ i = insert(self.table_with_metadata).values(dict(myid=1, name="foo"))
i = i.on_conflict_do_update(
constraint=self.table_with_metadata.primary_key,
set_=dict(name=i.excluded.name),
)
def test_do_update_pk_constraint_index_elements_target(self):
- i = insert(self.table_with_metadata, values=dict(myid=1, name="foo"))
+ i = insert(self.table_with_metadata).values(dict(myid=1, name="foo"))
i = i.on_conflict_do_update(
index_elements=self.table_with_metadata.primary_key,
set_=dict(name=i.excluded.name),
)
def test_do_update_named_unique_constraint_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
constraint=self.unique_constr, set_=dict(myid=i.excluded.myid)
)
)
def test_do_update_string_constraint_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
constraint=self.unique_constr.name, set_=dict(myid=i.excluded.myid)
)
def test_do_nothing_quoted_string_constraint_target(self):
"""test #6696"""
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_nothing(constraint="Some Constraint Name")
self.assert_compile(
i,
uq,
)
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_nothing(constraint=uq)
self.assert_compile(
def test_do_nothing_quoted_named_constraint_target(self):
"""test #6696"""
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
unique_constr = UniqueConstraint(
self.table1.c.myid, name="Some Constraint Name"
)
)
def test_do_update_index_elements_where_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
index_elements=self.goofy_index.expressions,
index_where=self.goofy_index.dialect_options["postgresql"][
)
def test_do_update_index_elements_where_target_multivalues(self):
- i = insert(
- self.table1,
- values=[dict(name="foo"), dict(name="bar"), dict(name="bat")],
+ i = insert(self.table1).values(
+ [dict(name="foo"), dict(name="bar"), dict(name="bat")],
)
i = i.on_conflict_do_update(
index_elements=self.goofy_index.expressions,
)
def test_do_update_unnamed_index_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
unnamed_goofy = Index(
None, self.table1.c.name, postgresql_where=self.table1.c.name > "m"
)
def test_do_update_unnamed_exclude_constraint_target(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
constraint=self.excl_constr_anon, set_=dict(name=i.excluded.name)
)
)
def test_do_update_add_whereclause(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
constraint=self.excl_constr_anon,
set_=dict(name=i.excluded.name),
)
def test_do_update_add_whereclause_references_excluded(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = i.on_conflict_do_update(
constraint=self.excl_constr_anon,
set_=dict(name=i.excluded.name),
)
def test_do_update_additional_colnames(self):
- i = insert(self.table1, values=dict(name="bar"))
+ i = insert(self.table1).values(dict(name="bar"))
i = i.on_conflict_do_update(
constraint=self.excl_constr_anon,
set_=dict(name="somename", unknown="unknown"),
)
def test_on_conflict_as_cte(self):
- i = insert(self.table1, values=dict(name="foo"))
+ i = insert(self.table1).values(dict(name="foo"))
i = (
i.on_conflict_do_update(
constraint=self.excl_constr_anon,