Fixed something bugs

Author: Peng Ren

pymongosql/cursor.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 import logging
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, TypeVar
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, TypeVar
 
 from pymongo.cursor import Cursor as MongoCursor
 from pymongo.errors import PyMongoError
@@ -206,7 +206,7 @@ def flush(self) -> None:
         # For now, this is a no-op
         pass
 
-    def fetchone(self) -> Optional[Dict[str, Any]]:
+    def fetchone(self) -> Optional[Sequence[Any]]:
         """Fetch the next row from the result set"""
         self._check_closed()
 
@@ -215,7 +215,7 @@ def fetchone(self) -> Optional[Dict[str, Any]]:
 
         return self._result_set.fetchone()
 
-    def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
+    def fetchmany(self, size: Optional[int] = None) -> List[Sequence[Any]]:
         """Fetch multiple rows from the result set"""
         self._check_closed()
 
@@ -224,7 +224,7 @@ def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
 
         return self._result_set.fetchmany(size)
 
-    def fetchall(self) -> List[Dict[str, Any]]:
+    def fetchall(self) -> List[Sequence[Any]]:
         """Fetch all remaining rows from the result set"""
         self._check_closed()
 

pymongosql/result_set.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 import logging
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 from pymongo.cursor import Cursor as MongoCursor
 from pymongo.errors import PyMongoError
@@ -32,12 +32,12 @@ def __init__(
             # Extract cursor info from command result
             self._result_cursor = command_result.get("cursor", {})
             self._raw_results = self._result_cursor.get("firstBatch", [])
-            self._cached_results: List[Dict[str, Any]] = []
+            self._cached_results: List[Sequence[Any]] = []
         elif mongo_cursor is not None:
             self._mongo_cursor = mongo_cursor
             self._command_result = None
             self._raw_results = []
-            self._cached_results: List[Dict[str, Any]] = []
+            self._cached_results: List[Sequence[Any]] = []
         else:
             raise ProgrammingError("Either command_result or mongo_cursor must be provided")
 
@@ -46,11 +46,15 @@ def __init__(
         self._cache_exhausted = False
         self._total_fetched = 0
         self._description: Optional[List[Tuple[str, str, None, None, None, None, None]]] = None
+        self._column_names: Optional[List[str]] = None  # Track column order for sequences
         self._errors: List[Dict[str, str]] = []
 
-        # Apply projection mapping for command results now that execution_plan is set
+        # Process firstBatch immediately if available (after all attributes are set)
         if command_result is not None and self._raw_results:
-            self._cached_results = [self._process_document(doc) for doc in self._raw_results]
+            processed_batch = [self._process_document(doc) for doc in self._raw_results]
+            # Convert dictionaries to sequences for DB API 2.0 compliance
+            sequence_batch = [self._dict_to_sequence(doc) for doc in processed_batch]
+            self._cached_results.extend(sequence_batch)
 
         # Build description from projection
         self._build_description()
@@ -102,7 +106,9 @@ def _ensure_results_available(self, count: int = 1) -> None:
 
                     # Process results through projection mapping
                     processed_batch = [self._process_document(doc) for doc in batch]
-                    self._cached_results.extend(processed_batch)
+                    # Convert dictionaries to sequences for DB API 2.0 compliance
+                    sequence_batch = [self._dict_to_sequence(doc) for doc in processed_batch]
+                    self._cached_results.extend(sequence_batch)
                     self._total_fetched += len(batch)
 
                 except PyMongoError as e:
@@ -127,6 +133,15 @@ def _process_document(self, doc: Dict[str, Any]) -> Dict[str, Any]:
 
         return processed
 
+    def _dict_to_sequence(self, doc: Dict[str, Any]) -> Tuple[Any, ...]:
+        """Convert document dictionary to sequence according to column order"""
+        if self._column_names is None:
+            # First time - establish column order
+            self._column_names = list(doc.keys())
+
+        # Return values in consistent column order
+        return tuple(doc.get(col_name) for col_name in self._column_names)
+
     @property
     def errors(self) -> List[Dict[str, str]]:
         return self._errors.copy()
@@ -145,18 +160,17 @@ def description(
             # Try to fetch one result to build description dynamically
             try:
                 self._ensure_results_available(1)
-                if self._cached_results:
-                    # Build description from first result
-                    first_result = self._cached_results[0]
+                if self._column_names:
+                    # Build description from established column names
                     self._description = [
-                        (col_name, "VARCHAR", None, None, None, None, None) for col_name in first_result.keys()
+                        (col_name, "VARCHAR", None, None, None, None, None) for col_name in self._column_names
                     ]
             except Exception as e:
                 _logger.warning(f"Could not build dynamic description: {e}")
 
         return self._description
 
-    def fetchone(self) -> Optional[Dict[str, Any]]:
+    def fetchone(self) -> Optional[Sequence[Any]]:
         """Fetch the next row from the result set"""
         if self._is_closed:
             raise ProgrammingError("ResultSet is closed")
@@ -172,7 +186,7 @@ def fetchone(self) -> Optional[Dict[str, Any]]:
         self._rownumber = (self._rownumber or 0) + 1
         return result
 
-    def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
+    def fetchmany(self, size: Optional[int] = None) -> List[Sequence[Any]]:
         """Fetch up to 'size' rows from the result set"""
         if self._is_closed:
             raise ProgrammingError("ResultSet is closed")
@@ -191,7 +205,7 @@ def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
 
         return results
 
-    def fetchall(self) -> List[Dict[str, Any]]:
+    def fetchall(self) -> List[Sequence[Any]]:
         """Fetch all remaining rows from the result set"""
         if self._is_closed:
             raise ProgrammingError("ResultSet is closed")
@@ -221,7 +235,9 @@ def fetchall(self) -> List[Dict[str, Any]]:
                     if remaining_docs:
                         # Process results through projection mapping
                         processed_docs = [self._process_document(doc) for doc in remaining_docs]
-                        all_results.extend(processed_docs)
+                        # Convert dictionaries to sequences for DB API 2.0 compliance
+                        sequence_docs = [self._dict_to_sequence(doc) for doc in processed_docs]
+                        all_results.extend(sequence_docs)
                         self._total_fetched += len(remaining_docs)
 
                     self._cache_exhausted = True