Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/testing/suite»

venv/Lib/site-packages/sqlalchemy/testing/suite/test_insert.py

@@ -0,0 +1,630 @@
+# testing/suite/test_insert.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+from decimal import Decimal
+import uuid
+
+from . import testing
+from .. import fixtures
+from ..assertions import eq_
+from ..config import requirements
+from ..schema import Column
+from ..schema import Table
+from ... import Double
+from ... import Float
+from ... import Identity
+from ... import Integer
+from ... import literal
+from ... import literal_column
+from ... import Numeric
+from ... import select
+from ... import String
+from ...types import LargeBinary
+from ...types import UUID
+from ...types import Uuid
+
+
+class LastrowidTest(fixtures.TablesTest):
+    run_deletes = "each"
+
+    __backend__ = True
+
+    __requires__ = "implements_get_lastrowid", "autoincrement_insert"
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "autoinc_pk",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+
+        Table(
+            "manual_pk",
+            metadata,
+            Column("id", Integer, primary_key=True, autoincrement=False),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+
+    def _assert_round_trip(self, table, conn):
+        row = conn.execute(table.select()).first()
+        eq_(
+            row,
+            (
+                conn.dialect.default_sequence_base,
+                "some data",
+            ),
+        )
+
+    def test_autoincrement_on_insert(self, connection):
+        connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        self._assert_round_trip(self.tables.autoinc_pk, connection)
+
+    def test_last_inserted_id(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        pk = connection.scalar(select(self.tables.autoinc_pk.c.id))
+        eq_(r.inserted_primary_key, (pk,))
+
+    @requirements.dbapi_lastrowid
+    def test_native_lastrowid_autoinc(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        lastrowid = r.lastrowid
+        pk = connection.scalar(select(self.tables.autoinc_pk.c.id))
+        eq_(lastrowid, pk)
+
+
+class InsertBehaviorTest(fixtures.TablesTest):
+    run_deletes = "each"
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "autoinc_pk",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+        )
+        Table(
+            "manual_pk",
+            metadata,
+            Column("id", Integer, primary_key=True, autoincrement=False),
+            Column("data", String(50)),
+        )
+        Table(
+            "no_implicit_returning",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+        Table(
+            "includes_defaults",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+            Column("x", Integer, default=5),
+            Column(
+                "y",
+                Integer,
+                default=literal_column("2", type_=Integer) + literal(2),
+            ),
+        )
+
+    @testing.variation("style", ["plain", "return_defaults"])
+    @testing.variation("executemany", [True, False])
+    def test_no_results_for_non_returning_insert(
+        self, connection, style, executemany
+    ):
+        """test another INSERT issue found during #10453"""
+
+        table = self.tables.no_implicit_returning
+
+        stmt = table.insert()
+        if style.return_defaults:
+            stmt = stmt.return_defaults()
+
+        if executemany:
+            data = [
+                {"data": "d1"},
+                {"data": "d2"},
+                {"data": "d3"},
+                {"data": "d4"},
+                {"data": "d5"},
+            ]
+        else:
+            data = {"data": "d1"}
+
+        r = connection.execute(stmt, data)
+        assert not r.returns_rows
+
+    @requirements.autoincrement_insert
+    def test_autoclose_on_insert(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        assert r._soft_closed
+        assert not r.closed
+        assert r.is_insert
+
+        # new as of I8091919d45421e3f53029b8660427f844fee0228; for the moment
+        # an insert where the PK was taken from a row that the dialect
+        # selected, as is the case for mssql/pyodbc, will still report
+        # returns_rows as true because there's a cursor description.  in that
+        # case, the row had to have been consumed at least.
+        assert not r.returns_rows or r.fetchone() is None
+
+    @requirements.insert_returning
+    def test_autoclose_on_insert_implicit_returning(self, connection):
+        r = connection.execute(
+            # return_defaults() ensures RETURNING will be used,
+            # new in 2.0 as sqlite/mariadb offer both RETURNING and
+            # cursor.lastrowid
+            self.tables.autoinc_pk.insert().return_defaults(),
+            dict(data="some data"),
+        )
+        assert r._soft_closed
+        assert not r.closed
+        assert r.is_insert
+
+        # note we are experimenting with having this be True
+        # as of I8091919d45421e3f53029b8660427f844fee0228 .
+        # implicit returning has fetched the row, but it still is a
+        # "returns rows"
+        assert r.returns_rows
+
+        # and we should be able to fetchone() on it, we just get no row
+        eq_(r.fetchone(), None)
+
+        # and the keys, etc.
+        eq_(r.keys(), ["id"])
+
+        # but the dialect took in the row already.   not really sure
+        # what the best behavior is.
+
+    @requirements.empty_inserts
+    def test_empty_insert(self, connection):
+        r = connection.execute(self.tables.autoinc_pk.insert())
+        assert r._soft_closed
+        assert not r.closed
+
+        r = connection.execute(
+            self.tables.autoinc_pk.select().where(
+                self.tables.autoinc_pk.c.id != None
+            )
+        )
+        eq_(len(r.all()), 1)
+
+    @requirements.empty_inserts_executemany
+    def test_empty_insert_multiple(self, connection):
+        r = connection.execute(self.tables.autoinc_pk.insert(), [{}, {}, {}])
+        assert r._soft_closed
+        assert not r.closed
+
+        r = connection.execute(
+            self.tables.autoinc_pk.select().where(
+                self.tables.autoinc_pk.c.id != None
+            )
+        )
+
+        eq_(len(r.all()), 3)
+
+    @requirements.insert_from_select
+    def test_insert_from_select_autoinc(self, connection):
+        src_table = self.tables.manual_pk
+        dest_table = self.tables.autoinc_pk
+        connection.execute(
+            src_table.insert(),
+            [
+                dict(id=1, data="data1"),
+                dict(id=2, data="data2"),
+                dict(id=3, data="data3"),
+            ],
+        )
+
+        result = connection.execute(
+            dest_table.insert().from_select(
+                ("data",),
+                select(src_table.c.data).where(
+                    src_table.c.data.in_(["data2", "data3"])
+                ),
+            )
+        )
+
+        eq_(result.inserted_primary_key, (None,))
+
+        result = connection.execute(
+            select(dest_table.c.data).order_by(dest_table.c.data)
+        )
+        eq_(result.fetchall(), [("data2",), ("data3",)])
+
+    @requirements.insert_from_select
+    def test_insert_from_select_autoinc_no_rows(self, connection):
+        src_table = self.tables.manual_pk
+        dest_table = self.tables.autoinc_pk
+
+        result = connection.execute(
+            dest_table.insert().from_select(
+                ("data",),
+                select(src_table.c.data).where(
+                    src_table.c.data.in_(["data2", "data3"])
+                ),
+            )
+        )
+        eq_(result.inserted_primary_key, (None,))
+
+        result = connection.execute(
+            select(dest_table.c.data).order_by(dest_table.c.data)
+        )
+
+        eq_(result.fetchall(), [])
+
+    @requirements.insert_from_select
+    def test_insert_from_select(self, connection):
+        table = self.tables.manual_pk
+        connection.execute(
+            table.insert(),
+            [
+                dict(id=1, data="data1"),
+                dict(id=2, data="data2"),
+                dict(id=3, data="data3"),
+            ],
+        )
+
+        connection.execute(
+            table.insert()
+            .inline()
+            .from_select(
+                ("id", "data"),
+                select(table.c.id + 5, table.c.data).where(
+                    table.c.data.in_(["data2", "data3"])
+                ),
+            )
+        )
+
+        eq_(
+            connection.execute(
+                select(table.c.data).order_by(table.c.data)
+            ).fetchall(),
+            [("data1",), ("data2",), ("data2",), ("data3",), ("data3",)],
+        )
+
+    @requirements.insert_from_select
+    def test_insert_from_select_with_defaults(self, connection):
+        table = self.tables.includes_defaults
+        connection.execute(
+            table.insert(),
+            [
+                dict(id=1, data="data1"),
+                dict(id=2, data="data2"),
+                dict(id=3, data="data3"),
+            ],
+        )
+
+        connection.execute(
+            table.insert()
+            .inline()
+            .from_select(
+                ("id", "data"),
+                select(table.c.id + 5, table.c.data).where(
+                    table.c.data.in_(["data2", "data3"])
+                ),
+            )
+        )
+
+        eq_(
+            connection.execute(
+                select(table).order_by(table.c.data, table.c.id)
+            ).fetchall(),
+            [
+                (1, "data1", 5, 4),
+                (2, "data2", 5, 4),
+                (7, "data2", 5, 4),
+                (3, "data3", 5, 4),
+                (8, "data3", 5, 4),
+            ],
+        )
+
+
+class ReturningTest(fixtures.TablesTest):
+    run_create_tables = "each"
+    __requires__ = "insert_returning", "autoincrement_insert"
+    __backend__ = True
+
+    def _assert_round_trip(self, table, conn):
+        row = conn.execute(table.select()).first()
+        eq_(
+            row,
+            (
+                conn.dialect.default_sequence_base,
+                "some data",
+            ),
+        )
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "autoinc_pk",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+        )
+
+    @requirements.fetch_rows_post_commit
+    def test_explicit_returning_pk_autocommit(self, connection):
+        table = self.tables.autoinc_pk
+        r = connection.execute(
+            table.insert().returning(table.c.id), dict(data="some data")
+        )
+        pk = r.first()[0]
+        fetched_pk = connection.scalar(select(table.c.id))
+        eq_(fetched_pk, pk)
+
+    def test_explicit_returning_pk_no_autocommit(self, connection):
+        table = self.tables.autoinc_pk
+        r = connection.execute(
+            table.insert().returning(table.c.id), dict(data="some data")
+        )
+
+        pk = r.first()[0]
+        fetched_pk = connection.scalar(select(table.c.id))
+        eq_(fetched_pk, pk)
+
+    def test_autoincrement_on_insert_implicit_returning(self, connection):
+        connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        self._assert_round_trip(self.tables.autoinc_pk, connection)
+
+    def test_last_inserted_id_implicit_returning(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
+        pk = connection.scalar(select(self.tables.autoinc_pk.c.id))
+        eq_(r.inserted_primary_key, (pk,))
+
+    @requirements.insert_executemany_returning
+    def test_insertmanyvalues_returning(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert().returning(
+                self.tables.autoinc_pk.c.id
+            ),
+            [
+                {"data": "d1"},
+                {"data": "d2"},
+                {"data": "d3"},
+                {"data": "d4"},
+                {"data": "d5"},
+            ],
+        )
+        rall = r.all()
+
+        pks = connection.execute(select(self.tables.autoinc_pk.c.id))
+
+        eq_(rall, pks.all())
+
+    @testing.combinations(
+        (Double(), 8.5514716, True),
+        (
+            Double(53),
+            8.5514716,
+            True,
+            testing.requires.float_or_double_precision_behaves_generically,
+        ),
+        (Float(), 8.5514, True),
+        (
+            Float(8),
+            8.5514,
+            True,
+            testing.requires.float_or_double_precision_behaves_generically,
+        ),
+        (
+            Numeric(precision=15, scale=12, asdecimal=False),
+            8.5514716,
+            True,
+            testing.requires.literal_float_coercion,
+        ),
+        (
+            Numeric(precision=15, scale=12, asdecimal=True),
+            Decimal("8.5514716"),
+            False,
+        ),
+        argnames="type_,value,do_rounding",
+    )
+    @testing.variation("sort_by_parameter_order", [True, False])
+    @testing.variation("multiple_rows", [True, False])
+    def test_insert_w_floats(
+        self,
+        connection,
+        metadata,
+        sort_by_parameter_order,
+        type_,
+        value,
+        do_rounding,
+        multiple_rows,
+    ):
+        """test #9701.
+
+        this tests insertmanyvalues as well as decimal / floating point
+        RETURNING types
+
+        """
+
+        t = Table(
+            # Oracle backends seems to be getting confused if
+            # this table is named the same as the one
+            # in test_imv_returning_datatypes.  use a different name
+            "f_t",
+            metadata,
+            Column("id", Integer, Identity(), primary_key=True),
+            Column("value", type_),
+        )
+
+        t.create(connection)
+
+        result = connection.execute(
+            t.insert().returning(
+                t.c.id,
+                t.c.value,
+                sort_by_parameter_order=bool(sort_by_parameter_order),
+            ),
+            (
+                [{"value": value} for i in range(10)]
+                if multiple_rows
+                else {"value": value}
+            ),
+        )
+
+        if multiple_rows:
+            i_range = range(1, 11)
+        else:
+            i_range = range(1, 2)
+
+        # we want to test only that we are getting floating points back
+        # with some degree of the original value maintained, that it is not
+        # being truncated to an integer.  there's too much variation in how
+        # drivers return floats, which should not be relied upon to be
+        # exact, for us to just compare as is (works for PG drivers but not
+        # others) so we use rounding here.  There's precedent for this
+        # in suite/test_types.py::NumericTest as well
+
+        if do_rounding:
+            eq_(
+                {(id_, round(val_, 5)) for id_, val_ in result},
+                {(id_, round(value, 5)) for id_ in i_range},
+            )
+
+            eq_(
+                {
+                    round(val_, 5)
+                    for val_ in connection.scalars(select(t.c.value))
+                },
+                {round(value, 5)},
+            )
+        else:
+            eq_(
+                set(result),
+                {(id_, value) for id_ in i_range},
+            )
+
+            eq_(
+                set(connection.scalars(select(t.c.value))),
+                {value},
+            )
+
+    @testing.combinations(
+        (
+            "non_native_uuid",
+            Uuid(native_uuid=False),
+            uuid.uuid4(),
+        ),
+        (
+            "non_native_uuid_str",
+            Uuid(as_uuid=False, native_uuid=False),
+            str(uuid.uuid4()),
+        ),
+        (
+            "generic_native_uuid",
+            Uuid(native_uuid=True),
+            uuid.uuid4(),
+            testing.requires.uuid_data_type,
+        ),
+        (
+            "generic_native_uuid_str",
+            Uuid(as_uuid=False, native_uuid=True),
+            str(uuid.uuid4()),
+            testing.requires.uuid_data_type,
+        ),
+        ("UUID", UUID(), uuid.uuid4(), testing.requires.uuid_data_type),
+        (
+            "LargeBinary1",
+            LargeBinary(),
+            b"this is binary",
+        ),
+        ("LargeBinary2", LargeBinary(), b"7\xe7\x9f"),
+        argnames="type_,value",
+        id_="iaa",
+    )
+    @testing.variation("sort_by_parameter_order", [True, False])
+    @testing.variation("multiple_rows", [True, False])
+    @testing.requires.insert_returning
+    def test_imv_returning_datatypes(
+        self,
+        connection,
+        metadata,
+        sort_by_parameter_order,
+        type_,
+        value,
+        multiple_rows,
+    ):
+        """test #9739, #9808 (similar to #9701).
+
+        this tests insertmanyvalues in conjunction with various datatypes.
+
+        These tests are particularly for the asyncpg driver which needs
+        most types to be explicitly cast for the new IMV format
+
+        """
+        t = Table(
+            "d_t",
+            metadata,
+            Column("id", Integer, Identity(), primary_key=True),
+            Column("value", type_),
+        )
+
+        t.create(connection)
+
+        result = connection.execute(
+            t.insert().returning(
+                t.c.id,
+                t.c.value,
+                sort_by_parameter_order=bool(sort_by_parameter_order),
+            ),
+            (
+                [{"value": value} for i in range(10)]
+                if multiple_rows
+                else {"value": value}
+            ),
+        )
+
+        if multiple_rows:
+            i_range = range(1, 11)
+        else:
+            i_range = range(1, 2)
+
+        eq_(
+            set(result),
+            {(id_, value) for id_ in i_range},
+        )
+
+        eq_(
+            set(connection.scalars(select(t.c.value))),
+            {value},
+        )
+
+
+__all__ = ("LastrowidTest", "InsertBehaviorTest", "ReturningTest")

venv/Lib/site-packages/sqlalchemy/testing/suite/test_results.py

@@ -0,0 +1,504 @@
+# testing/suite/test_results.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+import datetime
+import re
+
+from .. import engines
+from .. import fixtures
+from ..assertions import eq_
+from ..config import requirements
+from ..schema import Column
+from ..schema import Table
+from ... import DateTime
+from ... import func
+from ... import Integer
+from ... import select
+from ... import sql
+from ... import String
+from ... import testing
+from ... import text
+
+
+class RowFetchTest(fixtures.TablesTest):
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "plain_pk",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("data", String(50)),
+        )
+        Table(
+            "has_dates",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("today", DateTime),
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        connection.execute(
+            cls.tables.plain_pk.insert(),
+            [
+                {"id": 1, "data": "d1"},
+                {"id": 2, "data": "d2"},
+                {"id": 3, "data": "d3"},
+            ],
+        )
+
+        connection.execute(
+            cls.tables.has_dates.insert(),
+            [{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}],
+        )
+
+    def test_via_attr(self, connection):
+        row = connection.execute(
+            self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
+        ).first()
+
+        eq_(row.id, 1)
+        eq_(row.data, "d1")
+
+    def test_via_string(self, connection):
+        row = connection.execute(
+            self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
+        ).first()
+
+        eq_(row._mapping["id"], 1)
+        eq_(row._mapping["data"], "d1")
+
+    def test_via_int(self, connection):
+        row = connection.execute(
+            self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
+        ).first()
+
+        eq_(row[0], 1)
+        eq_(row[1], "d1")
+
+    def test_via_col_object(self, connection):
+        row = connection.execute(
+            self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
+        ).first()
+
+        eq_(row._mapping[self.tables.plain_pk.c.id], 1)
+        eq_(row._mapping[self.tables.plain_pk.c.data], "d1")
+
+    @requirements.duplicate_names_in_cursor_description
+    def test_row_with_dupe_names(self, connection):
+        result = connection.execute(
+            select(
+                self.tables.plain_pk.c.data,
+                self.tables.plain_pk.c.data.label("data"),
+            ).order_by(self.tables.plain_pk.c.id)
+        )
+        row = result.first()
+        eq_(result.keys(), ["data", "data"])
+        eq_(row, ("d1", "d1"))
+
+    def test_row_w_scalar_select(self, connection):
+        """test that a scalar select as a column is returned as such
+        and that type conversion works OK.
+
+        (this is half a SQLAlchemy Core test and half to catch database
+        backends that may have unusual behavior with scalar selects.)
+
+        """
+        datetable = self.tables.has_dates
+        s = select(datetable.alias("x").c.today).scalar_subquery()
+        s2 = select(datetable.c.id, s.label("somelabel"))
+        row = connection.execute(s2).first()
+
+        eq_(row.somelabel, datetime.datetime(2006, 5, 12, 12, 0, 0))
+
+
+class PercentSchemaNamesTest(fixtures.TablesTest):
+    """tests using percent signs, spaces in table and column names.
+
+    This didn't work for PostgreSQL / MySQL drivers for a long time
+    but is now supported.
+
+    """
+
+    __requires__ = ("percent_schema_names",)
+
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        cls.tables.percent_table = Table(
+            "percent%table",
+            metadata,
+            Column("percent%", Integer),
+            Column("spaces % more spaces", Integer),
+        )
+        cls.tables.lightweight_percent_table = sql.table(
+            "percent%table",
+            sql.column("percent%"),
+            sql.column("spaces % more spaces"),
+        )
+
+    def test_single_roundtrip(self, connection):
+        percent_table = self.tables.percent_table
+        for params in [
+            {"percent%": 5, "spaces % more spaces": 12},
+            {"percent%": 7, "spaces % more spaces": 11},
+            {"percent%": 9, "spaces % more spaces": 10},
+            {"percent%": 11, "spaces % more spaces": 9},
+        ]:
+            connection.execute(percent_table.insert(), params)
+        self._assert_table(connection)
+
+    def test_executemany_roundtrip(self, connection):
+        percent_table = self.tables.percent_table
+        connection.execute(
+            percent_table.insert(), {"percent%": 5, "spaces % more spaces": 12}
+        )
+        connection.execute(
+            percent_table.insert(),
+            [
+                {"percent%": 7, "spaces % more spaces": 11},
+                {"percent%": 9, "spaces % more spaces": 10},
+                {"percent%": 11, "spaces % more spaces": 9},
+            ],
+        )
+        self._assert_table(connection)
+
+    @requirements.insert_executemany_returning
+    def test_executemany_returning_roundtrip(self, connection):
+        percent_table = self.tables.percent_table
+        connection.execute(
+            percent_table.insert(), {"percent%": 5, "spaces % more spaces": 12}
+        )
+        result = connection.execute(
+            percent_table.insert().returning(
+                percent_table.c["percent%"],
+                percent_table.c["spaces % more spaces"],
+            ),
+            [
+                {"percent%": 7, "spaces % more spaces": 11},
+                {"percent%": 9, "spaces % more spaces": 10},
+                {"percent%": 11, "spaces % more spaces": 9},
+            ],
+        )
+        eq_(result.all(), [(7, 11), (9, 10), (11, 9)])
+        self._assert_table(connection)
+
+    def _assert_table(self, conn):
+        percent_table = self.tables.percent_table
+        lightweight_percent_table = self.tables.lightweight_percent_table
+
+        for table in (
+            percent_table,
+            percent_table.alias(),
+            lightweight_percent_table,
+            lightweight_percent_table.alias(),
+        ):
+            eq_(
+                list(
+                    conn.execute(table.select().order_by(table.c["percent%"]))
+                ),
+                [(5, 12), (7, 11), (9, 10), (11, 9)],
+            )
+
+            eq_(
+                list(
+                    conn.execute(
+                        table.select()
+                        .where(table.c["spaces % more spaces"].in_([9, 10]))
+                        .order_by(table.c["percent%"])
+                    )
+                ),
+                [(9, 10), (11, 9)],
+            )
+
+            row = conn.execute(
+                table.select().order_by(table.c["percent%"])
+            ).first()
+            eq_(row._mapping["percent%"], 5)
+            eq_(row._mapping["spaces % more spaces"], 12)
+
+            eq_(row._mapping[table.c["percent%"]], 5)
+            eq_(row._mapping[table.c["spaces % more spaces"]], 12)
+
+        conn.execute(
+            percent_table.update().values(
+                {percent_table.c["spaces % more spaces"]: 15}
+            )
+        )
+
+        eq_(
+            list(
+                conn.execute(
+                    percent_table.select().order_by(
+                        percent_table.c["percent%"]
+                    )
+                )
+            ),
+            [(5, 15), (7, 15), (9, 15), (11, 15)],
+        )
+
+
+class ServerSideCursorsTest(
+    fixtures.TestBase, testing.AssertsExecutionResults
+):
+    __requires__ = ("server_side_cursors",)
+
+    __backend__ = True
+
+    def _is_server_side(self, cursor):
+        # TODO: this is a huge issue as it prevents these tests from being
+        # usable by third party dialects.
+        if self.engine.dialect.driver == "psycopg2":
+            return bool(cursor.name)
+        elif self.engine.dialect.driver == "pymysql":
+            sscursor = __import__("pymysql.cursors").cursors.SSCursor
+            return isinstance(cursor, sscursor)
+        elif self.engine.dialect.driver in ("aiomysql", "asyncmy", "aioodbc"):
+            return cursor.server_side
+        elif self.engine.dialect.driver == "mysqldb":
+            sscursor = __import__("MySQLdb.cursors").cursors.SSCursor
+            return isinstance(cursor, sscursor)
+        elif self.engine.dialect.driver == "mariadbconnector":
+            return not cursor.buffered
+        elif self.engine.dialect.driver == "mysqlconnector":
+            return "buffered" not in type(cursor).__name__.lower()
+        elif self.engine.dialect.driver in ("asyncpg", "aiosqlite"):
+            return cursor.server_side
+        elif self.engine.dialect.driver == "pg8000":
+            return getattr(cursor, "server_side", False)
+        elif self.engine.dialect.driver == "psycopg":
+            return bool(getattr(cursor, "name", False))
+        elif self.engine.dialect.driver == "oracledb":
+            return getattr(cursor, "server_side", False)
+        else:
+            return False
+
+    def _fixture(self, server_side_cursors):
+        if server_side_cursors:
+            with testing.expect_deprecated(
+                "The create_engine.server_side_cursors parameter is "
+                "deprecated and will be removed in a future release.  "
+                "Please use the Connection.execution_options.stream_results "
+                "parameter."
+            ):
+                self.engine = engines.testing_engine(
+                    options={"server_side_cursors": server_side_cursors}
+                )
+        else:
+            self.engine = engines.testing_engine(
+                options={"server_side_cursors": server_side_cursors}
+            )
+        return self.engine
+
+    def stringify(self, str_):
+        return re.compile(r"SELECT (\d+)", re.I).sub(
+            lambda m: str(select(int(m.group(1))).compile(testing.db)), str_
+        )
+
+    @testing.combinations(
+        ("global_string", True, lambda stringify: stringify("select 1"), True),
+        (
+            "global_text",
+            True,
+            lambda stringify: text(stringify("select 1")),
+            True,
+        ),
+        ("global_expr", True, select(1), True),
+        (
+            "global_off_explicit",
+            False,
+            lambda stringify: text(stringify("select 1")),
+            False,
+        ),
+        (
+            "stmt_option",
+            False,
+            select(1).execution_options(stream_results=True),
+            True,
+        ),
+        (
+            "stmt_option_disabled",
+            True,
+            select(1).execution_options(stream_results=False),
+            False,
+        ),
+        ("for_update_expr", True, select(1).with_for_update(), True),
+        # TODO: need a real requirement for this, or dont use this test
+        (
+            "for_update_string",
+            True,
+            lambda stringify: stringify("SELECT 1 FOR UPDATE"),
+            True,
+            testing.skip_if(["sqlite", "mssql"]),
+        ),
+        (
+            "text_no_ss",
+            False,
+            lambda stringify: text(stringify("select 42")),
+            False,
+        ),
+        (
+            "text_ss_option",
+            False,
+            lambda stringify: text(stringify("select 42")).execution_options(
+                stream_results=True
+            ),
+            True,
+        ),
+        id_="iaaa",
+        argnames="engine_ss_arg, statement, cursor_ss_status",
+    )
+    def test_ss_cursor_status(
+        self, engine_ss_arg, statement, cursor_ss_status
+    ):
+        engine = self._fixture(engine_ss_arg)
+        with engine.begin() as conn:
+            if callable(statement):
+                statement = testing.resolve_lambda(
+                    statement, stringify=self.stringify
+                )
+
+            if isinstance(statement, str):
+                result = conn.exec_driver_sql(statement)
+            else:
+                result = conn.execute(statement)
+            eq_(self._is_server_side(result.cursor), cursor_ss_status)
+            result.close()
+
+    def test_conn_option(self):
+        engine = self._fixture(False)
+
+        with engine.connect() as conn:
+            # should be enabled for this one
+            result = conn.execution_options(
+                stream_results=True
+            ).exec_driver_sql(self.stringify("select 1"))
+            assert self._is_server_side(result.cursor)
+
+            # the connection has autobegun, which means at the end of the
+            # block, we will roll back, which on MySQL at least will fail
+            # with "Commands out of sync" if the result set
+            # is not closed, so we close it first.
+            #
+            # fun fact!  why did we not have this result.close() in this test
+            # before 2.0? don't we roll back in the connection pool
+            # unconditionally? yes!  and in fact if you run this test in 1.4
+            # with stdout shown, there is in fact "Exception during reset or
+            # similar" with "Commands out sync" emitted a warning!  2.0's
+            # architecture finds and fixes what was previously an expensive
+            # silent error condition.
+            result.close()
+
+    def test_stmt_enabled_conn_option_disabled(self):
+        engine = self._fixture(False)
+
+        s = select(1).execution_options(stream_results=True)
+
+        with engine.connect() as conn:
+            # not this one
+            result = conn.execution_options(stream_results=False).execute(s)
+            assert not self._is_server_side(result.cursor)
+
+    def test_aliases_and_ss(self):
+        engine = self._fixture(False)
+        s1 = (
+            select(sql.literal_column("1").label("x"))
+            .execution_options(stream_results=True)
+            .subquery()
+        )
+
+        # options don't propagate out when subquery is used as a FROM clause
+        with engine.begin() as conn:
+            result = conn.execute(s1.select())
+            assert not self._is_server_side(result.cursor)
+            result.close()
+
+        s2 = select(1).select_from(s1)
+        with engine.begin() as conn:
+            result = conn.execute(s2)
+            assert not self._is_server_side(result.cursor)
+            result.close()
+
+    def test_roundtrip_fetchall(self, metadata):
+        md = self.metadata
+
+        engine = self._fixture(True)
+        test_table = Table(
+            "test_table",
+            md,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+        )
+
+        with engine.begin() as connection:
+            test_table.create(connection, checkfirst=True)
+            connection.execute(test_table.insert(), dict(data="data1"))
+            connection.execute(test_table.insert(), dict(data="data2"))
+            eq_(
+                connection.execute(
+                    test_table.select().order_by(test_table.c.id)
+                ).fetchall(),
+                [(1, "data1"), (2, "data2")],
+            )
+            connection.execute(
+                test_table.update()
+                .where(test_table.c.id == 2)
+                .values(data=test_table.c.data + " updated")
+            )
+            eq_(
+                connection.execute(
+                    test_table.select().order_by(test_table.c.id)
+                ).fetchall(),
+                [(1, "data1"), (2, "data2 updated")],
+            )
+            connection.execute(test_table.delete())
+            eq_(
+                connection.scalar(
+                    select(func.count("*")).select_from(test_table)
+                ),
+                0,
+            )
+
+    def test_roundtrip_fetchmany(self, metadata):
+        md = self.metadata
+
+        engine = self._fixture(True)
+        test_table = Table(
+            "test_table",
+            md,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("data", String(50)),
+        )
+
+        with engine.begin() as connection:
+            test_table.create(connection, checkfirst=True)
+            connection.execute(
+                test_table.insert(),
+                [dict(data="data%d" % i) for i in range(1, 20)],
+            )
+
+            result = connection.execute(
+                test_table.select().order_by(test_table.c.id)
+            )
+
+            eq_(
+                result.fetchmany(5),
+                [(i, "data%d" % i) for i in range(1, 6)],
+            )
+            eq_(
+                result.fetchmany(10),
+                [(i, "data%d" % i) for i in range(6, 16)],
+            )
+            eq_(result.fetchall(), [(i, "data%d" % i) for i in range(16, 20)])

venv/Lib/site-packages/sqlalchemy/testing/suite/test_rowcount.py

@@ -0,0 +1,258 @@
+# testing/suite/test_rowcount.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+from sqlalchemy import bindparam
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import select
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy import testing
+from sqlalchemy import text
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+
+
+class RowCountTest(fixtures.TablesTest):
+    """test rowcount functionality"""
+
+    __requires__ = ("sane_rowcount",)
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "employees",
+            metadata,
+            Column(
+                "employee_id",
+                Integer,
+                autoincrement=False,
+                primary_key=True,
+            ),
+            Column("name", String(50)),
+            Column("department", String(1)),
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        cls.data = data = [
+            ("Angela", "A"),
+            ("Andrew", "A"),
+            ("Anand", "A"),
+            ("Bob", "B"),
+            ("Bobette", "B"),
+            ("Buffy", "B"),
+            ("Charlie", "C"),
+            ("Cynthia", "C"),
+            ("Chris", "C"),
+        ]
+
+        employees_table = cls.tables.employees
+        connection.execute(
+            employees_table.insert(),
+            [
+                {"employee_id": i, "name": n, "department": d}
+                for i, (n, d) in enumerate(data)
+            ],
+        )
+
+    def test_basic(self, connection):
+        employees_table = self.tables.employees
+        s = select(
+            employees_table.c.name, employees_table.c.department
+        ).order_by(employees_table.c.employee_id)
+        rows = connection.execute(s).fetchall()
+
+        eq_(rows, self.data)
+
+    @testing.variation("statement", ["update", "delete", "insert", "select"])
+    @testing.variation("close_first", [True, False])
+    def test_non_rowcount_scenarios_no_raise(
+        self, connection, statement, close_first
+    ):
+        employees_table = self.tables.employees
+
+        # WHERE matches 3, 3 rows changed
+        department = employees_table.c.department
+
+        if statement.update:
+            r = connection.execute(
+                employees_table.update().where(department == "C"),
+                {"department": "Z"},
+            )
+        elif statement.delete:
+            r = connection.execute(
+                employees_table.delete().where(department == "C"),
+                {"department": "Z"},
+            )
+        elif statement.insert:
+            r = connection.execute(
+                employees_table.insert(),
+                [
+                    {"employee_id": 25, "name": "none 1", "department": "X"},
+                    {"employee_id": 26, "name": "none 2", "department": "Z"},
+                    {"employee_id": 27, "name": "none 3", "department": "Z"},
+                ],
+            )
+        elif statement.select:
+            s = select(
+                employees_table.c.name, employees_table.c.department
+            ).where(employees_table.c.department == "C")
+            r = connection.execute(s)
+            r.all()
+        else:
+            statement.fail()
+
+        if close_first:
+            r.close()
+
+        assert r.rowcount in (-1, 3)
+
+    def test_update_rowcount1(self, connection):
+        employees_table = self.tables.employees
+
+        # WHERE matches 3, 3 rows changed
+        department = employees_table.c.department
+        r = connection.execute(
+            employees_table.update().where(department == "C"),
+            {"department": "Z"},
+        )
+        assert r.rowcount == 3
+
+    def test_update_rowcount2(self, connection):
+        employees_table = self.tables.employees
+
+        # WHERE matches 3, 0 rows changed
+        department = employees_table.c.department
+
+        r = connection.execute(
+            employees_table.update().where(department == "C"),
+            {"department": "C"},
+        )
+        eq_(r.rowcount, 3)
+
+    @testing.variation("implicit_returning", [True, False])
+    @testing.variation(
+        "dml",
+        [
+            ("update", testing.requires.update_returning),
+            ("delete", testing.requires.delete_returning),
+        ],
+    )
+    def test_update_delete_rowcount_return_defaults(
+        self, connection, implicit_returning, dml
+    ):
+        """note this test should succeed for all RETURNING backends
+        as of 2.0.  In
+        Idf28379f8705e403a3c6a937f6a798a042ef2540 we changed rowcount to use
+        len(rows) when we have implicit returning
+
+        """
+
+        if implicit_returning:
+            employees_table = self.tables.employees
+        else:
+            employees_table = Table(
+                "employees",
+                MetaData(),
+                Column(
+                    "employee_id",
+                    Integer,
+                    autoincrement=False,
+                    primary_key=True,
+                ),
+                Column("name", String(50)),
+                Column("department", String(1)),
+                implicit_returning=False,
+            )
+
+        department = employees_table.c.department
+
+        if dml.update:
+            stmt = (
+                employees_table.update()
+                .where(department == "C")
+                .values(name=employees_table.c.department + "Z")
+                .return_defaults()
+            )
+        elif dml.delete:
+            stmt = (
+                employees_table.delete()
+                .where(department == "C")
+                .return_defaults()
+            )
+        else:
+            dml.fail()
+
+        r = connection.execute(stmt)
+        eq_(r.rowcount, 3)
+
+    def test_raw_sql_rowcount(self, connection):
+        # test issue #3622, make sure eager rowcount is called for text
+        result = connection.exec_driver_sql(
+            "update employees set department='Z' where department='C'"
+        )
+        eq_(result.rowcount, 3)
+
+    def test_text_rowcount(self, connection):
+        # test issue #3622, make sure eager rowcount is called for text
+        result = connection.execute(
+            text("update employees set department='Z' where department='C'")
+        )
+        eq_(result.rowcount, 3)
+
+    def test_delete_rowcount(self, connection):
+        employees_table = self.tables.employees
+
+        # WHERE matches 3, 3 rows deleted
+        department = employees_table.c.department
+        r = connection.execute(
+            employees_table.delete().where(department == "C")
+        )
+        eq_(r.rowcount, 3)
+
+    @testing.requires.sane_multi_rowcount
+    def test_multi_update_rowcount(self, connection):
+        employees_table = self.tables.employees
+        stmt = (
+            employees_table.update()
+            .where(employees_table.c.name == bindparam("emp_name"))
+            .values(department="C")
+        )
+
+        r = connection.execute(
+            stmt,
+            [
+                {"emp_name": "Bob"},
+                {"emp_name": "Cynthia"},
+                {"emp_name": "nonexistent"},
+            ],
+        )
+
+        eq_(r.rowcount, 2)
+
+    @testing.requires.sane_multi_rowcount
+    def test_multi_delete_rowcount(self, connection):
+        employees_table = self.tables.employees
+
+        stmt = employees_table.delete().where(
+            employees_table.c.name == bindparam("emp_name")
+        )
+
+        r = connection.execute(
+            stmt,
+            [
+                {"emp_name": "Bob"},
+                {"emp_name": "Cynthia"},
+                {"emp_name": "nonexistent"},
+            ],
+        )
+
+        eq_(r.rowcount, 2)