summaryrefslogtreecommitdiff
path: root/test/sql/test_insert.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 01:14:26 -0500
committermike bayer <mike_mp@zzzcomputing.com>2019-01-06 17:34:50 +0000
commit1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch)
tree28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/sql/test_insert.py
parent404e69426b05a82d905cbb3ad33adafccddb00dd (diff)
downloadsqlalchemy-1e1a38e7801f410f244e4bbb44ec795ae152e04e.tar.gz
Run black -l 79 against all source files
This is a straight reformat run using black as is, with no edits applied at all. The black run will format code consistently, however in some cases that are prevalent in SQLAlchemy code it produces too-long lines. The too-long lines will be resolved in the following commit that will resolve all remaining flake8 issues including shadowed builtins, long lines, import order, unused imports, duplicate imports, and docstring issues. Change-Id: I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9
Diffstat (limited to 'test/sql/test_insert.py')
-rw-r--r--test/sql/test_insert.py1093
1 files changed, 591 insertions, 502 deletions
diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index 729c420c0..3643deabd 100644
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -1,144 +1,167 @@
#! coding:utf-8
-from sqlalchemy import Column, Integer, MetaData, String, Table,\
- bindparam, exc, func, insert, select, column, text, table,\
- Sequence
+from sqlalchemy import (
+ Column,
+ Integer,
+ MetaData,
+ String,
+ Table,
+ bindparam,
+ exc,
+ func,
+ insert,
+ select,
+ column,
+ text,
+ table,
+ Sequence,
+)
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.engine import default
-from sqlalchemy.testing import AssertsCompiledSQL,\
- assert_raises_message, fixtures, eq_, expect_warnings, assert_raises
+from sqlalchemy.testing import (
+ AssertsCompiledSQL,
+ assert_raises_message,
+ fixtures,
+ eq_,
+ expect_warnings,
+ assert_raises,
+)
from sqlalchemy.sql import crud
class _InsertTestBase(object):
-
@classmethod
def define_tables(cls, metadata):
- Table('mytable', metadata,
- Column('myid', Integer),
- Column('name', String(30)),
- Column('description', String(30)))
- Table('myothertable', metadata,
- Column('otherid', Integer, primary_key=True),
- Column('othername', String(30)))
- Table('table_w_defaults', metadata,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, default=10),
- Column('y', Integer, server_default=text('5')),
- Column('z', Integer, default=lambda: 10))
+ Table(
+ "mytable",
+ metadata,
+ Column("myid", Integer),
+ Column("name", String(30)),
+ Column("description", String(30)),
+ )
+ Table(
+ "myothertable",
+ metadata,
+ Column("otherid", Integer, primary_key=True),
+ Column("othername", String(30)),
+ )
+ Table(
+ "table_w_defaults",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, default=10),
+ Column("y", Integer, server_default=text("5")),
+ Column("z", Integer, default=lambda: 10),
+ )
class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
- __dialect__ = 'default'
+ __dialect__ = "default"
def test_binds_that_match_columns(self):
"""test bind params named after column names
replace the normal SET/VALUES generation."""
- t = table('foo', column('x'), column('y'))
+ t = table("foo", column("x"), column("y"))
- i = t.insert().values(x=3 + bindparam('x'))
- self.assert_compile(i,
- "INSERT INTO foo (x) VALUES ((:param_1 + :x))")
+ i = t.insert().values(x=3 + bindparam("x"))
+ self.assert_compile(i, "INSERT INTO foo (x) VALUES ((:param_1 + :x))")
self.assert_compile(
i,
"INSERT INTO foo (x, y) VALUES ((:param_1 + :x), :y)",
- params={
- 'x': 1,
- 'y': 2})
+ params={"x": 1, "y": 2},
+ )
- i = t.insert().values(x=bindparam('y'))
+ i = t.insert().values(x=bindparam("y"))
self.assert_compile(i, "INSERT INTO foo (x) VALUES (:y)")
- i = t.insert().values(x=bindparam('y'), y=5)
+ i = t.insert().values(x=bindparam("y"), y=5)
assert_raises(exc.CompileError, i.compile)
- i = t.insert().values(x=3 + bindparam('y'), y=5)
+ i = t.insert().values(x=3 + bindparam("y"), y=5)
assert_raises(exc.CompileError, i.compile)
- i = t.insert().values(x=3 + bindparam('x2'))
- self.assert_compile(i,
- "INSERT INTO foo (x) VALUES ((:param_1 + :x2))")
+ i = t.insert().values(x=3 + bindparam("x2"))
+ self.assert_compile(i, "INSERT INTO foo (x) VALUES ((:param_1 + :x2))")
self.assert_compile(
- i,
- "INSERT INTO foo (x) VALUES ((:param_1 + :x2))",
- params={})
+ i, "INSERT INTO foo (x) VALUES ((:param_1 + :x2))", params={}
+ )
self.assert_compile(
i,
"INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)",
- params={
- 'x': 1,
- 'y': 2})
+ params={"x": 1, "y": 2},
+ )
self.assert_compile(
i,
"INSERT INTO foo (x, y) VALUES ((:param_1 + :x2), :y)",
- params={
- 'x2': 1,
- 'y': 2})
+ params={"x2": 1, "y": 2},
+ )
def test_insert_literal_binds(self):
table1 = self.tables.mytable
- stmt = table1.insert().values(myid=3, name='jack')
+ stmt = table1.insert().values(myid=3, name="jack")
self.assert_compile(
stmt,
"INSERT INTO mytable (myid, name) VALUES (3, 'jack')",
- literal_binds=True)
+ literal_binds=True,
+ )
def test_insert_literal_binds_sequence_notimplemented(self):
- table = Table('x', MetaData(), Column('y', Integer, Sequence('y_seq')))
+ table = Table("x", MetaData(), Column("y", Integer, Sequence("y_seq")))
dialect = default.DefaultDialect()
dialect.supports_sequences = True
- stmt = table.insert().values(myid=3, name='jack')
+ stmt = table.insert().values(myid=3, name="jack")
assert_raises(
NotImplementedError,
stmt.compile,
- compile_kwargs=dict(literal_binds=True), dialect=dialect
+ compile_kwargs=dict(literal_binds=True),
+ dialect=dialect,
)
def test_inline_defaults(self):
m = MetaData()
- foo = Table('foo', m,
- Column('id', Integer))
+ foo = Table("foo", m, Column("id", Integer))
- t = Table('test', m,
- Column('col1', Integer, default=func.foo(1)),
- Column('col2', Integer, default=select(
- [func.coalesce(func.max(foo.c.id))])),
- )
+ t = Table(
+ "test",
+ m,
+ Column("col1", Integer, default=func.foo(1)),
+ Column(
+ "col2",
+ Integer,
+ default=select([func.coalesce(func.max(foo.c.id))]),
+ ),
+ )
self.assert_compile(
- t.insert(
- inline=True, values={}),
+ t.insert(inline=True, values={}),
"INSERT INTO test (col1, col2) VALUES (foo(:foo_1), "
"(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM "
- "foo))")
+ "foo))",
+ )
def test_generic_insert_bind_params_all_columns(self):
table1 = self.tables.mytable
- self.assert_compile(insert(table1),
- 'INSERT INTO mytable (myid, name, description) '
- 'VALUES (:myid, :name, :description)')
+ self.assert_compile(
+ insert(table1),
+ "INSERT INTO mytable (myid, name, description) "
+ "VALUES (:myid, :name, :description)",
+ )
def test_insert_with_values_dict(self):
table1 = self.tables.mytable
- checkparams = {
- 'myid': 3,
- 'name': 'jack'
- }
+ checkparams = {"myid": 3, "name": "jack"}
self.assert_compile(
- insert(
- table1,
- dict(
- myid=3,
- name='jack')),
- 'INSERT INTO mytable (myid, name) VALUES (:myid, :name)',
- checkparams=checkparams)
+ insert(table1, dict(myid=3, name="jack")),
+ "INSERT INTO mytable (myid, name) VALUES (:myid, :name)",
+ checkparams=checkparams,
+ )
def test_unconsumed_names_kwargs(self):
t = table("t", column("x"), column("y"))
@@ -151,136 +174,133 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
def test_bindparam_name_no_consume_error(self):
t = table("t", column("x"), column("y"))
# bindparam names don't get counted
- i = t.insert().values(x=3 + bindparam('x2'))
- self.assert_compile(
- i,
- "INSERT INTO t (x) VALUES ((:param_1 + :x2))"
- )
+ i = t.insert().values(x=3 + bindparam("x2"))
+ self.assert_compile(i, "INSERT INTO t (x) VALUES ((:param_1 + :x2))")
# even if in the params list
- i = t.insert().values(x=3 + bindparam('x2'))
+ i = t.insert().values(x=3 + bindparam("x2"))
self.assert_compile(
- i,
- "INSERT INTO t (x) VALUES ((:param_1 + :x2))",
- params={"x2": 1}
+ i, "INSERT INTO t (x) VALUES ((:param_1 + :x2))", params={"x2": 1}
)
def test_unconsumed_names_values_dict(self):
table1 = self.tables.mytable
- checkparams = {
- 'myid': 3,
- 'name': 'jack',
- 'unknowncol': 'oops'
- }
+ checkparams = {"myid": 3, "name": "jack", "unknowncol": "oops"}
stmt = insert(table1, values=checkparams)
assert_raises_message(
exc.CompileError,
- 'Unconsumed column names: unknowncol',
+ "Unconsumed column names: unknowncol",
stmt.compile,
- dialect=postgresql.dialect()
+ dialect=postgresql.dialect(),
)
def test_unconsumed_names_multi_values_dict(self):
table1 = self.tables.mytable
- checkparams = [{
- 'myid': 3,
- 'name': 'jack',
- 'unknowncol': 'oops'
- }, {
- 'myid': 4,
- 'name': 'someone',
- 'unknowncol': 'oops'
- }]
+ checkparams = [
+ {"myid": 3, "name": "jack", "unknowncol": "oops"},
+ {"myid": 4, "name": "someone", "unknowncol": "oops"},
+ ]
stmt = insert(table1, values=checkparams)
assert_raises_message(
exc.CompileError,
- 'Unconsumed column names: unknowncol',
+ "Unconsumed column names: unknowncol",
stmt.compile,
- dialect=postgresql.dialect()
+ dialect=postgresql.dialect(),
)
def test_insert_with_values_tuple(self):
table1 = self.tables.mytable
checkparams = {
- 'myid': 3,
- 'name': 'jack',
- 'description': 'mydescription'
+ "myid": 3,
+ "name": "jack",
+ "description": "mydescription",
}
- self.assert_compile(insert(table1, (3, 'jack', 'mydescription')),
- 'INSERT INTO mytable (myid, name, description) '
- 'VALUES (:myid, :name, :description)',
- checkparams=checkparams)
+ self.assert_compile(
+ insert(table1, (3, "jack", "mydescription")),
+ "INSERT INTO mytable (myid, name, description) "
+ "VALUES (:myid, :name, :description)",
+ checkparams=checkparams,
+ )
def test_insert_with_values_func(self):
table1 = self.tables.mytable
- self.assert_compile(insert(table1, values=dict(myid=func.lala())),
- 'INSERT INTO mytable (myid) VALUES (lala())')
+ self.assert_compile(
+ insert(table1, values=dict(myid=func.lala())),
+ "INSERT INTO mytable (myid) VALUES (lala())",
+ )
def test_insert_with_user_supplied_bind_params(self):
table1 = self.tables.mytable
values = {
- table1.c.myid: bindparam('userid'),
- table1.c.name: bindparam('username')
+ table1.c.myid: bindparam("userid"),
+ table1.c.name: bindparam("username"),
}
self.assert_compile(
- insert(
- table1,
- values),
- 'INSERT INTO mytable (myid, name) VALUES (:userid, :username)')
+ insert(table1, values),
+ "INSERT INTO mytable (myid, name) VALUES (:userid, :username)",
+ )
def test_insert_values(self):
table1 = self.tables.mytable
- values1 = {table1.c.myid: bindparam('userid')}
- values2 = {table1.c.name: bindparam('username')}
+ values1 = {table1.c.myid: bindparam("userid")}
+ values2 = {table1.c.name: bindparam("username")}
self.assert_compile(
- insert(
- table1,
- values=values1).values(values2),
- 'INSERT INTO mytable (myid, name) VALUES (:userid, :username)')
+ insert(table1, values=values1).values(values2),
+ "INSERT INTO mytable (myid, name) VALUES (:userid, :username)",
+ )
def test_prefix_with(self):
table1 = self.tables.mytable
- stmt = table1.insert().\
- prefix_with('A', 'B', dialect='mysql').\
- prefix_with('C', 'D')
+ stmt = (
+ table1.insert()
+ .prefix_with("A", "B", dialect="mysql")
+ .prefix_with("C", "D")
+ )
self.assert_compile(
stmt,
- 'INSERT C D INTO mytable (myid, name, description) '
- 'VALUES (:myid, :name, :description)')
+ "INSERT C D INTO mytable (myid, name, description) "
+ "VALUES (:myid, :name, :description)",
+ )
self.assert_compile(
stmt,
- 'INSERT A B C D INTO mytable (myid, name, description) '
- 'VALUES (%s, %s, %s)',
- dialect=mysql.dialect())
+ "INSERT A B C D INTO mytable (myid, name, description) "
+ "VALUES (%s, %s, %s)",
+ dialect=mysql.dialect(),
+ )
def test_inline_default(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=func.foobar()),
+ )
- self.assert_compile(table.insert(values={}, inline=True),
- 'INSERT INTO sometable (foo) VALUES (foobar())')
+ self.assert_compile(
+ table.insert(values={}, inline=True),
+ "INSERT INTO sometable (foo) VALUES (foobar())",
+ )
self.assert_compile(
- table.insert(
- inline=True),
- 'INSERT INTO sometable (foo) VALUES (foobar())',
- params={})
+ table.insert(inline=True),
+ "INSERT INTO sometable (foo) VALUES (foobar())",
+ params={},
+ )
def test_insert_returning_not_in_default(self):
table1 = self.tables.mytable
@@ -290,68 +310,75 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
exc.CompileError,
"RETURNING is not supported by this dialect's statement compiler.",
stmt.compile,
- dialect=default.DefaultDialect()
+ dialect=default.DefaultDialect(),
)
def test_insert_from_select_returning(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel).returning(
- self.tables.myothertable.c.otherid
- )
+ table1.c.name == "foo"
+ )
+ ins = (
+ self.tables.myothertable.insert()
+ .from_select(("otherid", "othername"), sel)
+ .returning(self.tables.myothertable.c.otherid)
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable "
"WHERE mytable.name = %(name_1)s RETURNING myothertable.otherid",
checkparams={"name_1": "foo"},
- dialect="postgresql"
+ dialect="postgresql",
)
def test_insert_from_select_select(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel)
+ table1.c.name == "foo"
+ )
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), sel
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable "
"WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_from_select_seq(self):
m = MetaData()
t1 = Table(
- 't', m,
- Column('id', Integer, Sequence('id_seq'), primary_key=True),
- Column('data', String)
+ "t",
+ m,
+ Column("id", Integer, Sequence("id_seq"), primary_key=True),
+ Column("data", String),
)
- stmt = t1.insert().from_select(('data', ), select([t1.c.data]))
+ stmt = t1.insert().from_select(("data",), select([t1.c.data]))
self.assert_compile(
stmt,
"INSERT INTO t (data, id) SELECT t.data, "
"nextval('id_seq') AS next_value_1 FROM t",
- dialect=postgresql.dialect()
+ dialect=postgresql.dialect(),
)
def test_insert_from_select_cte_one(self):
table1 = self.tables.mytable
- cte = select([table1.c.name]).where(table1.c.name == 'bar').cte()
+ cte = select([table1.c.name]).where(table1.c.name == "bar").cte()
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == cte.c.name)
+ table1.c.name == cte.c.name
+ )
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel)
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), sel
+ )
self.assert_compile(
ins,
"WITH anon_1 AS "
@@ -360,7 +387,7 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable, anon_1 "
"WHERE mytable.name = anon_1.name",
- checkparams={"name_1": "bar"}
+ checkparams={"name_1": "bar"},
)
def test_insert_from_select_cte_follows_insert_one(self):
@@ -369,13 +396,15 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
table1 = self.tables.mytable
- cte = select([table1.c.name]).where(table1.c.name == 'bar').cte()
+ cte = select([table1.c.name]).where(table1.c.name == "bar").cte()
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == cte.c.name)
+ table1.c.name == cte.c.name
+ )
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel)
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), sel
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
@@ -385,7 +414,7 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"SELECT mytable.myid, mytable.name FROM mytable, anon_1 "
"WHERE mytable.name = anon_1.name",
checkparams={"name_1": "bar"},
- dialect=dialect
+ dialect=dialect,
)
def test_insert_from_select_cte_two(self):
@@ -400,7 +429,7 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"WITH c AS (SELECT mytable.myid AS myid, mytable.name AS name, "
"mytable.description AS description FROM mytable) "
"INSERT INTO mytable (myid, name, description) "
- "SELECT c.myid, c.name, c.description FROM c"
+ "SELECT c.myid, c.name, c.description FROM c",
)
def test_insert_from_select_cte_follows_insert_two(self):
@@ -418,91 +447,101 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"WITH c AS (SELECT mytable.myid AS myid, mytable.name AS name, "
"mytable.description AS description FROM mytable) "
"SELECT c.myid, c.name, c.description FROM c",
- dialect=dialect
+ dialect=dialect,
)
def test_insert_from_select_select_alt_ordering(self):
table1 = self.tables.mytable
sel = select([table1.c.name, table1.c.myid]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("othername", "otherid"), sel)
+ table1.c.name == "foo"
+ )
+ ins = self.tables.myothertable.insert().from_select(
+ ("othername", "otherid"), sel
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (othername, otherid) "
"SELECT mytable.name, mytable.myid FROM mytable "
"WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_from_select_no_defaults(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=func.foobar()),
+ )
table1 = self.tables.mytable
- sel = select([table1.c.myid]).where(table1.c.name == 'foo')
- ins = table.insert().\
- from_select(["id"], sel, include_defaults=False)
+ sel = select([table1.c.myid]).where(table1.c.name == "foo")
+ ins = table.insert().from_select(["id"], sel, include_defaults=False)
self.assert_compile(
ins,
"INSERT INTO sometable (id) SELECT mytable.myid "
"FROM mytable WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_from_select_with_sql_defaults(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=func.foobar()),
+ )
table1 = self.tables.mytable
- sel = select([table1.c.myid]).where(table1.c.name == 'foo')
- ins = table.insert().\
- from_select(["id"], sel)
+ sel = select([table1.c.myid]).where(table1.c.name == "foo")
+ ins = table.insert().from_select(["id"], sel)
self.assert_compile(
ins,
"INSERT INTO sometable (id, foo) SELECT "
"mytable.myid, foobar() AS foobar_1 "
"FROM mytable WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_from_select_with_python_defaults(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=12))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=12),
+ )
table1 = self.tables.mytable
- sel = select([table1.c.myid]).where(table1.c.name == 'foo')
- ins = table.insert().\
- from_select(["id"], sel)
+ sel = select([table1.c.myid]).where(table1.c.name == "foo")
+ ins = table.insert().from_select(["id"], sel)
self.assert_compile(
ins,
"INSERT INTO sometable (id, foo) SELECT "
"mytable.myid, :foo AS anon_1 "
"FROM mytable WHERE mytable.name = :name_1",
# value filled in at execution time
- checkparams={"name_1": "foo", "foo": None}
+ checkparams={"name_1": "foo", "foo": None},
)
def test_insert_from_select_override_defaults(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=12))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=12),
+ )
table1 = self.tables.mytable
- sel = select(
- [table1.c.myid, table1.c.myid.label('q')]).where(
- table1.c.name == 'foo')
- ins = table.insert().\
- from_select(["id", "foo"], sel)
+ sel = select([table1.c.myid, table1.c.myid.label("q")]).where(
+ table1.c.name == "foo"
+ )
+ ins = table.insert().from_select(["id", "foo"], sel)
self.assert_compile(
ins,
"INSERT INTO sometable (id, foo) SELECT "
"mytable.myid, mytable.myid AS q "
"FROM mytable WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_from_select_fn_defaults(self):
@@ -511,159 +550,171 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
def foo(ctx):
return 12
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer, default=foo))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo", Integer, default=foo),
+ )
table1 = self.tables.mytable
- sel = select(
- [table1.c.myid]).where(
- table1.c.name == 'foo')
- ins = table.insert().\
- from_select(["id"], sel)
+ sel = select([table1.c.myid]).where(table1.c.name == "foo")
+ ins = table.insert().from_select(["id"], sel)
self.assert_compile(
ins,
"INSERT INTO sometable (id, foo) SELECT "
"mytable.myid, :foo AS anon_1 "
"FROM mytable WHERE mytable.name = :name_1",
# value filled in at execution time
- checkparams={"name_1": "foo", "foo": None}
+ checkparams={"name_1": "foo", "foo": None},
)
def test_insert_from_select_dont_mutate_raw_columns(self):
# test [ticket:3603]
from sqlalchemy import table
+
table_ = table(
- 'mytable',
- Column('foo', String),
- Column('bar', String, default='baz'),
+ "mytable",
+ Column("foo", String),
+ Column("bar", String, default="baz"),
)
stmt = select([table_.c.foo])
- insert = table_.insert().from_select(['foo'], stmt)
+ insert = table_.insert().from_select(["foo"], stmt)
self.assert_compile(stmt, "SELECT mytable.foo FROM mytable")
self.assert_compile(
insert,
"INSERT INTO mytable (foo, bar) "
- "SELECT mytable.foo, :bar AS anon_1 FROM mytable"
+ "SELECT mytable.foo, :bar AS anon_1 FROM mytable",
)
self.assert_compile(stmt, "SELECT mytable.foo FROM mytable")
self.assert_compile(
insert,
"INSERT INTO mytable (foo, bar) "
- "SELECT mytable.foo, :bar AS anon_1 FROM mytable"
+ "SELECT mytable.foo, :bar AS anon_1 FROM mytable",
)
def test_insert_mix_select_values_exception(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel)
+ table1.c.name == "foo"
+ )
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), sel
+ )
assert_raises_message(
exc.InvalidRequestError,
"This construct already inserts from a SELECT",
- ins.values, othername="5"
+ ins.values,
+ othername="5",
)
def test_insert_mix_values_select_exception(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
+ table1.c.name == "foo"
+ )
ins = self.tables.myothertable.insert().values(othername="5")
assert_raises_message(
exc.InvalidRequestError,
"This construct already inserts value expressions",
- ins.from_select, ("otherid", "othername"), sel
+ ins.from_select,
+ ("otherid", "othername"),
+ sel,
)
def test_insert_from_select_table(self):
table1 = self.tables.mytable
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), table1)
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), table1
+ )
# note we aren't checking the number of columns right now
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name, mytable.description "
"FROM mytable",
- checkparams={}
+ checkparams={},
)
def test_insert_from_select_union(self):
mytable = self.tables.mytable
- name = column('name')
- description = column('desc')
- sel = select(
- [name, mytable.c.description],
- ).union(
+ name = column("name")
+ description = column("desc")
+ sel = select([name, mytable.c.description]).union(
select([name, description])
)
- ins = mytable.insert().\
- from_select(
- [mytable.c.name, mytable.c.description], sel)
+ ins = mytable.insert().from_select(
+ [mytable.c.name, mytable.c.description], sel
+ )
self.assert_compile(
ins,
"INSERT INTO mytable (name, description) "
"SELECT name, mytable.description FROM mytable "
- 'UNION SELECT name, "desc"'
+ 'UNION SELECT name, "desc"',
)
def test_insert_from_select_col_values(self):
table1 = self.tables.mytable
table2 = self.tables.myothertable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = table2.insert().\
- from_select((table2.c.otherid, table2.c.othername), sel)
+ table1.c.name == "foo"
+ )
+ ins = table2.insert().from_select(
+ (table2.c.otherid, table2.c.othername), sel
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable "
"WHERE mytable.name = :name_1",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_anticipate_no_pk_composite_pk(self):
t = Table(
- 't', MetaData(), Column('x', Integer, primary_key=True),
- Column('y', Integer, primary_key=True)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True),
+ Column("y", Integer, primary_key=True),
)
with expect_warnings(
"Column 't.y' is marked as a member.*"
- "Note that as of SQLAlchemy 1.1,",
+ "Note that as of SQLAlchemy 1.1,"
):
self.assert_compile(
- t.insert(),
- "INSERT INTO t (x) VALUES (:x)",
- params={'x': 5},
+ t.insert(), "INSERT INTO t (x) VALUES (:x)", params={"x": 5}
)
def test_anticipate_no_pk_composite_pk_implicit_returning(self):
t = Table(
- 't', MetaData(), Column('x', Integer, primary_key=True),
- Column('y', Integer, primary_key=True)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True),
+ Column("y", Integer, primary_key=True),
)
d = postgresql.dialect()
d.implicit_returning = True
with expect_warnings(
"Column 't.y' is marked as a member.*"
- "Note that as of SQLAlchemy 1.1,",
+ "Note that as of SQLAlchemy 1.1,"
):
self.assert_compile(
t.insert(),
"INSERT INTO t (x) VALUES (%(x)s)",
params={"x": 5},
- dialect=d
+ dialect=d,
)
def test_anticipate_no_pk_composite_pk_prefetch(self):
t = Table(
- 't', MetaData(), Column('x', Integer, primary_key=True),
- Column('y', Integer, primary_key=True)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True),
+ Column("y", Integer, primary_key=True),
)
d = postgresql.dialect()
d.implicit_returning = False
@@ -674,202 +725,193 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
self.assert_compile(
t.insert(),
"INSERT INTO t (x) VALUES (%(x)s)",
- params={'x': 5},
- dialect=d
+ params={"x": 5},
+ dialect=d,
)
def test_anticipate_nullable_composite_pk(self):
t = Table(
- 't', MetaData(), Column('x', Integer, primary_key=True),
- Column('y', Integer, primary_key=True, nullable=True)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True),
+ Column("y", Integer, primary_key=True, nullable=True),
)
self.assert_compile(
- t.insert(),
- "INSERT INTO t (x) VALUES (:x)",
- params={'x': 5},
+ t.insert(), "INSERT INTO t (x) VALUES (:x)", params={"x": 5}
)
def test_anticipate_no_pk_non_composite_pk(self):
t = Table(
- 't', MetaData(),
- Column('x', Integer, primary_key=True, autoincrement=False),
- Column('q', Integer)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ Column("q", Integer),
)
with expect_warnings(
- "Column 't.x' is marked as a member.*"
- "may not store NULL.$"
+ "Column 't.x' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
- t.insert(),
- "INSERT INTO t (q) VALUES (:q)",
- params={"q": 5}
+ t.insert(), "INSERT INTO t (q) VALUES (:q)", params={"q": 5}
)
def test_anticipate_no_pk_non_composite_pk_implicit_returning(self):
t = Table(
- 't', MetaData(),
- Column('x', Integer, primary_key=True, autoincrement=False),
- Column('q', Integer)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ Column("q", Integer),
)
d = postgresql.dialect()
d.implicit_returning = True
with expect_warnings(
- "Column 't.x' is marked as a member.*"
- "may not store NULL.$",
+ "Column 't.x' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
t.insert(),
"INSERT INTO t (q) VALUES (%(q)s)",
params={"q": 5},
- dialect=d
+ dialect=d,
)
def test_anticipate_no_pk_non_composite_pk_prefetch(self):
t = Table(
- 't', MetaData(),
- Column('x', Integer, primary_key=True, autoincrement=False),
- Column('q', Integer)
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ Column("q", Integer),
)
d = postgresql.dialect()
d.implicit_returning = False
with expect_warnings(
- "Column 't.x' is marked as a member.*"
- "may not store NULL.$"
+ "Column 't.x' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
t.insert(),
"INSERT INTO t (q) VALUES (%(q)s)",
params={"q": 5},
- dialect=d
+ dialect=d,
)
def test_anticipate_no_pk_lower_case_table(self):
t = table(
- 't',
- Column(
- 'id', Integer, primary_key=True, autoincrement=False),
- Column('notpk', String(10), nullable=True)
+ "t",
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("notpk", String(10), nullable=True),
)
with expect_warnings(
- "Column 't.id' is marked as a member.*"
- "may not store NULL.$"
+ "Column 't.id' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
- t.insert(),
- "INSERT INTO t () VALUES ()",
- params={}
+ t.insert(), "INSERT INTO t () VALUES ()", params={}
)
class InsertImplicitReturningTest(
- _InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
+ _InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL
+):
__dialect__ = postgresql.dialect(implicit_returning=True)
def test_insert_select(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel)
+ table1.c.name == "foo"
+ )
+ ins = self.tables.myothertable.insert().from_select(
+ ("otherid", "othername"), sel
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable "
"WHERE mytable.name = %(name_1)s",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_select_return_defaults(self):
table1 = self.tables.mytable
sel = select([table1.c.myid, table1.c.name]).where(
- table1.c.name == 'foo')
- ins = self.tables.myothertable.insert().\
- from_select(("otherid", "othername"), sel).\
- return_defaults(self.tables.myothertable.c.otherid)
+ table1.c.name == "foo"
+ )
+ ins = (
+ self.tables.myothertable.insert()
+ .from_select(("otherid", "othername"), sel)
+ .return_defaults(self.tables.myothertable.c.otherid)
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (otherid, othername) "
"SELECT mytable.myid, mytable.name FROM mytable "
"WHERE mytable.name = %(name_1)s",
- checkparams={"name_1": "foo"}
+ checkparams={"name_1": "foo"},
)
def test_insert_multiple_values(self):
- ins = self.tables.myothertable.insert().values([
- {"othername": "foo"},
- {"othername": "bar"},
- ])
+ ins = self.tables.myothertable.insert().values(
+ [{"othername": "foo"}, {"othername": "bar"}]
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (othername) "
"VALUES (%(othername_m0)s), "
"(%(othername_m1)s)",
- checkparams={
- 'othername_m1': 'bar',
- 'othername_m0': 'foo'}
+ checkparams={"othername_m1": "bar", "othername_m0": "foo"},
)
def test_insert_multiple_values_literal_binds(self):
- ins = self.tables.myothertable.insert().values([
- {"othername": "foo"},
- {"othername": "bar"},
- ])
+ ins = self.tables.myothertable.insert().values(
+ [{"othername": "foo"}, {"othername": "bar"}]
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (othername) VALUES ('foo'), ('bar')",
checkparams={},
- literal_binds=True
+ literal_binds=True,
)
def test_insert_multiple_values_return_defaults(self):
# TODO: not sure if this should raise an
# error or what
- ins = self.tables.myothertable.insert().values([
- {"othername": "foo"},
- {"othername": "bar"},
- ]).return_defaults(self.tables.myothertable.c.otherid)
+ ins = (
+ self.tables.myothertable.insert()
+ .values([{"othername": "foo"}, {"othername": "bar"}])
+ .return_defaults(self.tables.myothertable.c.otherid)
+ )
self.assert_compile(
ins,
"INSERT INTO myothertable (othername) "
"VALUES (%(othername_m0)s), "
"(%(othername_m1)s)",
- checkparams={
- 'othername_m1': 'bar',
- 'othername_m0': 'foo'}
+ checkparams={"othername_m1": "bar", "othername_m0": "foo"},
)
def test_insert_single_list_values(self):
- ins = self.tables.myothertable.insert().values([
- {"othername": "foo"},
- ])
+ ins = self.tables.myothertable.insert().values([{"othername": "foo"}])
self.assert_compile(
ins,
"INSERT INTO myothertable (othername) "
"VALUES (%(othername_m0)s)",
- checkparams={'othername_m0': 'foo'}
+ checkparams={"othername_m0": "foo"},
)
def test_insert_single_element_values(self):
- ins = self.tables.myothertable.insert().values(
- {"othername": "foo"},
- )
+ ins = self.tables.myothertable.insert().values({"othername": "foo"})
self.assert_compile(
ins,
"INSERT INTO myothertable (othername) "
"VALUES (%(othername)s) RETURNING myothertable.otherid",
- checkparams={'othername': 'foo'}
+ checkparams={"othername": "foo"},
)
class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
- __dialect__ = 'default'
+ __dialect__ = "default"
def test_empty_insert_default(self):
table1 = self.tables.mytable
stmt = table1.insert().values({}) # hide from 2to3
- self.assert_compile(stmt, 'INSERT INTO mytable () VALUES ()')
+ self.assert_compile(stmt, "INSERT INTO mytable () VALUES ()")
def test_supports_empty_insert_true(self):
table1 = self.tables.mytable
@@ -878,9 +920,9 @@ class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
dialect.supports_empty_insert = dialect.supports_default_values = True
stmt = table1.insert().values({}) # hide from 2to3
- self.assert_compile(stmt,
- 'INSERT INTO mytable DEFAULT VALUES',
- dialect=dialect)
+ self.assert_compile(
+ stmt, "INSERT INTO mytable DEFAULT VALUES", dialect=dialect
+ )
def test_supports_empty_insert_false(self):
table1 = self.tables.mytable
@@ -894,21 +936,24 @@ class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"The 'default' dialect with current database version "
"settings does not support empty inserts.",
stmt.compile,
- dialect=dialect)
+ dialect=dialect,
+ )
def _test_insert_with_empty_collection_values(self, collection):
table1 = self.tables.mytable
ins = table1.insert().values(collection)
- self.assert_compile(ins,
- 'INSERT INTO mytable () VALUES ()',
- checkparams={})
+ self.assert_compile(
+ ins, "INSERT INTO mytable () VALUES ()", checkparams={}
+ )
# empty dict populates on next values call
- self.assert_compile(ins.values(myid=3),
- 'INSERT INTO mytable (myid) VALUES (:myid)',
- checkparams={'myid': 3})
+ self.assert_compile(
+ ins.values(myid=3),
+ "INSERT INTO mytable (myid) VALUES (:myid)",
+ checkparams={"myid": 3},
+ )
def test_insert_with_empty_list_values(self):
self._test_insert_with_empty_collection_values([])
@@ -921,38 +966,40 @@ class EmptyTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
- __dialect__ = 'default'
+ __dialect__ = "default"
def test_not_supported(self):
table1 = self.tables.mytable
dialect = default.DefaultDialect()
- stmt = table1.insert().values([{'myid': 1}, {'myid': 2}])
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
assert_raises_message(
exc.CompileError,
"The 'default' dialect with current database version settings "
"does not support in-place multirow inserts.",
- stmt.compile, dialect=dialect)
+ stmt.compile,
+ dialect=dialect,
+ )
def test_named(self):
table1 = self.tables.mytable
values = [
- {'myid': 1, 'name': 'a', 'description': 'b'},
- {'myid': 2, 'name': 'c', 'description': 'd'},
- {'myid': 3, 'name': 'e', 'description': 'f'}
+ {"myid": 1, "name": "a", "description": "b"},
+ {"myid": 2, "name": "c", "description": "d"},
+ {"myid": 3, "name": "e", "description": "f"},
]
checkparams = {
- 'myid_m0': 1,
- 'myid_m1': 2,
- 'myid_m2': 3,
- 'name_m0': 'a',
- 'name_m1': 'c',
- 'name_m2': 'e',
- 'description_m0': 'b',
- 'description_m1': 'd',
- 'description_m2': 'f',
+ "myid_m0": 1,
+ "myid_m1": 2,
+ "myid_m2": 3,
+ "name_m0": "a",
+ "name_m1": "c",
+ "name_m2": "e",
+ "description_m0": "b",
+ "description_m1": "d",
+ "description_m2": "f",
}
dialect = default.DefaultDialect()
@@ -960,32 +1007,33 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
self.assert_compile(
table1.insert().values(values),
- 'INSERT INTO mytable (myid, name, description) VALUES '
- '(:myid_m0, :name_m0, :description_m0), '
- '(:myid_m1, :name_m1, :description_m1), '
- '(:myid_m2, :name_m2, :description_m2)',
+ "INSERT INTO mytable (myid, name, description) VALUES "
+ "(:myid_m0, :name_m0, :description_m0), "
+ "(:myid_m1, :name_m1, :description_m1), "
+ "(:myid_m2, :name_m2, :description_m2)",
checkparams=checkparams,
- dialect=dialect)
+ dialect=dialect,
+ )
def test_named_with_column_objects(self):
table1 = self.tables.mytable
values = [
- {table1.c.myid: 1, table1.c.name: 'a', table1.c.description: 'b'},
- {table1.c.myid: 2, table1.c.name: 'c', table1.c.description: 'd'},
- {table1.c.myid: 3, table1.c.name: 'e', table1.c.description: 'f'},
+ {table1.c.myid: 1, table1.c.name: "a", table1.c.description: "b"},
+ {table1.c.myid: 2, table1.c.name: "c", table1.c.description: "d"},
+ {table1.c.myid: 3, table1.c.name: "e", table1.c.description: "f"},
]
checkparams = {
- 'myid_m0': 1,
- 'myid_m1': 2,
- 'myid_m2': 3,
- 'name_m0': 'a',
- 'name_m1': 'c',
- 'name_m2': 'e',
- 'description_m0': 'b',
- 'description_m1': 'd',
- 'description_m2': 'f',
+ "myid_m0": 1,
+ "myid_m1": 2,
+ "myid_m2": 3,
+ "name_m0": "a",
+ "name_m1": "c",
+ "name_m2": "e",
+ "description_m0": "b",
+ "description_m1": "d",
+ "description_m2": "f",
}
dialect = default.DefaultDialect()
@@ -993,50 +1041,48 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
self.assert_compile(
table1.insert().values(values),
- 'INSERT INTO mytable (myid, name, description) VALUES '
- '(:myid_m0, :name_m0, :description_m0), '
- '(:myid_m1, :name_m1, :description_m1), '
- '(:myid_m2, :name_m2, :description_m2)',
+ "INSERT INTO mytable (myid, name, description) VALUES "
+ "(:myid_m0, :name_m0, :description_m0), "
+ "(:myid_m1, :name_m1, :description_m1), "
+ "(:myid_m2, :name_m2, :description_m2)",
checkparams=checkparams,
- dialect=dialect)
+ dialect=dialect,
+ )
def test_positional(self):
table1 = self.tables.mytable
values = [
- {'myid': 1, 'name': 'a', 'description': 'b'},
- {'myid': 2, 'name': 'c', 'description': 'd'},
- {'myid': 3, 'name': 'e', 'description': 'f'}
+ {"myid": 1, "name": "a", "description": "b"},
+ {"myid": 2, "name": "c", "description": "d"},
+ {"myid": 3, "name": "e", "description": "f"},
]
- checkpositional = (1, 'a', 'b', 2, 'c', 'd', 3, 'e', 'f')
+ checkpositional = (1, "a", "b", 2, "c", "d", 3, "e", "f")
dialect = default.DefaultDialect()
dialect.supports_multivalues_insert = True
- dialect.paramstyle = 'format'
+ dialect.paramstyle = "format"
dialect.positional = True
self.assert_compile(
table1.insert().values(values),
- 'INSERT INTO mytable (myid, name, description) VALUES '
- '(%s, %s, %s), (%s, %s, %s), (%s, %s, %s)',
+ "INSERT INTO mytable (myid, name, description) VALUES "
+ "(%s, %s, %s), (%s, %s, %s), (%s, %s, %s)",
checkpositional=checkpositional,
- dialect=dialect)
+ dialect=dialect,
+ )
def test_positional_w_defaults(self):
table1 = self.tables.table_w_defaults
- values = [
- {'id': 1},
- {'id': 2},
- {'id': 3}
- ]
+ values = [{"id": 1}, {"id": 2}, {"id": 3}]
checkpositional = (1, None, None, 2, None, None, 3, None, None)
dialect = default.DefaultDialect()
dialect.supports_multivalues_insert = True
- dialect.paramstyle = 'format'
+ dialect.paramstyle = "format"
dialect.positional = True
self.assert_compile(
@@ -1045,128 +1091,163 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"(%s, %s, %s), (%s, %s, %s), (%s, %s, %s)",
checkpositional=checkpositional,
check_prefetch=[
- table1.c.x, table1.c.z,
+ table1.c.x,
+ table1.c.z,
crud._multiparam_column(table1.c.x, 0),
crud._multiparam_column(table1.c.z, 0),
crud._multiparam_column(table1.c.x, 1),
- crud._multiparam_column(table1.c.z, 1)
+ crud._multiparam_column(table1.c.z, 1),
],
- dialect=dialect)
+ dialect=dialect,
+ )
def test_inline_default(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer, default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer, default=func.foobar()),
+ )
values = [
- {'id': 1, 'data': 'data1'},
- {'id': 2, 'data': 'data2', 'foo': 'plainfoo'},
- {'id': 3, 'data': 'data3'},
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": "plainfoo"},
+ {"id": 3, "data": "data3"},
]
checkparams = {
- 'id_m0': 1,
- 'id_m1': 2,
- 'id_m2': 3,
- 'data_m0': 'data1',
- 'data_m1': 'data2',
- 'data_m2': 'data3',
- 'foo_m1': 'plainfoo',
+ "id_m0": 1,
+ "id_m1": 2,
+ "id_m2": 3,
+ "data_m0": "data1",
+ "data_m1": "data2",
+ "data_m2": "data3",
+ "foo_m1": "plainfoo",
}
self.assert_compile(
table.insert().values(values),
- 'INSERT INTO sometable (id, data, foo) VALUES '
- '(%(id_m0)s, %(data_m0)s, foobar()), '
- '(%(id_m1)s, %(data_m1)s, %(foo_m1)s), '
- '(%(id_m2)s, %(data_m2)s, foobar())',
+ "INSERT INTO sometable (id, data, foo) VALUES "
+ "(%(id_m0)s, %(data_m0)s, foobar()), "
+ "(%(id_m1)s, %(data_m1)s, %(foo_m1)s), "
+ "(%(id_m2)s, %(data_m2)s, foobar())",
checkparams=checkparams,
- dialect=postgresql.dialect())
+ dialect=postgresql.dialect(),
+ )
def test_python_scalar_default(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer, default=10))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer, default=10),
+ )
values = [
- {'id': 1, 'data': 'data1'},
- {'id': 2, 'data': 'data2', 'foo': 15},
- {'id': 3, 'data': 'data3'},
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": 15},
+ {"id": 3, "data": "data3"},
]
checkparams = {
- 'id_m0': 1,
- 'id_m1': 2,
- 'id_m2': 3,
- 'data_m0': 'data1',
- 'data_m1': 'data2',
- 'data_m2': 'data3',
- 'foo': None, # evaluated later
- 'foo_m1': 15,
- 'foo_m2': None # evaluated later
+ "id_m0": 1,
+ "id_m1": 2,
+ "id_m2": 3,
+ "data_m0": "data1",
+ "data_m1": "data2",
+ "data_m2": "data3",
+ "foo": None, # evaluated later
+ "foo_m1": 15,
+ "foo_m2": None, # evaluated later
}
stmt = table.insert().values(values)
eq_(
- dict([
- (k, v.type._type_affinity)
- for (k, v) in
- stmt.compile(dialect=postgresql.dialect()).binds.items()]),
+ dict(
+ [
+ (k, v.type._type_affinity)
+ for (k, v) in stmt.compile(
+ dialect=postgresql.dialect()
+ ).binds.items()
+ ]
+ ),
{
- 'foo': Integer, 'data_m2': String, 'id_m0': Integer,
- 'id_m2': Integer, 'foo_m1': Integer, 'data_m1': String,
- 'id_m1': Integer, 'foo_m2': Integer, 'data_m0': String}
+ "foo": Integer,
+ "data_m2": String,
+ "id_m0": Integer,
+ "id_m2": Integer,
+ "foo_m1": Integer,
+ "data_m1": String,
+ "id_m1": Integer,
+ "foo_m2": Integer,
+ "data_m0": String,
+ },
)
self.assert_compile(
stmt,
- 'INSERT INTO sometable (id, data, foo) VALUES '
- '(%(id_m0)s, %(data_m0)s, %(foo)s), '
- '(%(id_m1)s, %(data_m1)s, %(foo_m1)s), '
- '(%(id_m2)s, %(data_m2)s, %(foo_m2)s)',
+ "INSERT INTO sometable (id, data, foo) VALUES "
+ "(%(id_m0)s, %(data_m0)s, %(foo)s), "
+ "(%(id_m1)s, %(data_m1)s, %(foo_m1)s), "
+ "(%(id_m2)s, %(data_m2)s, %(foo_m2)s)",
checkparams=checkparams,
- dialect=postgresql.dialect())
+ dialect=postgresql.dialect(),
+ )
def test_python_fn_default(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer, default=lambda: 10))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer, default=lambda: 10),
+ )
values = [
- {'id': 1, 'data': 'data1'},
- {'id': 2, 'data': 'data2', 'foo': 15},
- {'id': 3, 'data': 'data3'},
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": 15},
+ {"id": 3, "data": "data3"},
]
checkparams = {
- 'id_m0': 1,
- 'id_m1': 2,
- 'id_m2': 3,
- 'data_m0': 'data1',
- 'data_m1': 'data2',
- 'data_m2': 'data3',
- 'foo': None, # evaluated later
- 'foo_m1': 15,
- 'foo_m2': None, # evaluated later
+ "id_m0": 1,
+ "id_m1": 2,
+ "id_m2": 3,
+ "data_m0": "data1",
+ "data_m1": "data2",
+ "data_m2": "data3",
+ "foo": None, # evaluated later
+ "foo_m1": 15,
+ "foo_m2": None, # evaluated later
}
stmt = table.insert().values(values)
eq_(
- dict([
- (k, v.type._type_affinity)
- for (k, v) in
- stmt.compile(dialect=postgresql.dialect()).binds.items()]),
+ dict(
+ [
+ (k, v.type._type_affinity)
+ for (k, v) in stmt.compile(
+ dialect=postgresql.dialect()
+ ).binds.items()
+ ]
+ ),
{
- 'foo': Integer, 'data_m2': String, 'id_m0': Integer,
- 'id_m2': Integer, 'foo_m1': Integer, 'data_m1': String,
- 'id_m1': Integer, 'foo_m2': Integer, 'data_m0': String}
+ "foo": Integer,
+ "data_m2": String,
+ "id_m0": Integer,
+ "id_m2": Integer,
+ "foo_m1": Integer,
+ "data_m1": String,
+ "id_m1": Integer,
+ "foo_m2": Integer,
+ "data_m0": String,
+ },
)
self.assert_compile(
@@ -1176,14 +1257,18 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"(%(id_m1)s, %(data_m1)s, %(foo_m1)s), "
"(%(id_m2)s, %(data_m2)s, %(foo_m2)s)",
checkparams=checkparams,
- dialect=postgresql.dialect())
+ dialect=postgresql.dialect(),
+ )
def test_sql_functions(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer),
+ )
values = [
{"id": 1, "data": "foo", "foo": func.foob()},
@@ -1193,21 +1278,17 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
{"id": 5, "data": "bar", "foo": func.foob()},
]
checkparams = {
- 'id_m0': 1,
- 'data_m0': 'foo',
-
- 'id_m1': 2,
- 'data_m1': 'bar',
-
- 'id_m2': 3,
- 'data_m2': 'bar',
-
- 'id_m3': 4,
- 'data_m3': 'bar',
- 'foo_m3': 15,
-
- 'id_m4': 5,
- 'data_m4': 'bar'
+ "id_m0": 1,
+ "data_m0": "foo",
+ "id_m1": 2,
+ "data_m1": "bar",
+ "id_m2": 3,
+ "data_m2": "bar",
+ "id_m3": 4,
+ "data_m3": "bar",
+ "foo_m3": 15,
+ "id_m4": 5,
+ "data_m4": "bar",
}
self.assert_compile(
@@ -1219,50 +1300,58 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"(%(id_m3)s, %(data_m3)s, %(foo_m3)s), "
"(%(id_m4)s, %(data_m4)s, foob())",
checkparams=checkparams,
- dialect=postgresql.dialect())
+ dialect=postgresql.dialect(),
+ )
def test_server_default(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer, server_default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer, server_default=func.foobar()),
+ )
values = [
- {'id': 1, 'data': 'data1'},
- {'id': 2, 'data': 'data2', 'foo': 'plainfoo'},
- {'id': 3, 'data': 'data3'},
+ {"id": 1, "data": "data1"},
+ {"id": 2, "data": "data2", "foo": "plainfoo"},
+ {"id": 3, "data": "data3"},
]
checkparams = {
- 'id_m0': 1,
- 'id_m1': 2,
- 'id_m2': 3,
- 'data_m0': 'data1',
- 'data_m1': 'data2',
- 'data_m2': 'data3',
+ "id_m0": 1,
+ "id_m1": 2,
+ "id_m2": 3,
+ "data_m0": "data1",
+ "data_m1": "data2",
+ "data_m2": "data3",
}
self.assert_compile(
table.insert().values(values),
- 'INSERT INTO sometable (id, data) VALUES '
- '(%(id_m0)s, %(data_m0)s), '
- '(%(id_m1)s, %(data_m1)s), '
- '(%(id_m2)s, %(data_m2)s)',
+ "INSERT INTO sometable (id, data) VALUES "
+ "(%(id_m0)s, %(data_m0)s), "
+ "(%(id_m1)s, %(data_m1)s), "
+ "(%(id_m2)s, %(data_m2)s)",
checkparams=checkparams,
- dialect=postgresql.dialect())
+ dialect=postgresql.dialect(),
+ )
def test_server_default_absent_value(self):
metadata = MetaData()
- table = Table('sometable', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String),
- Column('foo', Integer, server_default=func.foobar()))
+ table = Table(
+ "sometable",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", String),
+ Column("foo", Integer, server_default=func.foobar()),
+ )
values = [
- {'id': 1, 'data': 'data1', 'foo': 'plainfoo'},
- {'id': 2, 'data': 'data2'},
- {'id': 3, 'data': 'data3', 'foo': 'otherfoo'},
+ {"id": 1, "data": "data1", "foo": "plainfoo"},
+ {"id": 2, "data": "data2"},
+ {"id": 3, "data": "data3", "foo": "otherfoo"},
]
assert_raises_message(
@@ -1270,5 +1359,5 @@ class MultirowTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"INSERT value for column sometable.foo is explicitly rendered "
"as a boundparameter in the VALUES clause; a Python-side value or "
"SQL expression is required",
- table.insert().values(values).compile
+ table.insert().values(values).compile,
)