summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
authormike bayer <mike_mp@zzzcomputing.com>2021-10-31 20:48:33 +0000
committerGerrit Code Review <gerrit@ci3.zzzcomputing.com>2021-10-31 20:48:33 +0000
commitf5956e13d5bc456802a09a0ed89b3c596161e390 (patch)
treed02b0e34e549c9b00aaad6fef25edf7f000380e6 /test/sql
parenta1adb21cbe83ecce467194ab0e3277db581dfa4d (diff)
parentaa026c302c6b188a7e28508f9ecb603809b9e03f (diff)
downloadsqlalchemy-f5956e13d5bc456802a09a0ed89b3c596161e390.tar.gz
Merge "2.0 removals: LegacyRow, connectionless execution, close_with_result" into main
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_deprecations.py720
-rw-r--r--test/sql/test_resultset.py311
2 files changed, 194 insertions, 837 deletions
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index 9b74ab1fa..c70e474fe 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -43,7 +43,6 @@ from sqlalchemy.sql import update
from sqlalchemy.sql import visitors
from sqlalchemy.sql.selectable import LABEL_STYLE_NONE
from sqlalchemy.sql.selectable import SelectStatementGrouping
-from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assertions
from sqlalchemy.testing import AssertsCompiledSQL
@@ -55,7 +54,6 @@ from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
-from sqlalchemy.testing import not_in
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.util import compat
@@ -1199,182 +1197,6 @@ class KeyTargetingTest(fixtures.TablesTest):
dict(b="a1", q="c1"),
)
- def test_column_label_overlap_fallback(self, connection):
- content, bar = self.tables.content, self.tables.bar
- row = connection.execute(
- select(content.c.type.label("content_type"))
- ).first()
-
- not_in(content.c.type, row)
- not_in(bar.c.content_type, row)
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(sql.column("content_type"), row)
-
- row = connection.execute(
- select(func.now().label("content_type"))
- ).first()
- not_in(content.c.type, row)
- not_in(bar.c.content_type, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(sql.column("content_type"), row)
-
- def test_columnclause_schema_column_one(self, connection):
- keyed2 = self.tables.keyed2
-
- # this is addressed by [ticket:2932]
- # ColumnClause._compare_name_for_result allows the
- # columns which the statement is against to be lightweight
- # cols, which results in a more liberal comparison scheme
- a, b = sql.column("a"), sql.column("b")
- stmt = select(a, b).select_from(table("keyed2"))
- row = connection.execute(stmt).first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.b, row)
-
- def test_columnclause_schema_column_two(self, connection):
- keyed2 = self.tables.keyed2
-
- a, b = sql.column("a"), sql.column("b")
- stmt = select(keyed2.c.a, keyed2.c.b)
- row = connection.execute(stmt).first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(b, row)
-
- def test_columnclause_schema_column_three(self, connection):
- keyed2 = self.tables.keyed2
-
- # originally addressed by [ticket:2932], however liberalized
- # Column-targeting rules are deprecated
-
- a, b = sql.column("a"), sql.column("b")
- stmt = text("select a, b from keyed2").columns(a=CHAR, b=CHAR)
- row = connection.execute(stmt).first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.b, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(b, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.b, row)
-
- def test_columnclause_schema_column_four(self, connection):
- keyed2 = self.tables.keyed2
-
- # this is also addressed by [ticket:2932]
-
- a, b = sql.column("keyed2_a"), sql.column("keyed2_b")
- stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
- a, b
- )
- row = connection.execute(stmt).first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.b, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.keyed2_a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.keyed2_b, row)
-
- def test_columnclause_schema_column_five(self, connection):
- keyed2 = self.tables.keyed2
-
- # this is also addressed by [ticket:2932]
-
- stmt = text("select a AS keyed2_a, b AS keyed2_b from keyed2").columns(
- keyed2_a=CHAR, keyed2_b=CHAR
- )
- row = connection.execute(stmt).first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(keyed2.c.b, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.keyed2_a, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "The SelectBase.c and SelectBase.columns",
- ):
- in_(stmt.c.keyed2_b, row)
-
class PKIncrementTest(fixtures.TablesTest):
run_define_tables = "each"
@@ -1454,87 +1276,6 @@ class PKIncrementTest(fixtures.TablesTest):
self._test_autoincrement(conn)
-class ConnectionlessCursorResultTest(fixtures.TablesTest):
- __backend__ = True
-
- @classmethod
- def define_tables(cls, metadata):
- Table(
- "users",
- metadata,
- Column(
- "user_id", INT, primary_key=True, test_needs_autoincrement=True
- ),
- Column("user_name", VARCHAR(20)),
- test_needs_acid=True,
- )
-
- def test_connectionless_autoclose_rows_exhausted(self):
- users = self.tables.users
- with testing.db.begin() as conn:
- conn.execute(users.insert(), dict(user_id=1, user_name="john"))
-
- with testing.expect_deprecated_20(
- r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) method"
- ):
- result = testing.db.execute(text("select * from users"))
- connection = result.connection
- assert not connection.closed
- eq_(result.fetchone(), (1, "john"))
- assert not connection.closed
- eq_(result.fetchone(), None)
- assert connection.closed
-
- @testing.requires.returning
- def test_connectionless_autoclose_crud_rows_exhausted(self):
- users = self.tables.users
- stmt = (
- users.insert()
- .values(user_id=1, user_name="john")
- .returning(users.c.user_id)
- )
- with testing.expect_deprecated_20(
- r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) method"
- ):
- result = testing.db.execute(stmt)
- connection = result.connection
- assert not connection.closed
- eq_(result.fetchone(), (1,))
- assert not connection.closed
- eq_(result.fetchone(), None)
- assert connection.closed
-
- def test_connectionless_autoclose_no_rows(self):
- with testing.expect_deprecated_20(
- r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) method"
- ):
- result = testing.db.execute(text("select * from users"))
- connection = result.connection
- assert not connection.closed
- eq_(result.fetchone(), None)
- assert connection.closed
-
- @testing.requires.updateable_autoincrement_pks
- def test_connectionless_autoclose_no_metadata(self):
- with testing.expect_deprecated_20(
- r"The (?:Executable|Engine)\.(?:execute|scalar)\(\) method"
- ):
- result = testing.db.execute(text("update users set user_id=5"))
- connection = result.connection
- assert connection.closed
-
- assert_raises_message(
- exc.ResourceClosedError,
- "This result object does not return rows.",
- result.fetchone,
- )
-
- with testing.expect_deprecated_20(
- r"Calling the .keys\(\) method on a result set that does not "
- ):
- eq_(result.keys(), [])
-
-
class CursorResultTest(fixtures.TablesTest):
__backend__ = True
@@ -1583,233 +1324,6 @@ class CursorResultTest(fixtures.TablesTest):
],
)
- def test_column_accessor_textual_select(self, connection):
- users = self.tables.users
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names",
- "Using non-integer/slice indices on Row is "
- "deprecated and will be removed in version 2.0",
- ):
- # this will create column() objects inside
- # the select(), these need to match on name anyway
- r = connection.execute(
- select(column("user_id"), column("user_name"))
- .select_from(table("users"))
- .where(text("user_id=2"))
- ).first()
-
- eq_(r[users.c.user_id], 2)
-
- r._keymap.pop(users.c.user_id) # reset lookup
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(r._mapping[users.c.user_id], 2)
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(r._mapping[users.c.user_name], "jack")
-
- def test_keys_no_rows(self, connection):
-
- for i in range(2):
- r = connection.execute(
- text("update users set user_name='new' where user_id=10")
- )
-
- with testing.expect_deprecated(
- r"Calling the .keys\(\) method on a result set that does not "
- r"return rows is deprecated and will raise "
- r"ResourceClosedError in SQLAlchemy 2.0."
- ):
- list_ = r.keys()
- eq_(list_, [])
- list_.append("Don't cache me")
-
- def test_column_accessor_basic_text(self, connection):
- users = self.tables.users
-
- with testing.expect_deprecated(
- "Using non-integer/slice indices on Row is deprecated "
- "and will be removed in version 2.0",
- "Retrieving row values using Column objects "
- "with only matching names",
- ):
- r = connection.execute(
- text("select * from users where user_id=2")
- ).first()
-
- eq_(r[users.c.user_id], 2)
-
- r._keymap.pop(users.c.user_id)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(r._mapping[users.c.user_id], 2)
-
- with testing.expect_deprecated(
- "Using non-integer/slice indices on Row is deprecated "
- "and will be removed in version 2.0",
- "Retrieving row values using Column objects "
- "with only matching names",
- ):
- eq_(r[users.c.user_name], "jack")
-
- r._keymap.pop(users.c.user_name)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(r._mapping[users.c.user_name], "jack")
-
- @testing.provide_metadata
- def test_column_label_overlap_fallback(self, connection):
- content = Table("content", self.metadata, Column("type", String(30)))
- bar = Table("bar", self.metadata, Column("content_type", String(30)))
- self.metadata.create_all(testing.db)
- connection.execute(content.insert().values(type="t1"))
-
- row = connection.execute(
- content.select().set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
- ).first()
- in_(content.c.type, row._mapping)
- not_in(bar.c.content_type, row)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(sql.column("content_type"), row)
-
- row = connection.execute(
- select(content.c.type.label("content_type"))
- ).first()
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(content.c.type, row)
-
- not_in(bar.c.content_type, row)
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(sql.column("content_type"), row)
-
- row = connection.execute(
- select(func.now().label("content_type"))
- ).first()
-
- not_in(content.c.type, row)
-
- not_in(bar.c.content_type, row)
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- in_(sql.column("content_type"), row)
-
- def test_pickled_rows(self):
- users = self.tables.users
- addresses = self.tables.addresses
- with testing.db.begin() as conn:
- conn.execute(users.delete())
- conn.execute(
- users.insert(),
- [
- {"user_id": 7, "user_name": "jack"},
- {"user_id": 8, "user_name": "ed"},
- {"user_id": 9, "user_name": "fred"},
- ],
- )
-
- for pickle in False, True:
- for use_labels in False, True:
- stmt = users.select()
- if use_labels:
- stmt = stmt.set_label_style(
- LABEL_STYLE_TABLENAME_PLUS_COL
- )
-
- result = conn.execute(
- stmt.order_by(users.c.user_id)
- ).fetchall()
-
- if pickle:
- result = util.pickle.loads(util.pickle.dumps(result))
-
- if pickle:
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[users.c.user_id], 7)
-
- result[0]._keymap.pop(users.c.user_id)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[users.c.user_id], 7)
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[users.c.user_name], "jack")
-
- result[0]._keymap.pop(users.c.user_name)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[users.c.user_name], "jack")
-
- if not pickle or use_labels:
- assert_raises(
- exc.NoSuchColumnError,
- lambda: result[0][addresses.c.user_id],
- )
-
- assert_raises(
- exc.NoSuchColumnError,
- lambda: result[0]._mapping[addresses.c.user_id],
- )
- else:
- # test with a different table. name resolution is
- # causing 'user_id' to match when use_labels wasn't
- # used.
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[addresses.c.user_id], 7)
-
- result[0]._keymap.pop(addresses.c.user_id)
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "from a row that was unpickled"
- ):
- eq_(result[0]._mapping[addresses.c.user_id], 7)
-
- assert_raises(
- exc.NoSuchColumnError,
- lambda: result[0][addresses.c.address_id],
- )
-
- assert_raises(
- exc.NoSuchColumnError,
- lambda: result[0]._mapping[addresses.c.address_id],
- )
-
@testing.requires.duplicate_names_in_cursor_description
def test_ambiguous_column_case_sensitive(self):
with testing.expect_deprecated(
@@ -1831,30 +1345,6 @@ class CursorResultTest(fixtures.TablesTest):
lambda: row._mapping["somecol"],
)
- def test_row_getitem_string(self, connection):
- col = literal_column("1").label("foo")
-
- with testing.expect_deprecated(
- "Using non-integer/slice indices on Row is deprecated "
- "and will be removed in version 2.0;"
- ):
- row = connection.execute(select(col)).first()
- eq_(row["foo"], 1)
-
- eq_(row._mapping["foo"], 1)
-
- def test_row_getitem_column(self, connection):
- col = literal_column("1").label("foo")
-
- with testing.expect_deprecated(
- "Using non-integer/slice indices on Row is deprecated "
- "and will be removed in version 2.0;"
- ):
- row = connection.execute(select(col)).first()
- eq_(row[col], 1)
-
- eq_(row._mapping[col], 1)
-
def test_row_case_insensitive(self):
with testing.expect_deprecated(
"The create_engine.case_sensitive parameter is deprecated"
@@ -1914,126 +1404,6 @@ class CursorResultTest(fixtures.TablesTest):
eq_(row._mapping["casesensitive"], 2)
eq_(row._mapping["screw_UP_the_cols"], 3)
- def test_row_keys_deprecated(self, connection):
- r = connection.execute(
- text("select * from users where user_id=2")
- ).first()
-
- with testing.expect_deprecated_20(
- r"The Row.keys\(\) method is considered legacy "
- ):
- eq_(r.keys(), ["user_id", "user_name"])
-
- def test_row_contains_key_deprecated(self, connection):
- r = connection.execute(
- text("select * from users where user_id=2")
- ).first()
-
- with testing.expect_deprecated(
- "Using the 'in' operator to test for string or column keys, or "
- "integer indexes, .* is deprecated"
- ):
- in_("user_name", r)
-
- # no warning if the key is not there
- not_in("foobar", r)
-
- # this seems to happen only with Python BaseRow
- # with testing.expect_deprecated(
- # "Using the 'in' operator to test for string or column keys, or "
- # "integer indexes, .* is deprecated"
- # ):
- # in_(1, r)
-
-
-class PositionalTextTest(fixtures.TablesTest):
- run_inserts = "once"
- run_deletes = None
- __backend__ = True
-
- @classmethod
- def define_tables(cls, metadata):
- Table(
- "text1",
- metadata,
- Column("a", CHAR(2)),
- Column("b", CHAR(2)),
- Column("c", CHAR(2)),
- Column("d", CHAR(2)),
- )
-
- @classmethod
- def insert_data(cls, connection):
- connection.execute(
- cls.tables.text1.insert(),
- [dict(a="a1", b="b1", c="c1", d="d1")],
- )
-
- def test_anon_aliased_overlapping(self, connection):
- text1 = self.tables.text1
-
- c1 = text1.c.a.label(None)
- c2 = text1.alias().c.a
- c3 = text1.alias().c.a.label(None)
- c4 = text1.c.a.label(None)
-
- stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4)
- result = connection.execute(stmt)
- row = result.first()
-
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(row._mapping[text1.c.a], "a1")
-
- def test_anon_aliased_unique(self, connection):
- text1 = self.tables.text1
-
- c1 = text1.c.a.label(None)
- c2 = text1.alias().c.c
- c3 = text1.alias().c.b
- c4 = text1.alias().c.d.label(None)
-
- stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4)
- result = connection.execute(stmt)
- row = result.first()
-
- eq_(row._mapping[c1], "a1")
- eq_(row._mapping[c2], "b1")
- eq_(row._mapping[c3], "c1")
- eq_(row._mapping[c4], "d1")
-
- # key fallback rules still match this to a column
- # unambiguously based on its name
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(row._mapping[text1.c.a], "a1")
-
- # key fallback rules still match this to a column
- # unambiguously based on its name
- with testing.expect_deprecated(
- "Retrieving row values using Column objects "
- "with only matching names"
- ):
- eq_(row._mapping[text1.c.d], "d1")
-
- # text1.c.b goes nowhere....because we hit key fallback
- # but the text1.c.b doesn't derive from text1.c.c
- assert_raises_message(
- exc.NoSuchColumnError,
- "Could not locate column in row for column 'text1.b'",
- lambda: row[text1.c.b],
- )
-
- assert_raises_message(
- exc.NoSuchColumnError,
- "Could not locate column in row for column 'text1.b'",
- lambda: row._mapping[text1.c.b],
- )
-
class DefaultTest(fixtures.TestBase):
__backend__ = True
@@ -2610,96 +1980,6 @@ class LegacyOperatorTest(AssertsCompiledSQL, fixtures.TestBase):
assert _op_modern == _op_legacy
-class LegacySequenceExecTest(fixtures.TestBase):
- __requires__ = ("sequences",)
- __backend__ = True
-
- @classmethod
- def setup_test_class(cls):
- cls.seq = Sequence("my_sequence")
- cls.seq.create(testing.db)
-
- @classmethod
- def teardown_test_class(cls):
- cls.seq.drop(testing.db)
-
- def _assert_seq_result(self, ret):
- """asserts return of next_value is an int"""
-
- assert isinstance(ret, util.int_types)
- assert ret >= testing.db.dialect.default_sequence_base
-
- def test_implicit_connectionless(self):
- with testing.expect_deprecated_20(
- r"The MetaData.bind argument is deprecated"
- ):
- s = Sequence("my_sequence", metadata=MetaData(testing.db))
-
- with testing.expect_deprecated_20(
- r"The DefaultGenerator.execute\(\) method is considered legacy "
- "as of the 1.x",
- ):
- self._assert_seq_result(s.execute())
-
- def test_explicit(self, connection):
- s = Sequence("my_sequence")
- with testing.expect_deprecated_20(
- r"The DefaultGenerator.execute\(\) method is considered legacy"
- ):
- self._assert_seq_result(s.execute(connection))
-
- def test_explicit_optional(self):
- """test dialect executes a Sequence, returns nextval, whether
- or not "optional" is set"""
-
- s = Sequence("my_sequence", optional=True)
- with testing.expect_deprecated_20(
- r"The DefaultGenerator.execute\(\) method is considered legacy"
- ):
- self._assert_seq_result(s.execute(testing.db))
-
- def test_func_implicit_connectionless_execute(self):
- """test func.next_value().execute()/.scalar() works
- with connectionless execution."""
-
- with testing.expect_deprecated_20(
- r"The MetaData.bind argument is deprecated"
- ):
- s = Sequence("my_sequence", metadata=MetaData(testing.db))
- with testing.expect_deprecated_20(
- r"The Executable.execute\(\) method is considered legacy"
- ):
- self._assert_seq_result(s.next_value().execute().scalar())
-
- def test_func_explicit(self):
- s = Sequence("my_sequence")
- with testing.expect_deprecated_20(
- r"The Engine.scalar\(\) method is considered legacy"
- ):
- self._assert_seq_result(testing.db.scalar(s.next_value()))
-
- def test_func_implicit_connectionless_scalar(self):
- """test func.next_value().execute()/.scalar() works."""
-
- with testing.expect_deprecated_20(
- r"The MetaData.bind argument is deprecated"
- ):
- s = Sequence("my_sequence", metadata=MetaData(testing.db))
- with testing.expect_deprecated_20(
- r"The Executable.execute\(\) method is considered legacy"
- ):
- self._assert_seq_result(s.next_value().scalar())
-
- def test_func_embedded_select(self):
- """test can use next_value() in select column expr"""
-
- s = Sequence("my_sequence")
- with testing.expect_deprecated_20(
- r"The Engine.scalar\(\) method is considered legacy"
- ):
- self._assert_seq_result(testing.db.scalar(select(s.next_value())))
-
-
class DDLDeprecatedBindTest(fixtures.TestBase):
def teardown_test(self):
with testing.db.begin() as conn:
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index d2d2b1041..47f26ddab 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -30,8 +30,6 @@ from sqlalchemy.engine import default
from sqlalchemy.engine import Row
from sqlalchemy.engine.result import SimpleResultMetaData
from sqlalchemy.engine.row import KEY_INTEGER_ONLY
-from sqlalchemy.engine.row import KEY_OBJECTS_BUT_WARN
-from sqlalchemy.engine.row import LegacyRow
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import ColumnElement
from sqlalchemy.sql import expression
@@ -45,6 +43,7 @@ from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assertions
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
@@ -97,6 +96,47 @@ class CursorResultTest(fixtures.TablesTest):
test_needs_acid=True,
)
+ def test_keys_no_rows(self, connection):
+
+ for i in range(2):
+ r = connection.execute(
+ text("update users set user_name='new' where user_id=10")
+ )
+
+ with expect_raises_message(
+ exc.ResourceClosedError,
+ "This result object does not return rows",
+ ):
+ r.keys()
+
+ def test_row_keys_removed(self, connection):
+ r = connection.execute(
+ text("select * from users where user_id=2")
+ ).first()
+
+ with expect_raises(AttributeError):
+ r.keys()
+
+ def test_row_contains_key_no_strings(self, connection):
+ users = self.tables.users
+
+ connection.execute(
+ users.insert(),
+ [
+ dict(user_id=1, user_name="john"),
+ dict(user_id=2, user_name="jack"),
+ ],
+ )
+
+ r = connection.execute(
+ text("select * from users where user_id=2")
+ ).first()
+
+ not_in("user_name", r)
+ in_("user_name", r._mapping)
+ not_in("foobar", r)
+ not_in("foobar", r._mapping)
+
def test_row_iteration(self, connection):
users = self.tables.users
@@ -220,6 +260,9 @@ class CursorResultTest(fixtures.TablesTest):
in_(content.c.type, row._mapping)
not_in(bar.c.content_type, row._mapping)
+ # in 1.x, would warn for string match, but return a result
+ not_in(sql.column("content_type"), row)
+
not_in(bar.c.content_type, row._mapping)
row = connection.execute(
@@ -230,8 +273,12 @@ class CursorResultTest(fixtures.TablesTest):
not_in(bar.c.content_type, row._mapping)
+ # in 1.x, would warn for string match, but return a result
+ not_in(sql.column("content_type"), row._mapping)
+
def test_pickled_rows(self, connection):
users = self.tables.users
+ addresses = self.tables.addresses
connection.execute(
users.insert(),
@@ -271,12 +318,40 @@ class CursorResultTest(fixtures.TablesTest):
eq_(result[0][0], 7)
assert_raises(
- exc.NoSuchColumnError, lambda: result[0]["fake key"]
+ exc.NoSuchColumnError,
+ lambda: result[0]._mapping["fake key"],
)
+ # previously would warn
+
+ if pickle:
+ with expect_raises_message(
+ exc.NoSuchColumnError,
+ "Row was unpickled; lookup by ColumnElement is "
+ "unsupported",
+ ):
+ result[0]._mapping[users.c.user_id]
+ else:
+ eq_(result[0]._mapping[users.c.user_id], 7)
+
+ if pickle:
+ with expect_raises_message(
+ exc.NoSuchColumnError,
+ "Row was unpickled; lookup by ColumnElement is "
+ "unsupported",
+ ):
+ result[0]._mapping[users.c.user_name]
+ else:
+ eq_(result[0]._mapping[users.c.user_name], "jack")
+
assert_raises(
exc.NoSuchColumnError,
- lambda: result[0]._mapping["fake key"],
+ lambda: result[0]._mapping[addresses.c.user_id],
+ )
+
+ assert_raises(
+ exc.NoSuchColumnError,
+ lambda: result[0]._mapping[addresses.c.address_id],
)
def test_column_error_printing(self, connection):
@@ -460,8 +535,21 @@ class CursorResultTest(fixtures.TablesTest):
text("select * from users where user_id=2")
).first()
+ with expect_raises_message(TypeError, "tuple indices must be"):
+ r["foo"]
+
eq_(r._mapping["user_name"], "jack")
+ def test_row_getitem_column(self, connection):
+ col = literal_column("1").label("foo")
+
+ row = connection.execute(select(col)).first()
+
+ with expect_raises_message(TypeError, "tuple indices must be"):
+ row[col]
+
+ eq_(row._mapping[col], 1)
+
def test_column_accessor_basic_text(self, connection):
users = self.tables.users
@@ -485,6 +573,16 @@ class CursorResultTest(fixtures.TablesTest):
eq_(r.user_name, "jack")
eq_(r._mapping["user_name"], "jack")
+ # cases which used to succeed w warning
+ with expect_raises_message(
+ exc.NoSuchColumnError, "Could not locate column in row"
+ ):
+ r._mapping[users.c.user_id]
+ with expect_raises_message(
+ exc.NoSuchColumnError, "Could not locate column in row"
+ ):
+ r._mapping[users.c.user_name]
+
def test_column_accessor_text_colexplicit(self, connection):
users = self.tables.users
@@ -534,6 +632,16 @@ class CursorResultTest(fixtures.TablesTest):
eq_(r.user_name, "jack")
eq_(r._mapping["user_name"], "jack")
+ # error cases that previously would warn
+ with expect_raises_message(
+ exc.NoSuchColumnError, "Could not locate column in row"
+ ):
+ r._mapping[users.c.user_id]
+ with expect_raises_message(
+ exc.NoSuchColumnError, "Could not locate column in row"
+ ):
+ r._mapping[users.c.user_name]
+
def test_column_accessor_dotted_union(self, connection):
users = self.tables.users
@@ -651,12 +759,7 @@ class CursorResultTest(fixtures.TablesTest):
lambda: r._mapping["foo"],
)
- @testing.combinations(
- (True,),
- (False,),
- argnames="future",
- )
- def test_graceful_fetch_on_non_rows(self, future):
+ def test_graceful_fetch_on_non_rows(self):
"""test that calling fetchone() etc. on a result that doesn't
return rows fails gracefully.
@@ -671,8 +774,6 @@ class CursorResultTest(fixtures.TablesTest):
users = self.tables.users
conn = testing.db.connect()
- if future:
- conn = conn.execution_options(future_result=True)
keys_lambda = lambda r: r.keys() # noqa: E731
for meth in [
@@ -689,20 +790,13 @@ class CursorResultTest(fixtures.TablesTest):
trans = conn.begin()
result = conn.execute(users.insert(), dict(user_id=1))
- if not future and meth is keys_lambda:
- with testing.expect_deprecated(
- r"Calling the .keys\(\) method on a result set that does "
- r"not return rows is deprecated"
- ):
- eq_(meth(result), [])
- else:
- assert_raises_message(
- exc.ResourceClosedError,
- "This result object does not return rows. "
- "It has been closed automatically.",
- meth,
- result,
- )
+ assert_raises_message(
+ exc.ResourceClosedError,
+ "This result object does not return rows. "
+ "It has been closed automatically.",
+ meth,
+ result,
+ )
trans.rollback()
def test_fetchone_til_end(self, connection):
@@ -1095,29 +1189,17 @@ class CursorResultTest(fixtures.TablesTest):
(lambda result: result.first()._mapping),
argnames="get_object",
)
- @testing.combinations(
- (True,),
- (False,),
- argnames="future",
- )
- def test_keys(self, connection, get_object, future):
+ def test_keys(self, connection, get_object):
users = self.tables.users
addresses = self.tables.addresses
- if future:
- connection = connection.execution_options(future_result=True)
-
connection.execute(users.insert(), dict(user_id=1, user_name="foo"))
result = connection.execute(users.select())
obj = get_object(result)
- # Row still has a .keys() method as well as LegacyRow
- # as in 1.3.x, the KeyedTuple object also had a keys() method.
- # it emits a 2.0 deprecation warning.
if isinstance(obj, Row):
- with assertions.expect_deprecated_20("The Row.keys()"):
- keys = obj.keys()
+ keys = obj._mapping.keys()
else:
keys = obj.keys()
@@ -1147,12 +1229,9 @@ class CursorResultTest(fixtures.TablesTest):
eq_(list(row._mapping.keys()), ["user_id", "user_name"])
eq_(row._fields, ("user_id", "user_name"))
- with assertions.expect_deprecated_20("The Row.keys()"):
- in_("user_id", row.keys())
- with assertions.expect_deprecated_20("The Row.keys()"):
- not_in("foo", row.keys())
- with assertions.expect_deprecated_20("The Row.keys()"):
- in_(users.c.user_id, row.keys())
+ in_("user_id", row._fields)
+ not_in("foo", row._fields)
+ in_(users.c.user_id, row._mapping.keys())
def test_row_keys_legacy_dont_warn(self, connection):
users = self.tables.users
@@ -1160,13 +1239,10 @@ class CursorResultTest(fixtures.TablesTest):
connection.execute(users.insert(), dict(user_id=1, user_name="foo"))
result = connection.execute(users.select())
row = result.first()
- # DO NOT WARN DEPRECATED IN 1.x, ONLY 2.0 WARNING
- with assertions.expect_deprecated_20("The Row.keys()"):
- eq_(dict(row), {"user_id": 1, "user_name": "foo"})
+ eq_(dict(row._mapping), {"user_id": 1, "user_name": "foo"})
- with assertions.expect_deprecated_20("The Row.keys()"):
- eq_(row.keys(), ["user_id", "user_name"])
+ eq_(row._fields, ("user_id", "user_name"))
def test_row_namedtuple_legacy_ok(self, connection):
users = self.tables.users
@@ -1395,14 +1471,13 @@ class CursorResultTest(fixtures.TablesTest):
)
is_true(isinstance(row, collections_abc.Sequence))
- @testing.combinations((Row,), (LegacyRow,))
- def test_row_special_names(self, row_cls):
+ def test_row_special_names(self):
metadata = SimpleResultMetaData(["key", "count", "index", "foo"])
- row = row_cls(
+ row = Row(
metadata,
[None, None, None, None],
metadata._keymap,
- row_cls._default_key_style,
+ Row._default_key_style,
["kv", "cv", "iv", "f"],
)
is_true(isinstance(row, collections_abc.Sequence))
@@ -1411,21 +1486,13 @@ class CursorResultTest(fixtures.TablesTest):
eq_(row.count, "cv")
eq_(row.index, "iv")
- with assertions.expect_deprecated_20(
- "Retrieving row members using strings or other non-integers "
- "is deprecated; use row._mapping for a dictionary interface "
- "to the row"
- ):
- eq_(row["foo"], "f")
- eq_(row["count"], "cv")
- eq_(row["index"], "iv")
-
+ eq_(row._mapping["foo"], "f")
eq_(row._mapping["count"], "cv")
eq_(row._mapping["index"], "iv")
metadata = SimpleResultMetaData(["key", "q", "p"])
- row = row_cls(
+ row = Row(
metadata,
[None, None, None],
metadata._keymap,
@@ -1441,45 +1508,6 @@ class CursorResultTest(fixtures.TablesTest):
eq_(row.count("cv"), 1)
eq_(row.count("x"), 0)
- @testing.combinations((Row,), (LegacyRow,))
- def test_row_dict_behaviors_warn_mode(self, row_cls):
- metadata = SimpleResultMetaData(
- [
- "a",
- "b",
- "count",
- ]
- )
- row = row_cls(
- metadata,
- [None, None, None],
- metadata._keymap,
- KEY_OBJECTS_BUT_WARN,
- ["av", "bv", "cv"],
- )
-
- # as of #6218, dict(row) and row["x"] work for
- # both LegacyRow and Row, with 2.0 deprecation warnings
- # for both
- with assertions.expect_deprecated_20(
- "Retrieving row members using strings or other non-integers "
- "is deprecated; use row._mapping for a dictionary interface "
- "to the row"
- ):
- eq_(dict(row), {"a": "av", "b": "bv", "count": "cv"})
-
- with assertions.expect_deprecated_20(
- "Retrieving row members using strings or other non-integers "
- "is deprecated; use row._mapping for a dictionary interface "
- "to the row"
- ):
- eq_(row["a"], "av")
- eq_(row["count"], "cv")
-
- # keys is keys
- with assertions.expect_deprecated_20("The Row.keys()"):
- eq_(list(row.keys()), ["a", "b", "count"])
-
def test_new_row_no_dict_behaviors(self):
"""This mode is not used currently but will be once we are in 2.0."""
metadata = SimpleResultMetaData(
@@ -1497,12 +1525,7 @@ class CursorResultTest(fixtures.TablesTest):
["av", "bv", "cv"],
)
- with assertions.expect_raises_message(
- TypeError,
- "TypeError: tuple indices must be integers or slices, not str",
- ):
- with assertions.expect_deprecated_20("The Row.keys()"):
- eq_(dict(row), {"a": "av", "b": "bv", "count": "cv"})
+ eq_(dict(row._mapping), {"a": "av", "b": "bv", "count": "cv"})
with assertions.expect_raises_message(
TypeError,
@@ -1516,9 +1539,7 @@ class CursorResultTest(fixtures.TablesTest):
):
eq_(row["count"], "cv")
- # keys is keys
- with assertions.expect_deprecated_20("The Row.keys()"):
- eq_(list(row.keys()), ["a", "b", "count"])
+ eq_(list(row._mapping), ["a", "b", "count"])
def test_row_is_hashable(self):
@@ -1913,8 +1934,6 @@ class KeyTargetingTest(fixtures.TablesTest):
eq_(row.keyed2_a, "a2")
eq_(row.keyed2_b, "b2")
- assert_raises(KeyError, lambda: row["keyed2_c"])
- assert_raises(KeyError, lambda: row["keyed2_q"])
assert_raises(KeyError, lambda: row._mapping["keyed2_c"])
assert_raises(KeyError, lambda: row._mapping["keyed2_q"])
@@ -1992,6 +2011,11 @@ class KeyTargetingTest(fixtures.TablesTest):
in_(a, row._mapping)
in_(b, row._mapping)
+ keyed2 = self.tables.keyed2
+
+ not_in(keyed2.c.a, row._mapping)
+ not_in(keyed2.c.b, row._mapping)
+
def test_columnclause_schema_column_two(self, connection):
keyed2 = self.tables.keyed2
@@ -2001,6 +2025,11 @@ class KeyTargetingTest(fixtures.TablesTest):
in_(keyed2.c.a, row._mapping)
in_(keyed2.c.b, row._mapping)
+ # in 1.x, would warn for string match, but return a result
+ a, b = sql.column("a"), sql.column("b")
+ not_in(a, row._mapping)
+ not_in(b, row._mapping)
+
def test_columnclause_schema_column_three(self, connection):
# this is also addressed by [ticket:2932]
stmt = text("select a, b from keyed2").columns(a=CHAR, b=CHAR)
@@ -2009,6 +2038,17 @@ class KeyTargetingTest(fixtures.TablesTest):
in_(stmt.selected_columns.a, row._mapping)
in_(stmt.selected_columns.b, row._mapping)
+ keyed2 = self.tables.keyed2
+ a, b = sql.column("a"), sql.column("b")
+
+ # in 1.x, would warn for string match, but return a result
+ not_in(keyed2.c.a, row._mapping)
+ not_in(keyed2.c.b, row._mapping)
+ not_in(a, row._mapping)
+ not_in(b, row._mapping)
+ not_in(stmt.subquery().c.a, row._mapping)
+ not_in(stmt.subquery().c.b, row._mapping)
+
def test_columnclause_schema_column_four(self, connection):
# originally addressed by [ticket:2932], however liberalized
# Column-targeting rules are deprecated
@@ -2025,6 +2065,14 @@ class KeyTargetingTest(fixtures.TablesTest):
in_(stmt.selected_columns.keyed2_a, row._mapping)
in_(stmt.selected_columns.keyed2_b, row._mapping)
+ keyed2 = self.tables.keyed2
+
+ # in 1.x, would warn for string match, but return a result
+ not_in(keyed2.c.a, row._mapping)
+ not_in(keyed2.c.b, row._mapping)
+ not_in(stmt.subquery().c.keyed2_a, row._mapping)
+ not_in(stmt.subquery().c.keyed2_b, row._mapping)
+
def test_columnclause_schema_column_five(self, connection):
# this is also addressed by [ticket:2932]
@@ -2036,6 +2084,14 @@ class KeyTargetingTest(fixtures.TablesTest):
in_(stmt.selected_columns.keyed2_a, row._mapping)
in_(stmt.selected_columns.keyed2_b, row._mapping)
+ keyed2 = self.tables.keyed2
+
+ # in 1.x, would warn for string match, but return a result
+ not_in(keyed2.c.a, row._mapping)
+ not_in(keyed2.c.b, row._mapping)
+ not_in(stmt.subquery().c.keyed2_a, row._mapping)
+ not_in(stmt.subquery().c.keyed2_b, row._mapping)
+
def _adapt_result_columns_fixture_one(self):
keyed1 = self.tables.keyed1
stmt = (
@@ -2164,14 +2220,14 @@ class KeyTargetingTest(fixtures.TablesTest):
cache = {}
result = connection._execute_20(
stmt1,
- execution_options={"compiled_cache": cache, "future_result": True},
+ execution_options={"compiled_cache": cache},
)
result.close()
assert cache
result = connection._execute_20(
stmt2,
- execution_options={"compiled_cache": cache, "future_result": True},
+ execution_options={"compiled_cache": cache},
)
row = result.first()
@@ -2343,6 +2399,20 @@ class PositionalTextTest(fixtures.TablesTest):
eq_(row._mapping[c3], "c1")
eq_(row._mapping[c4], "d1")
+ # in 1.x, would warn for string match, but return a result
+ assert_raises_message(
+ exc.NoSuchColumnError,
+ "Could not locate column in row for column 'text1.a'",
+ lambda: row._mapping[text1.c.a],
+ )
+
+ # in 1.x, would warn for string match, but return a result
+ assert_raises_message(
+ exc.NoSuchColumnError,
+ "Could not locate column in row for column 'text1.d'",
+ lambda: row._mapping[text1.c.d],
+ )
+
# text1.c.b goes nowhere....because we hit key fallback
# but the text1.c.b doesn't derive from text1.c.c
assert_raises_message(
@@ -2368,6 +2438,13 @@ class PositionalTextTest(fixtures.TablesTest):
eq_(row._mapping[c3], "c1")
eq_(row._mapping[c4], "d1")
+ # in 1.x, would warn for string match, but return a result
+ assert_raises_message(
+ exc.NoSuchColumnError,
+ "Could not locate column in row for column 'text1.a'",
+ lambda: row._mapping[text1.c.a],
+ )
+
def test_anon_aliased_name_conflict(self, connection):
text1 = self.tables.text1