Support returning rows as dictionaries (#1487)

Closes #171

Co-authored-by: Michael Kleehammer <michael@kleehammer.com>

Author: bkline

src/cursor.cpp

@@ -1264,6 +1264,34 @@ static PyObject* Cursor_fetch(Cursor* cur)
         apValues[i] = value;
     }
 
+    // Return a dict instead of a Row if so requested.
+    // See https://github.com/mkleehammer/pyodbc/issues/171,
+    if (cur->rows_as_dicts)
+    {
+        PyObject* dict = PyDict_New();
+        if (!dict)
+        {
+            FreeRowValues(field_count, apValues);
+            return 0;
+        }
+
+        PyObject* name;
+        PyObject* index;
+        Py_ssize_t pos = 0;
+        while (PyDict_Next(cur->map_name_to_index, &pos, &name, &index))
+        {
+            Py_ssize_t i = PyNumber_AsSsize_t(index, PyExc_IndexError);
+            if (PyDict_SetItem(dict, name, apValues[i]) == -1)
+            {
+                Py_DECREF(dict);
+                FreeRowValues(field_count, apValues);
+                return 0;
+            }
+        }
+        FreeRowValues(field_count, apValues);
+        return dict;
+    }
+
     return (PyObject*)Row_InternalNew(cur->description, cur->map_name_to_index, field_count, apValues);
 }
 
@@ -2481,6 +2509,8 @@ static char messages_doc[] =
     "This read-only attribute is a list of all the diagnostic messages in the\n" \
     "current result set.";
 
+static char rowsasdicts_doc[] = "If True, rows are returned as dicts instead of Row objects.";
+
 static PyMemberDef Cursor_members[] =
 {
     {"rowcount",    T_INT,       offsetof(Cursor, rowcount),        READONLY, rowcount_doc },
@@ -2489,6 +2519,7 @@ static PyMemberDef Cursor_members[] =
     {"connection",  T_OBJECT_EX, offsetof(Cursor, cnxn),            READONLY, connection_doc },
     {"fast_executemany",T_BOOL,  offsetof(Cursor, fastexecmany),    0,        fastexecmany_doc },
     {"messages",    T_OBJECT_EX, offsetof(Cursor, messages),        READONLY, messages_doc },
+    {"rows_as_dicts", T_BOOL,    offsetof(Cursor, rows_as_dicts),   0,        rowsasdicts_doc },
     { 0 }
 };
 
@@ -2792,6 +2823,7 @@ Cursor_New(Connection* cnxn)
         cur->rowcount          = -1;
         cur->map_name_to_index = 0;
         cur->fastexecmany      = 0;
+        cur->rows_as_dicts     = 0;
         cur->messages          = Py_None;
 
         Py_INCREF(cnxn);

src/cursor.h

@@ -135,6 +135,9 @@ struct Cursor
     // Result Information
     //
 
+    // If true, return rows as dicts instead of Row objects.
+    char rows_as_dicts;
+
     // An array of ColumnInfos, allocated via malloc.  This will be zero when closed or when there are no query
     // results.
     ColumnInfo* colinfos;

src/pyodbc.pyi

@@ -610,6 +610,15 @@ class Cursor:
     def fast_executemany(self, value: bool) -> None:
         ...
 
+    @property
+    def rows_as_dicts(self) -> bool:
+        """If True, rows are returned as dicts instead of Row objects."""
+        ...
+
+    @rows_as_dicts.setter
+    def rows_as_dicts(self, value: bool) -> None:
+        ...
+
     @property
     def hstmt(self) -> ctypes.c_void_p | None:
         """ODBC handle for the statement."""

tests/postgresql_test.py

@@ -645,6 +645,49 @@ def _test():
     assert count_after == count_before
 
 
+def test_rows_as_dicts(cursor: pyodbc.Cursor):
+    """Test enhancement for ticket #171"""
+
+    # Create and populate a test table.
+    cursor.execute("create table t1 (id int, name varchar(20))")
+    cursor.execute("insert into t1 values (42, 'Kathleen')")
+
+    # Verify the default behavior
+    assert cursor.rows_as_dicts is False
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, dict)
+    assert isinstance(row, pyodbc.Row)
+    assert isinstance(row[0], int)
+    assert isinstance(row[1], str)
+    assert len(row) == 2
+    with pytest.raises(TypeError, match="row indices must be integers"):
+        print(row["name"])
+
+    # Test the dict option
+    cursor.rows_as_dicts = True
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, pyodbc.Row)
+    assert isinstance(row, dict)
+    assert row == {"id": 42, "name": "Kathleen"}
+    assert isinstance(row["id"], int)
+    assert isinstance(row["name"], str)
+    assert len(row) == 2
+    with pytest.raises(KeyError):
+        print(row[1])
+
+    # Test aliasing
+    row = cursor.execute("select name as n1, name as n2 from t1").fetchone()
+    assert len(row) == 2
+    assert row == {"n1": "Kathleen", "n2": "Kathleen"}
+    with pytest.raises(KeyError):
+        print(row["name"])
+
+    # Test with a duplicate name
+    row = cursor.execute("select name, name from t1").fetchone()
+    assert len(row) == 1
+    assert row == {"name": "Kathleen"}
+
+
 def test_handles(cursor: pyodbc.Cursor):
     """Test the exposed native ODBC handles"""
 

tests/sqlite_test.py

@@ -756,3 +756,46 @@ def test_handles(cursor: pyodbc.Cursor):
     conn.close()
     assert not isinstance(conn.hdbc, ctypes.c_void_p)
     assert conn.hdbc is None
+
+
+def test_rows_as_dicts(cursor: pyodbc.Cursor):
+    """Test enhancement for ticket #171"""
+
+    # Create and populate a test table.
+    cursor.execute("create table t1 (id int, name varchar(20))")
+    cursor.execute("insert into t1 values (42, 'Kathleen')")
+
+    # Verify the default behavior
+    assert cursor.rows_as_dicts is False
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, dict)
+    assert isinstance(row, pyodbc.Row)
+    assert isinstance(row[0], int)
+    assert isinstance(row[1], str)
+    assert len(row) == 2
+    with pytest.raises(TypeError, match="row indices must be integers"):
+        print(row["name"])
+
+    # Test the dict option
+    cursor.rows_as_dicts = True
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, pyodbc.Row)
+    assert isinstance(row, dict)
+    assert row == {"id": 42, "name": "Kathleen"}
+    assert isinstance(row["id"], int)
+    assert isinstance(row["name"], str)
+    assert len(row) == 2
+    with pytest.raises(KeyError):
+        print(row[1])
+
+    # Test aliasing
+    row = cursor.execute("select name as n1, name as n2 from t1").fetchone()
+    assert len(row) == 2
+    assert row == {"n1": "Kathleen", "n2": "Kathleen"}
+    with pytest.raises(KeyError):
+        print(row["name"])
+
+    # Test with a duplicate name
+    row = cursor.execute("select name, name from t1").fetchone()
+    assert len(row) == 1
+    assert row == {"name": "Kathleen"}

tests/sqlserver_test.py

@@ -1803,6 +1803,49 @@ def test_sql_variant(cursor: pyodbc.Cursor):
         assert results[index] == expected_value
 
 
+def test_rows_as_dicts(cursor: pyodbc.Cursor):
+    """Test enhancement for ticket #171"""
+
+    # Create and populate a test table.
+    cursor.execute("create table t1 (id int, name varchar(20))")
+    cursor.execute("insert into t1 values (42, 'Kathleen')")
+
+    # Verify the default behavior
+    assert cursor.rows_as_dicts is False
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, dict)
+    assert isinstance(row, pyodbc.Row)
+    assert isinstance(row[0], int)
+    assert isinstance(row[1], str)
+    assert len(row) == 2
+    with pytest.raises(TypeError, match="row indices must be integers"):
+        print(row["name"])
+
+    # Test the dict option
+    cursor.rows_as_dicts = True
+    row = cursor.execute("select * from t1").fetchone()
+    assert not isinstance(row, pyodbc.Row)
+    assert isinstance(row, dict)
+    assert row == {"id": 42, "name": "Kathleen"}
+    assert isinstance(row["id"], int)
+    assert isinstance(row["name"], str)
+    assert len(row) == 2
+    with pytest.raises(KeyError):
+        print(row[1])
+
+    # Test aliasing
+    row = cursor.execute("select name as n1, name as n2 from t1").fetchone()
+    assert len(row) == 2
+    assert row == {"n1": "Kathleen", "n2": "Kathleen"}
+    with pytest.raises(KeyError):
+        print(row["name"])
+
+    # Test with a duplicate name
+    row = cursor.execute("select name, name from t1").fetchone()
+    assert len(row) == 1
+    assert row == {"name": "Kathleen"}
+
+
 def test_handles(cursor: pyodbc.Cursor):
     """Test the exposed native ODBC handles"""