diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index 2a6684e4..63137e36 100644 --- a/mssql_python/cursor.py +++ b/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,38 @@ def execute( self.description = None self._reset_inputsizes() # Reset input sizes after execution + + # Reset lastrowid at the start of each execute + self._lastrowid = None + + 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 + log('debug', "Could not retrieve lastrowid: %s", e) + # Return self for method chaining return self @@ -1538,6 +1589,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) diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index 8faeea5a..257db171 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -11409,6 +11409,133 @@ 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() + + cursor.execute("INSERT INTO #test_lastrowid (name) VALUES ('test1'), ('test2'), ('test3')") + db_connection.commit() + + assert cursor.lastrowid == 3, "lastrowid should be 3 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: