FEAT: Adding cursor.skip(n)

mssql_python/cursor.py

@@ -1175,4 +1175,33 @@ def scroll(self, value: int, mode: str = 'relative') -> None:
         except Exception as e:
             if isinstance(e, (IndexError, NotSupportedError)):
                 raise
-            raise IndexError(f"Scroll operation failed: {e}") from e
+            raise IndexError(f"Scroll operation failed: {e}") from e
+
+    def skip(self, count: int) -> None:
+        """
+        Skip the next 'count' records in the query result set.
+
+        This is a convenience method that advances the cursor by 'count' 
+        positions without returning the skipped rows.
+
+        Args:
+            count: Number of records to skip. Must be non-negative.
+
+        Returns:
+            None
+
+        Raises:
+            ProgrammingError: If the cursor is closed or no result set is available.
+            NotSupportedError: If count is negative (backward scrolling not supported).
+            IndexError: If attempting to skip past the end of the result set.
+
+        Note:
+            For convenience, skip(0) is accepted and will do nothing.
+        """
+        self._check_closed()
+
+        if count == 0:  # Skip 0 is a no-op
+            return
+
+        # Use existing scroll method with relative mode
+        self.scroll(count, 'relative')

tests/test_004_cursor.py

@@ -4022,6 +4022,184 @@ def test_scroll_edge_cases_and_validation(cursor, db_connection):
     finally:
         _drop_if_exists_scroll(cursor, "#t_scroll_validation")
 
+def test_cursor_skip_basic_functionality(cursor, db_connection):
+    """Test basic skip functionality that advances cursor position"""
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip")
+        cursor.execute("CREATE TABLE #test_skip (id INTEGER)")
+        cursor.executemany("INSERT INTO #test_skip VALUES (?)", [(i,) for i in range(1, 11)])
+        db_connection.commit()
+
+        # Execute query
+        cursor.execute("SELECT id FROM #test_skip ORDER BY id")
+
+        # Skip 3 rows
+        cursor.skip(3)
+
+        # After skip(3), last-returned index is 2
+        assert cursor.rownumber == 2, "After skip(3), last-returned index should be 2"
+
+        # Verify correct position by fetching - should get id=4
+        row = cursor.fetchone()
+        assert row[0] == 4, "After skip(3), next row should be id=4"
+
+        # Skip another 2 rows
+        cursor.skip(2)
+
+        # Verify position again
+        row = cursor.fetchone()
+        assert row[0] == 7, "After skip(2) more, next row should be id=7"
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip")
+
+def test_cursor_skip_zero_is_noop(cursor, db_connection):
+    """Test that skip(0) is a no-op"""
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip_zero")
+        cursor.execute("CREATE TABLE #test_skip_zero (id INTEGER)")
+        cursor.executemany("INSERT INTO #test_skip_zero VALUES (?)", [(i,) for i in range(1, 6)])
+        db_connection.commit()
+
+        # Execute query
+        cursor.execute("SELECT id FROM #test_skip_zero ORDER BY id")
+
+        # Get initial position
+        initial_rownumber = cursor.rownumber
+
+        # Skip 0 rows (should be no-op)
+        cursor.skip(0)
+
+        # Verify position unchanged
+        assert cursor.rownumber == initial_rownumber, "skip(0) should not change position"
+        row = cursor.fetchone()
+        assert row[0] == 1, "After skip(0), first row should still be id=1"
+
+        # Skip some rows, then skip(0)
+        cursor.skip(2)
+        position_after_skip = cursor.rownumber
+        cursor.skip(0)
+
+        # Verify position unchanged after second skip(0)
+        assert cursor.rownumber == position_after_skip, "skip(0) should not change position"
+        row = cursor.fetchone()
+        assert row[0] == 4, "After skip(2) then skip(0), should fetch id=4"
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip_zero")
+
+def test_cursor_skip_empty_result_set(cursor, db_connection):
+    """Test skip behavior with empty result set"""
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip_empty")
+        cursor.execute("CREATE TABLE #test_skip_empty (id INTEGER)")
+        db_connection.commit()
+
+        # Execute query on empty table
+        cursor.execute("SELECT id FROM #test_skip_empty")
+
+        # Skip should raise IndexError on empty result set
+        with pytest.raises(IndexError):
+            cursor.skip(1)
+
+        # Verify row is still None
+        assert cursor.fetchone() is None, "Empty result should return None"
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip_empty")
+
+def test_cursor_skip_past_end(cursor, db_connection):
+    """Test skip past end of result set"""
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip_end")
+        cursor.execute("CREATE TABLE #test_skip_end (id INTEGER)")
+        cursor.executemany("INSERT INTO #test_skip_end VALUES (?)", [(i,) for i in range(1, 4)])
+        db_connection.commit()
+
+        # Execute query
+        cursor.execute("SELECT id FROM #test_skip_end ORDER BY id")
+
+        # Skip beyond available rows
+        with pytest.raises(IndexError):
+            cursor.skip(5)  # Only 3 rows available
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip_end")
+
+def test_cursor_skip_invalid_arguments(cursor, db_connection):
+    """Test skip with invalid arguments"""
+    from mssql_python.exceptions import ProgrammingError, NotSupportedError
+
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip_args")
+        cursor.execute("CREATE TABLE #test_skip_args (id INTEGER)")
+        cursor.execute("INSERT INTO #test_skip_args VALUES (1)")
+        db_connection.commit()
+
+        cursor.execute("SELECT id FROM #test_skip_args")
+
+        # Test with non-integer
+        with pytest.raises(ProgrammingError):
+            cursor.skip("one")
+
+        # Test with float
+        with pytest.raises(ProgrammingError):
+            cursor.skip(1.5)
+
+        # Test with negative value
+        with pytest.raises(NotSupportedError):
+            cursor.skip(-1)
+
+        # Verify cursor still works after these errors
+        row = cursor.fetchone()
+        assert row[0] == 1, "Cursor should still be usable after error handling"
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip_args")
+
+def test_cursor_skip_closed_cursor(db_connection):
+    """Test skip on closed cursor"""
+    cursor = db_connection.cursor()
+    cursor.close()
+
+    with pytest.raises(Exception) as exc_info:
+        cursor.skip(1)
+
+    assert "closed" in str(exc_info.value).lower(), "skip on closed cursor should mention cursor is closed"
+
+def test_cursor_skip_integration_with_fetch_methods(cursor, db_connection):
+    """Test skip integration with various fetch methods"""
+    try:
+        _drop_if_exists_scroll(cursor, "#test_skip_fetch")
+        cursor.execute("CREATE TABLE #test_skip_fetch (id INTEGER)")
+        cursor.executemany("INSERT INTO #test_skip_fetch VALUES (?)", [(i,) for i in range(1, 11)])
+        db_connection.commit()
+
+        # Test with fetchone
+        cursor.execute("SELECT id FROM #test_skip_fetch ORDER BY id")
+        cursor.fetchone()  # Fetch first row (id=1), rownumber=0
+        cursor.skip(2)     # Skip next 2 rows (id=2,3), rownumber=2
+        row = cursor.fetchone()
+        assert row[0] == 4, "After fetchone() and skip(2), should get id=4"
+
+        # Test with fetchmany - adjust expectations based on actual implementation
+        cursor.execute("SELECT id FROM #test_skip_fetch ORDER BY id")
+        rows = cursor.fetchmany(2)  # Fetch first 2 rows (id=1,2)
+        assert [r[0] for r in rows] == [1, 2], "Should fetch first 2 rows"
+        cursor.skip(3)  # Skip 3 positions from current position
+        rows = cursor.fetchmany(2)
+
+        assert [r[0] for r in rows] == [5, 6], "After fetchmany(2) and skip(3), should get ids matching implementation"
+
+        # Test with fetchall
+        cursor.execute("SELECT id FROM #test_skip_fetch ORDER BY id")
+        cursor.skip(5)  # Skip first 5 rows
+        rows = cursor.fetchall()  # Fetch all remaining
+        assert [r[0] for r in rows] == [6, 7, 8, 9, 10], "After skip(5), fetchall() should get id=6-10"
+
+    finally:
+        _drop_if_exists_scroll(cursor, "#test_skip_fetch")
+
 def test_close(db_connection):
     """Test closing the cursor"""
     try: