Merged PR 5306: Cursor rowcount, description + Tests

#### AI description  (iteration 1)
#### PR Classification
New feature and tests

#### PR Summary
This pull request implements new features for the `Cursor` class and adds corresponding tests.
- Implemented `Cursor.rowcount` to track the number of affected rows in `mssql_python/cursor.py`.
- Added tests for `Cursor.rowcount` in `tests/test_cursor.py`.
- Updated `main.py` to include examples of using `Cursor.rowcount`.
- Modified `.coveragerc` to exclude test files from coverage.
<!-- GitOpsUserAgent=GitOps.Apps.Server.pullrequestcopilot -->

Related work items: #32727, #32728, #33076

Author: bewithgaurav

.coveragerc

@@ -1,6 +1,7 @@
 [run]
 omit =
     mssql_python/testing_ddbc_bindings.py
+    tests/*
 
 [report]
 # Add any report-specific settings here, if needed

mssql_python/constants.py

@@ -104,3 +104,4 @@ class ConstantsODBC(Enum):
     SQL_PARAM_OUTPUT = 2
     SQL_PARAM_INPUT_OUTPUT = 3
     SQL_C_WCHAR = -8
+    SQL_NULLABLE = 1

mssql_python/cursor.py

@@ -409,6 +409,62 @@ def _create_parameter_types_list(self, parameter, ParamInfo, parameters_list, i)
         paraminfo.decimalDigits = decimal_digits
         return paraminfo
 
+    def _initialize_description(self):
+        """
+        Initialize the description attribute using SQLDescribeCol.
+        """
+        col_metadata = []
+        ret = ddbc_bindings.DDBCSQLDescribeCol(self.hstmt.value, col_metadata)
+        check_error(odbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt.value, ret)
+
+        self.description = [
+            (
+                col["ColumnName"],
+                self._map_data_type(col["DataType"]),
+                None,
+                col["ColumnSize"],
+                col["ColumnSize"],
+                col["DecimalDigits"],
+                col["Nullable"] == odbc_sql_const.SQL_NULLABLE.value,
+            )
+            for col in col_metadata
+        ]
+
+    def _map_data_type(self, sql_type):
+        """
+        Map SQL data type to Python data type.
+        
+        Args:
+            sql_type: SQL data type.
+        
+        Returns:
+            Corresponding Python data type.
+        """
+        sql_to_python_type = {
+            odbc_sql_const.SQL_INTEGER.value: int,
+            odbc_sql_const.SQL_VARCHAR.value: str,
+            odbc_sql_const.SQL_WVARCHAR.value: str,
+            odbc_sql_const.SQL_CHAR.value: str,
+            odbc_sql_const.SQL_WCHAR.value: str,
+            odbc_sql_const.SQL_FLOAT.value: float,
+            odbc_sql_const.SQL_DOUBLE.value: float,
+            odbc_sql_const.SQL_DECIMAL.value: decimal.Decimal,
+            odbc_sql_const.SQL_NUMERIC.value: decimal.Decimal,
+            odbc_sql_const.SQL_DATE.value: datetime.date,
+            odbc_sql_const.SQL_TIMESTAMP.value: datetime.datetime,
+            odbc_sql_const.SQL_TIME.value: datetime.time,
+            odbc_sql_const.SQL_BIT.value: bool,
+            odbc_sql_const.SQL_TINYINT.value: int,
+            odbc_sql_const.SQL_SMALLINT.value: int,
+            odbc_sql_const.SQL_BIGINT.value: int,
+            odbc_sql_const.SQL_BINARY.value: bytes,
+            odbc_sql_const.SQL_VARBINARY.value: bytes,
+            odbc_sql_const.SQL_LONGVARBINARY.value: bytes,
+            odbc_sql_const.SQL_GUID.value: uuid.UUID,
+            # Add more mappings as needed
+        }
+        return sql_to_python_type.get(sql_type, str)
+
     def execute(self, operation: str, *parameters, use_prepare: bool = True, reset_cursor: bool = True):
         """
         Prepare and execute a database operation (query or command).
@@ -452,6 +508,12 @@ def execute(self, operation: str, *parameters, use_prepare: bool = True, reset_c
                                                self.is_stmt_prepared, use_prepare)
             check_error(odbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt.value, ret)
             self.last_executed_stmt = operation
+
+            # Update rowcount after execution
+            self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt.value)
+
+            # Initialize description after execution
+            self._initialize_description()
         except Exception as e:
             if ENABLE_LOGGING:
                 logging.error("An error occurred while executing query: %s", e)
@@ -475,6 +537,7 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
             self._reset_cursor()
 
             first_execution = True
+            total_rowcount = 0
             for parameters in seq_of_parameters:
                 # Execute the operation with the current set of parameters without 
                 # Converting the parameters to a list
@@ -491,6 +554,13 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
                     prepare_stmt = False
                 # Execute statement with one parameter set
                 self.execute(operation, parameters, use_prepare=prepare_stmt, reset_cursor=False)
+                if self.rowcount != -1:
+                    # Rowcount would get updated inside execute method, add it to the current rowcount
+                    total_rowcount += self.rowcount
+                else:
+                    total_rowcount = -1
+            # Update the rowcount after all executions
+            self.rowcount = total_rowcount
         except Exception as e:
             if ENABLE_LOGGING:
                 logging.error("An error occurred while executing multiple queries: %s", e)

mssql_python/msvcp140.dll

@@ -196,6 +196,12 @@ SQLFreeStmtFunc SQLFreeStmt_ptr = nullptr;
 // Diagnostic record function pointer
 SQLGetDiagRecFunc SQLGetDiagRec_ptr = nullptr;
 
+// Function pointer typedef for SQLRowCount
+typedef SQLRETURN (*SQLRowCountFunc)(SQLHSTMT, SQLLEN*);
+
+// Function pointer for SQLRowCount
+SQLRowCountFunc SQLRowCount_ptr = nullptr;
+
 namespace {
 // Helper to load the driver
 bool LoadDriver() {
@@ -247,6 +253,9 @@ bool LoadDriver() {
     // Diagnostic record function Loading
     SQLGetDiagRec_ptr = (SQLGetDiagRecFunc)GetProcAddress(hModule, "SQLGetDiagRecW");
 
+    // Load SQLRowCount function
+    SQLRowCount_ptr = (SQLRowCountFunc)GetProcAddress(hModule, "SQLRowCount");
+
 #ifdef _DEBUG
     std::cout << "Driver loaded successfully." << std::endl;
 #endif
@@ -256,7 +265,8 @@ bool LoadDriver() {
            SQLBindParameter_ptr && SQLExecute_ptr && SQLFetch_ptr && SQLFetchScroll_ptr &&
            SQLGetData_ptr && SQLNumResultCols_ptr && SQLBindCol_ptr && SQLDescribeCol_ptr &&
            SQLMoreResults_ptr && SQLColAttribute_ptr && SQLColAttribute_ptr && SQLEndTran_ptr &&
-           SQLFreeHandle_ptr && SQLDisconnect_ptr && SQLFreeStmt_ptr && SQLGetDiagRec_ptr;
+           SQLFreeHandle_ptr && SQLDisconnect_ptr && SQLFreeStmt_ptr && SQLGetDiagRec_ptr &&
+           SQLRowCount_ptr;
 }
 
 // TODO: Add more nuanced exception classes
@@ -1598,6 +1608,23 @@ SQLRETURN SQLDisconnect_wrap(intptr_t ConnectionHandle) {
     return SQLDisconnect_ptr(reinterpret_cast<SQLHDBC>(ConnectionHandle));
 }
 
+// Wrap SQLRowCount
+SQLLEN SQLRowCount_wrap(intptr_t StatementHandle) {
+    if (!SQLRowCount_ptr && !LoadDriver()) {
+        return -1;
+    }
+
+    SQLLEN rowCount;
+    SQLRETURN ret = SQLRowCount_ptr(reinterpret_cast<SQLHSTMT>(StatementHandle), &rowCount);
+    if (!SQL_SUCCEEDED(ret)) {
+        std::cerr << "SQLRowCount failed with error code: " << ret << std::endl;
+        return -1;
+    }
+
+    std::cout << "SQLRowCount returned: " << rowCount << std::endl;  // Debug print
+    return rowCount;
+}
+
 // Bind the functions to the module
 PYBIND11_MODULE(ddbc_bindings, m) {
     m.doc() = "msodbcsql driver api bindings for Python";  // optional module docstring
@@ -1629,7 +1656,7 @@ PYBIND11_MODULE(ddbc_bindings, m) {
     m.def("DDBCSQLGetConnectionAttr", &SQLGetConnectionAttr_wrap,
           "Get an attribute that governs aspects of connections");
     m.def("DDBCSQLDriverConnect", &SQLDriverConnect_wrap,
-          "Connect to a data source with a connection string");
+            "Connect to a data source with a connection string");
     m.def("DDBCSQLExecDirect", &SQLExecDirect_wrap, "Execute a SQL query directly");
     m.def("DDBCSQLExecute", &SQLExecute_wrap, "Prepare and execute T-SQL statements");
     m.def("DDBCSQLFetch", &SQLFetch_wrap, "Fetch the next row from the result set");
@@ -1647,4 +1674,5 @@ PYBIND11_MODULE(ddbc_bindings, m) {
     m.def("DDBCSQLFreeHandle", &SQLFreeHandle_wrap, "Free a handle");
     m.def("DDBCSQLDisconnect", &SQLDisconnect_wrap, "Disconnect from a data source");
     m.def("DDBCSQLCheckError", &SQLCheckError_Wrap, "Check for driver errors");
+    m.def("DDBCSQLRowCount", &SQLRowCount_wrap, "Get the number of rows affected by the last statement");
 }

mssql_python/pybind/ddbc_bindings.cpp

@@ -68,6 +68,7 @@ def test_cursor(cursor):
 def test_insert_id_column(cursor, db_connection):
     """Test inserting data into the id column"""
     try:
+        drop_table_if_exists(cursor, "single_column")
         cursor.execute("CREATE TABLE single_column (id INTEGER PRIMARY KEY)")
         db_connection.commit()
         cursor.execute("INSERT INTO single_column (id) VALUES (?)", [1])
@@ -328,12 +329,64 @@ def test_parametrized_insert(cursor, db_connection, data):
     except Exception as e:
         pytest.fail(f"Parameterized data insertion failed: {e}")
 
-# def test_rowcount(cursor, db_connection):
-#     """Test rowcount"""
-#     cursor.execute("SELECT * FROM all_data_types")
-#     assert cursor.rowcount == -1, "Affected Rowcount should be -1"
-#     cursor.execute("UPDATE all_data_types SET wvarchar_column = 'updated' where id > 2")
-#     assert cursor.rowcount == 2, "Affected Rowcount should be 2"
+def test_rowcount(cursor, db_connection):
+    """Test rowcount after insert operations"""
+    try:
+        cursor.execute("CREATE TABLE test_rowcount (id INT IDENTITY(1,1) PRIMARY KEY, name NVARCHAR(100))")
+        db_connection.commit()
+
+        cursor.execute("INSERT INTO test_rowcount (name) VALUES ('JohnDoe1');")
+        assert cursor.rowcount == 1, "Rowcount should be 1 after first insert"
+
+        cursor.execute("INSERT INTO test_rowcount (name) VALUES ('JohnDoe2');")
+        assert cursor.rowcount == 1, "Rowcount should be 1 after second insert"
+
+        cursor.execute("INSERT INTO test_rowcount (name) VALUES ('JohnDoe3');")
+        assert cursor.rowcount == 1, "Rowcount should be 1 after third insert"
+
+        cursor.execute("""
+            INSERT INTO test_rowcount (name) 
+            VALUES 
+            ('JohnDoe4'), 
+            ('JohnDoe5'), 
+            ('JohnDoe6');
+        """)
+        assert cursor.rowcount == 3, "Rowcount should be 3 after inserting multiple rows"
+
+        cursor.execute("SELECT * FROM test_rowcount;")
+        assert cursor.rowcount == -1, "Rowcount should be -1 after a SELECT statement"
+
+        db_connection.commit()
+    except Exception as e:
+        pytest.fail(f"Rowcount test failed: {e}")
+    finally:
+        cursor.execute("DROP TABLE test_rowcount")
+        db_connection.commit()
+
+def test_rowcount_executemany(cursor, db_connection):
+    """Test rowcount after executemany operations"""
+    try:
+        cursor.execute("CREATE TABLE test_rowcount (id INT IDENTITY(1,1) PRIMARY KEY, name NVARCHAR(100))")
+        db_connection.commit()
+
+        data = [
+            ('JohnDoe1',),
+            ('JohnDoe2',),
+            ('JohnDoe3',)
+        ]
+
+        cursor.executemany("INSERT INTO test_rowcount (name) VALUES (?)", data)
+        assert cursor.rowcount == 3, "Rowcount should be 3 after executemany insert"
+
+        cursor.execute("SELECT * FROM test_rowcount;")
+        assert cursor.rowcount == -1, "Rowcount should be -1 after a SELECT statement"
+
+        db_connection.commit()
+    except Exception as e:
+        pytest.fail(f"Rowcount executemany test failed: {e}")
+    finally:
+        cursor.execute("DROP TABLE test_rowcount")
+        db_connection.commit()
 
 def test_fetchone(cursor):
     """Test fetching a single row"""
@@ -585,6 +638,18 @@ def test_drop_tables_for_join(cursor, db_connection):
     except Exception as e:
         pytest.fail(f"Failed to drop tables for join operations: {e}")
 
+def test_cursor_description(cursor):
+    """Test cursor description"""
+    cursor.execute("SELECT database_id, name FROM sys.databases;")
+    description = cursor.description
+    expected_description = [
+        ('database_id', int, None, 10, 10, 0, False),
+        ('name', str, None, 128, 128, 0, False)
+    ]
+    assert len(description) == len(expected_description), "Description length mismatch"
+    for desc, expected in zip(description, expected_description):
+        assert desc == expected, f"Description mismatch: {desc} != {expected}"
+
 def test_close(cursor):
     """Test closing the cursor"""
     cursor.close()