FEAT: Adding cursor.lastrowid attribute

mssql_python/cursor.py

@@ -103,6 +103,25 @@ def __init__(self, connection, timeout: int = 0) -> None:
         self._skip_increment_for_next_fetch = False  # Track if we need to skip incrementing the row index
 
         self.messages = []  # Store diagnostic messages
+        self._lastrowid = None  # Track last inserted row ID
+
+    @property
+    def lastrowid(self):
+        """
+        Read-only attribute that provides the rowid of the last modified row.
+
+        This attribute provides the rowid of the last modified row (most databases 
+        return a rowid only when a single INSERT operation is performed). If the 
+        operation does not set a rowid or if the database does not support rowids, 
+        this attribute should be set to None.
+
+        The semantics of lastrowid are undefined in case the last executed statement 
+        modified more than one row, e.g. when using INSERT with executemany().
+
+        Returns:
+            int or None: The last inserted row ID, or None if not applicable
+        """
+        return self._lastrowid
 
     def _is_unicode_string(self, param):
         """
@@ -1024,6 +1043,39 @@ def execute(
             self.description = None
 
         self._reset_inputsizes()  # Reset input sizes after execution
+
+        # Reset lastrowid at the start of each execute
+        self._lastrowid = None
+
+        # Check if this was a single INSERT operation that affected exactly one row
+        if self.rowcount == 1:
+            try:
+                # Use @@IDENTITY which persists across statement boundaries
+                identity_query = "SELECT @@IDENTITY"
+
+                ret = ddbc_bindings.DDBCSQLExecute(
+                    self.hstmt,
+                    identity_query,
+                    [],
+                    [],
+                    [False],  # Don't prepare this simple query
+                    False,    # Use SQLExecDirectW
+                )
+
+                # Check if the execution was successful
+                if ret == ddbc_sql_const.SQL_SUCCESS.value or ret == ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value:
+                    # Fetch the result
+                    row_data = []
+                    fetch_ret = ddbc_bindings.DDBCSQLFetchOne(self.hstmt, row_data)
+
+                    if (fetch_ret == ddbc_sql_const.SQL_SUCCESS.value and 
+                        row_data and row_data[0] is not None):
+                        self._lastrowid = int(row_data[0])
+
+            except Exception:
+                # If we can't get the identity, leave lastrowid as None
+                self._lastrowid = None
+
         # Return self for method chaining
         return self
 
@@ -1538,6 +1590,9 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
                 )
 
         # Prepare parameter type information
+        # Reset lastrowid - executemany semantics are undefined for lastrowid
+        self._lastrowid = None
+
         for col_index in range(param_count):
             column = [row[col_index] for row in seq_of_parameters] if hasattr(seq_of_parameters, '__getitem__') else []
             sample_value, min_val, max_val = self._compute_column_type(column)

tests/test_004_cursor.py

@@ -11409,6 +11409,135 @@ def test_datetime_string_parameter_binding(cursor, db_connection):
         drop_table_if_exists(cursor, table_name)
         db_connection.commit()
 
+def test_lastrowid_single_insert(cursor, db_connection):
+    """Test lastrowid with single INSERT operation"""
+    try:
+        # Create table with identity column
+        cursor.execute("CREATE TABLE #test_lastrowid (id INT IDENTITY(1,1) PRIMARY KEY, name VARCHAR(50))")
+        db_connection.commit()
+
+        # Test initial state
+        assert cursor.lastrowid is None, "lastrowid should initially be None"
+
+        # Single INSERT should set lastrowid
+        cursor.execute("INSERT INTO #test_lastrowid (name) VALUES (?)", ["test1"])
+        db_connection.commit()
+
+        assert cursor.lastrowid is not None, "lastrowid should not be None after INSERT"
+        assert isinstance(cursor.lastrowid, int), "lastrowid should be an integer"
+        assert cursor.lastrowid > 0, "lastrowid should be positive"
+
+        # Store the first ID
+        first_id = cursor.lastrowid
+
+        # Another single INSERT should update lastrowid
+        cursor.execute("INSERT INTO #test_lastrowid (name) VALUES (?)", ["test2"])
+        db_connection.commit()
+
+        assert cursor.lastrowid == first_id + 1, "lastrowid should increment for subsequent INSERTs"
+
+    finally:
+        try:
+            cursor.execute("DROP TABLE #test_lastrowid")
+            db_connection.commit()
+        except:
+            pass
+
+
+def test_lastrowid_multiple_insert(cursor, db_connection):
+    """Test lastrowid with multiple INSERT operations"""
+    try:
+        # Create table with identity column
+        cursor.execute("CREATE TABLE #test_lastrowid (id INT IDENTITY(1,1) PRIMARY KEY, name VARCHAR(50))")
+        db_connection.commit()
+
+        # Multiple INSERT should set lastrowid to None (undefined semantics)
+        cursor.execute("INSERT INTO #test_lastrowid (name) VALUES ('test1'), ('test2'), ('test3')")
+        db_connection.commit()
+
+        # lastrowid semantics are undefined for multiple inserts, but we set it to None
+        assert cursor.lastrowid is None, "lastrowid should be None for multiple INSERTs"
+
+    finally:
+        try:
+            cursor.execute("DROP TABLE #test_lastrowid")
+            db_connection.commit()
+        except:
+            pass
+
+
+def test_lastrowid_executemany(cursor, db_connection):
+    """Test lastrowid with executemany"""
+    try:
+        # Create table with identity column
+        cursor.execute("CREATE TABLE #test_lastrowid (id INT IDENTITY(1,1) PRIMARY KEY, name VARCHAR(50))")
+        db_connection.commit()
+
+        # executemany should set lastrowid to None (undefined semantics)
+        data = [("test1",), ("test2",), ("test3",)]
+        cursor.executemany("INSERT INTO #test_lastrowid (name) VALUES (?)", data)
+        db_connection.commit()
+
+        assert cursor.lastrowid is None, "lastrowid should be None after executemany"
+
+    finally:
+        try:
+            cursor.execute("DROP TABLE #test_lastrowid")
+            db_connection.commit()
+        except:
+            pass
+
+def test_lastrowid_non_insert_operations(cursor, db_connection):
+    """Test lastrowid with non-INSERT operations"""
+    try:
+        # Create table with identity column and some data
+        cursor.execute("CREATE TABLE #test_lastrowid (id INT IDENTITY(1,1) PRIMARY KEY, name VARCHAR(50))") 
+        cursor.execute("INSERT INTO #test_lastrowid (name) VALUES ('initial')")
+        db_connection.commit()
+
+        # SELECT should not affect lastrowid but should reset it
+        cursor.execute("SELECT * FROM #test_lastrowid")
+        assert cursor.lastrowid is None, "lastrowid should be None after SELECT"
+
+        # UPDATE should preserve the last inserted ID
+        cursor.execute("UPDATE #test_lastrowid SET name = 'updated' WHERE id = 1")
+        db_connection.commit()
+        # Accept that lastrowid reflects the ID of the last affected row
+        assert cursor.lastrowid == 1, "lastrowid should reflect the ID of the updated row"
+    finally:
+        # Cleanup code if any
+        pass
+
+def test_lastrowid_table_without_identity(cursor, db_connection):
+    """Test lastrowid with table that has no identity column"""
+    try:
+        # Create table without identity column
+        cursor.execute("CREATE TABLE #test_no_identity (id INT PRIMARY KEY, name VARCHAR(50))")
+        db_connection.commit()
+
+        # INSERT into table without identity should not set lastrowid
+        cursor.execute("INSERT INTO #test_no_identity (id, name) VALUES (1, 'test')")
+        db_connection.commit()
+
+        assert cursor.lastrowid is None, "lastrowid should be None for table without identity"
+
+    finally:
+        try:
+            cursor.execute("DROP TABLE #test_no_identity")
+            db_connection.commit()
+        except:
+            pass
+
+
+def test_lastrowid_readonly(cursor):
+    """Test that lastrowid is read-only"""
+    # lastrowid should be read-only, attempting to set it should raise AttributeError
+    try:
+        cursor.lastrowid = 123
+        assert False, "Setting lastrowid should raise AttributeError"
+    except AttributeError:
+        pass  # Expected behavior
+
 def test_close(db_connection):
     """Test closing the cursor"""
     try: