diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-08-31 13:34:43 -0400 |
|---|---|---|
| committer | mike bayer <mike_mp@zzzcomputing.com> | 2021-09-02 14:01:56 +0000 |
| commit | bf1fe670513abeb1596bc5266f50db1ffe62f3bd (patch) | |
| tree | 95509c88bfe615271d5953db74b019ec7fb37b79 /test/sql | |
| parent | a4fc07fae69aa130fc032e7b5204f2e1fe3acaa3 (diff) | |
| download | sqlalchemy-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.py | 73 |
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(), |
