diff --git a/mssql_python/row.py b/mssql_python/row.py index 8ebe0dab..86902c01 100644 --- a/mssql_python/row.py +++ b/mssql_python/row.py @@ -16,14 +16,21 @@ class Row: A row of data from a cursor fetch operation. Provides both tuple-like indexing and attribute access to column values. + Row supports dict-like access via keys(), values(), items(), and to_dict(). + However, iteration (for x in row) yields values, not keys — consistent with + pyodbc.Row and sqlite3.Row. Use row.keys() to iterate column names. + Column attribute access behavior depends on the global 'lowercase' setting: - When enabled: Case-insensitive attribute access - When disabled (default): Case-sensitive attribute access matching original column names Example: row = cursor.fetchone() - print(row[0]) # Access by index - print(row.column_name) # Access by column name (case sensitivity varies) + print(row[0]) # Access by index + print(row.column_name) # Access by column name + print(row.to_dict()) # Convert to dict + for value in row: # Iterates values, not keys + print(value) """ def __init__( @@ -73,6 +80,7 @@ def __init__( # Lowercase map is pre-built once per result set in the cursor and shared # across all rows. None when lowercase is off (the default) — zero cost. self._column_map_lower = column_map_lower + self._column_names = None # Lazy-computed on first access via _get_column_names() def _stringify_uuids(self, indices): """ @@ -209,6 +217,41 @@ def __getattr__(self, name: str) -> Any: raise AttributeError(f"Row has no attribute '{name}'") + def _get_column_names(self) -> tuple: + """Lazy-compute and cache deduplicated column names on first access.""" + if self._column_names is not None: + return self._column_names + + if self._cursor and hasattr(self._cursor, "description") and self._cursor.description: + column_names = tuple(desc[0] for desc in self._cursor.description) + elif self._column_map: + idx_to_name: dict = {} + for name, idx in self._column_map.items(): + if idx not in idx_to_name: + idx_to_name[idx] = name + column_names = tuple(idx_to_name[i] for i in sorted(idx_to_name)) + else: + column_names = () + + self._column_names = column_names + return column_names + + def keys(self) -> tuple: + """Return column names, like dict.keys().""" + return self._get_column_names() + + def values(self) -> tuple: + """Return column values as a tuple, like dict.values().""" + return tuple(self._values) + + def items(self) -> list: + """Return (column_name, value) pairs, like dict.items().""" + return list(zip(self._get_column_names(), self._values)) + + def to_dict(self) -> dict: + """Return the row as a plain dict mapping column names to values.""" + return dict(zip(self._get_column_names(), self._values)) + def __eq__(self, other: Any) -> bool: """ Support comparison with lists for test compatibility. diff --git a/tests/test_001_globals.py b/tests/test_001_globals.py index 08d31b5a..9e44e011 100644 --- a/tests/test_001_globals.py +++ b/tests/test_001_globals.py @@ -996,66 +996,3 @@ def test_stringify_uuids_with_tuple_values(): assert row[2] == "hello" # Internal storage should now be a list (converted from tuple) assert isinstance(row._values, list) - - -def test_row_string_key_indexing(): - """Test Row supports string-key indexing via __getitem__ (row['col']).""" - from mssql_python.row import Row - - row = Row( - [1, "foo", 3.14], - {"ProductID": 0, "Name": 1, "Price": 2}, - cursor=None, - ) - - # String-key access - assert row["ProductID"] == 1 - assert row["Name"] == "foo" - assert row["Price"] == 3.14 - - # Integer index access still works - assert row[0] == 1 - assert row[1] == "foo" - assert row[2] == 3.14 - - # Slice access still works - assert row[0:2] == [1, "foo"] - - # Missing key raises KeyError - with pytest.raises(KeyError): - row["nonexistent"] - - # Unsupported index types raise TypeError - with pytest.raises(TypeError): - row[3.5] - with pytest.raises(TypeError): - row[None] - - -def test_row_string_key_case_insensitive_with_lowercase(): - """Test Row string-key indexing is case-insensitive when column_map_lower is provided.""" - from mssql_python.row import Row - - column_map = {"productid": 0, "name": 1} - column_map_lower = {k.lower(): v for k, v in column_map.items()} - - row = Row( - [1, "bar"], - column_map, - cursor=None, - column_map_lower=column_map_lower, - ) - - # Exact match via __getitem__ - assert row["productid"] == 1 - # Exact match via __getattr__ - assert row.productid == 1 - # Case-insensitive match via __getitem__ - assert row["ProductID"] == 1 - assert row["NAME"] == "bar" - # Case-insensitive match via __getattr__ (attribute access) - assert row.ProductID == 1 - assert row.NAME == "bar" - # Non-existent attribute raises AttributeError - with pytest.raises(AttributeError): - row.nonexistent diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index 9b36c210..79670502 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -3081,6 +3081,226 @@ def test_row_string_key_indexing(cursor, db_connection): pass +def test_row_to_dict(cursor, db_connection): + """Test Row.to_dict() returns a plain dict from a real cursor row.""" + try: + cursor.execute( + "CREATE TABLE #pytest_row_todict (id INT PRIMARY KEY, name VARCHAR(50), price FLOAT)" + ) + db_connection.commit() + + cursor.execute("INSERT INTO #pytest_row_todict VALUES (1, 'Widget', 9.99)") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_todict") + row = cursor.fetchone() + + d = row.to_dict() + assert isinstance(d, dict) + assert d["id"] == 1 + assert d["name"] == "Widget" + assert d["price"] == 9.99 + assert len(d) == len(row) + + except Exception as e: + pytest.fail(f"Row to_dict test failed: {e}") + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_todict") + db_connection.commit() + except Exception: + pass + + +def test_row_keys_values_items(cursor, db_connection): + """Test Row.keys(), values(), items() from a real cursor row.""" + try: + cursor.execute("CREATE TABLE #pytest_row_kvi (id INT PRIMARY KEY, name VARCHAR(50))") + db_connection.commit() + + cursor.execute("INSERT INTO #pytest_row_kvi VALUES (42, 'Alice')") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_kvi") + row = cursor.fetchone() + + # keys() returns column names matching description + keys = list(row.keys()) + assert len(keys) == 2 + assert keys == [desc[0] for desc in cursor.description] + + # values() matches positional access + vals = list(row.values()) + assert vals == [row[0], row[1]] + + # items() returns (name, value) pairs + items = row.items() + assert len(items) == 2 + assert items[0] == (keys[0], row[0]) + assert items[1] == (keys[1], row[1]) + + # items() is reusable (not one-shot iterator) + assert list(items) == list(items) + + # len consistency + assert len(keys) == len(row) + assert len(vals) == len(row) + assert len(items) == len(row) + + except Exception as e: + pytest.fail(f"Row keys/values/items test failed: {e}") + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_kvi") + db_connection.commit() + except Exception: + pass + + +def test_row_dict_no_duplicate_keys(cursor, db_connection): + """Test that dict-like methods don't produce duplicate keys from cursor rows. + + The cursor may inject lowercase aliases into _column_map, but keys(), + items(), and to_dict() must return exactly N entries with original casing. + """ + try: + cursor.execute("CREATE TABLE #pytest_row_nodup (ProductID INT, MixedCase VARCHAR(20))") + db_connection.commit() + + cursor.execute("INSERT INTO #pytest_row_nodup VALUES (1, 'test')") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_nodup") + row = cursor.fetchone() + + keys = list(row.keys()) + items = row.items() + d = row.to_dict() + + # Exactly 2 columns, no duplicates + assert len(keys) == 2 + assert len(items) == 2 + assert len(d) == 2 + assert len(keys) == len(row) + + except Exception as e: + pytest.fail(f"Row dict no-duplicate-keys test failed: {e}") + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_nodup") + db_connection.commit() + except Exception: + pass + + +def test_row_getitem_type_guard(cursor, db_connection): + """Test Row.__getitem__ raises TypeError for unsupported index types.""" + try: + cursor.execute("CREATE TABLE #pytest_row_typeguard (id INT)") + db_connection.commit() + cursor.execute("INSERT INTO #pytest_row_typeguard VALUES (1)") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_typeguard") + row = cursor.fetchone() + + with pytest.raises(TypeError): + row[3.5] + with pytest.raises(TypeError): + row[None] + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_typeguard") + db_connection.commit() + except Exception: + pass + + +def test_row_case_sensitive_access(cursor, db_connection): + """Test exact-case access via cursor rows. + + Normal SELECT uses _cached_column_map (original casing only, no lowercase + aliases). Case-insensitive access only works when the global lowercase + setting is enabled or when rows come from metadata methods. + """ + try: + cursor.execute("CREATE TABLE #pytest_row_ci (ProductID INT, Name VARCHAR(50))") + db_connection.commit() + cursor.execute("INSERT INTO #pytest_row_ci VALUES (1, 'bar')") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_ci") + row = cursor.fetchone() + + # Original casing works via all access methods + assert row["ProductID"] == 1 + assert row.ProductID == 1 + + assert row["Name"] == "bar" + assert row.Name == "bar" + + # Non-existent + with pytest.raises(KeyError): + row["nonexistent"] + with pytest.raises(AttributeError): + row.nonexistent + + except Exception as e: + pytest.fail(f"Row case-insensitive access test failed: {e}") + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_ci") + db_connection.commit() + except Exception: + pass + + +def test_row_none_column_map(): + """Test Row edge case with column_map=None (covers _column_names=() branch).""" + from mssql_python.row import Row + + row = Row([], None, cursor=None) + assert list(row.keys()) == [] + assert list(row.values()) == [] + assert list(row.items()) == [] + assert row.to_dict() == {} + + +def test_row_dict_dedup_fallback(): + """Test dict-like methods deduplicate _column_map when cursor is None.""" + from mssql_python.row import Row + + column_map = {"ProductID": 0, "Name": 1} + row = Row([1, "foo"], column_map, cursor=None) + + assert list(row.keys()) == ["ProductID", "Name"] + assert row.to_dict() == {"ProductID": 1, "Name": "foo"} + + +def test_row_items_is_reusable(cursor, db_connection): + """Test items() returns a reusable list, not a one-shot iterator.""" + try: + cursor.execute("CREATE TABLE #pytest_row_reuse (id INT, name VARCHAR(50))") + db_connection.commit() + cursor.execute("INSERT INTO #pytest_row_reuse VALUES (1, 'foo')") + db_connection.commit() + + cursor.execute("SELECT * FROM #pytest_row_reuse") + row = cursor.fetchone() + + items = row.items() + first = list(items) + second = list(items) + assert first == second + assert len(first) > 0 + finally: + try: + cursor.execute("DROP TABLE IF EXISTS #pytest_row_reuse") + db_connection.commit() + except Exception: + pass + + def test_row_comparison_with_list(cursor, db_connection): """Test comparing Row objects with lists (__eq__ method)""" try: