From 5f25cb57df383691b5810f8f6ad046e3e43c090b Mon Sep 17 00:00:00 2001 From: Abhishek Date: Tue, 10 Nov 2026 18:36:19 -0600 Subject: [PATCH] [PyMySQL] Correct cursor fetch result containers --- stubs/PyMySQL/@tests/test_cases/check_cursors.py | 14 ++++++++++++++ stubs/PyMySQL/pymysql/cursors.pyi | 10 ++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) create mode 100644 stubs/PyMySQL/@tests/test_cases/check_cursors.py diff --git a/stubs/PyMySQL/@tests/test_cases/check_cursors.py b/stubs/PyMySQL/@tests/test_cases/check_cursors.py new file mode 100644 index 000000000000..5ea216c6f997 --- /dev/null +++ b/stubs/PyMySQL/@tests/test_cases/check_cursors.py @@ -0,0 +1,14 @@ +from typing import Any +from typing_extensions import assert_type + +from pymysql.cursors import Cursor, DictCursor, SSCursor, SSDictCursor + + +def check_fetch_results(cursor: Cursor, dictionary: DictCursor, streaming: SSCursor, stream_dictionary: SSDictCursor) -> None: + # Mixin and streaming overrides must preserve both the row type and container. + assert_type(cursor.fetchall(), tuple[tuple[Any, ...], ...] | list[tuple[Any, ...]]) + assert_type(dictionary.fetchall(), list[dict[str, Any]] | tuple[()]) + assert_type(dictionary.fetchmany(), list[dict[str, Any]] | tuple[()]) + assert_type(streaming.fetchmany(), list[tuple[Any, ...]] | tuple[()]) + assert_type(stream_dictionary.fetchmany(), list[dict[str, Any]] | tuple[()]) + assert_type(stream_dictionary.fetchall(), list[dict[str, Any]]) diff --git a/stubs/PyMySQL/pymysql/cursors.pyi b/stubs/PyMySQL/pymysql/cursors.pyi index 4cb666119cf7..deebee098daf 100644 --- a/stubs/PyMySQL/pymysql/cursors.pyi +++ b/stubs/PyMySQL/pymysql/cursors.pyi @@ -45,26 +45,28 @@ class Cursor: # Methods returning result tuples are below. def fetchone(self) -> tuple[Any, ...] | None: ... def fetchmany(self, size: int | None = None) -> tuple[tuple[Any, ...], ...]: ... - def fetchall(self) -> tuple[tuple[Any, ...], ...]: ... + def fetchall(self) -> tuple[tuple[Any, ...], ...] | list[tuple[Any, ...]]: ... def __iter__(self) -> Iterator[tuple[Any, ...]]: ... def __next__(self): ... class DictCursorMixin: dict_type: Any # TODO: add support if someone needs this def fetchone(self) -> dict[str, Any] | None: ... - def fetchmany(self, size: int | None = ...) -> tuple[dict[str, Any], ...]: ... - def fetchall(self) -> tuple[dict[str, Any], ...]: ... + def fetchmany(self, size: int | None = ...) -> list[dict[str, Any]] | tuple[()]: ... + def fetchall(self) -> list[dict[str, Any]] | tuple[()]: ... def __iter__(self) -> Iterator[dict[str, Any]]: ... class SSCursor(Cursor): def __del__(self) -> None: ... def read_next(self) -> tuple[Any, ...] | None: ... - def fetchall(self) -> list[tuple[Any, ...]]: ... # type: ignore[override] + def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]] | tuple[()]: ... # type: ignore[override] + def fetchall(self) -> list[tuple[Any, ...]]: ... def fetchall_unbuffered(self) -> Iterator[tuple[Any, ...]]: ... def scroll(self, value: int, mode: str = "relative") -> None: ... class DictCursor(DictCursorMixin, Cursor): ... # type: ignore[misc] # pyrefly: ignore [inconsistent-inheritance] class SSDictCursor(DictCursorMixin, SSCursor): # type: ignore[misc] # pyrefly: ignore [inconsistent-inheritance] + def fetchall(self) -> list[dict[str, Any]]: ... # type: ignore[override] def fetchall_unbuffered(self) -> Iterator[dict[str, Any]]: ... # type: ignore[override] def read_next(self) -> dict[str, Any] | None: ... # type: ignore[override]