summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2021-08-31 13:34:43 -0400
committermike bayer <mike_mp@zzzcomputing.com>2021-09-02 14:01:56 +0000
commitbf1fe670513abeb1596bc5266f50db1ffe62f3bd (patch)
tree95509c88bfe615271d5953db74b019ec7fb37b79 /test/sql
parenta4fc07fae69aa130fc032e7b5204f2e1fe3acaa3 (diff)
downloadsqlalchemy-bf1fe670513abeb1596bc5266f50db1ffe62f3bd.tar.gz
Fix and test sequences w/ executemany in pre-exec scenarios
Fixed issue where an engine that had ``implicit_returning`` set to False would fail to function when PostgreSQL's "fast insertmany" feature were used in conjunction with a ``Sequence``, as well as if any kind of "executemany" with "return_defaults()" were used in conjunction with a ``Sequence``. Note that PostgreSQL "fast insertmany" uses "RETURNING" by definition, when the SQL statement is passed to the driver; overall, the ``implicit_returning`` flag is legacy and has no real use in modern SQLAlchemy, and will be deprecated in a separate change. Fixes: #6963 Change-Id: Id8e3dd50a21b9124f338067b0fdb57b8f608dca8
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_sequences.py73
1 files changed, 72 insertions, 1 deletions
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index d1d46afa3..a0fef99be 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -208,6 +208,8 @@ class SequenceExecTest(fixtures.TestBase):
("implicit_returning",),
("no_implicit_returning",),
("explicit_returning", testing.requires.returning),
+ ("return_defaults_no_implicit_returning", testing.requires.returning),
+ ("return_defaults_implicit_returning", testing.requires.returning),
argnames="returning",
)
@testing.requires.multivalues_inserts
@@ -221,7 +223,7 @@ class SequenceExecTest(fixtures.TestBase):
e = engines.testing_engine(
options={
- "implicit_returning": returning != "no_implicit_returning"
+ "implicit_returning": "no_implicit_returning" not in returning
}
)
metadata.create_all(e)
@@ -232,10 +234,79 @@ class SequenceExecTest(fixtures.TestBase):
)
if returning == "explicit_returning":
stmt = stmt.returning(t1.c.x)
+ elif "return_defaults" in returning:
+ stmt = stmt.return_defaults()
r = conn.execute(stmt)
if returning == "explicit_returning":
eq_(r.all(), [(1,), (2,), (3,)])
+ elif "return_defaults" in returning:
+ eq_(r.returned_defaults_rows, None)
+
+ # TODO: not sure what this is
+ eq_(r.inserted_primary_key_rows, [(None,)])
+
+ eq_(
+ conn.execute(t1.select().order_by(t1.c.x)).all(),
+ [(1, "d1"), (2, "d2"), (3, "d3")],
+ )
+
+ @testing.combinations(
+ ("implicit_returning",),
+ ("no_implicit_returning",),
+ (
+ "explicit_returning",
+ testing.requires.returning
+ + testing.requires.insert_executemany_returning,
+ ),
+ (
+ "return_defaults_no_implicit_returning",
+ testing.requires.returning
+ + testing.requires.insert_executemany_returning,
+ ),
+ (
+ "return_defaults_implicit_returning",
+ testing.requires.returning
+ + testing.requires.insert_executemany_returning,
+ ),
+ argnames="returning",
+ )
+ def test_seq_multivalues_executemany(
+ self, metadata, testing_engine, returning
+ ):
+ t1 = Table(
+ "t",
+ metadata,
+ Column("x", Integer, Sequence("my_seq"), primary_key=True),
+ Column("data", String(50)),
+ )
+
+ e = engines.testing_engine(
+ options={
+ "implicit_returning": "no_implicit_returning" not in returning
+ }
+ )
+ metadata.create_all(e)
+ with e.begin() as conn:
+
+ stmt = t1.insert()
+ if returning == "explicit_returning":
+ stmt = stmt.returning(t1.c.x)
+ elif "return_defaults" in returning:
+ stmt = stmt.return_defaults()
+
+ r = conn.execute(
+ stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+ )
+ if returning == "explicit_returning":
+ eq_(r.all(), [(1,), (2,), (3,)])
+ elif "return_defaults" in returning:
+ if "no_implicit_returning" in returning:
+ eq_(r.returned_defaults_rows, None)
+ eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+ else:
+ eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
+ eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
eq_(
conn.execute(t1.select().order_by(t1.c.x)).all(),