FIX: Extract SqlTypeCode into own module, eliminate circular import workaround

Author: dlevy-msft-sql

CHANGELOG.md

@@ -7,22 +7,31 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ## [Unreleased]
 
 ### Added
+
 - New feature: Support for macOS and Linux.
 - Documentation: Added API documentation in the Wiki.
+- Added `SqlTypeCode` class for dual-compatible type codes in `cursor.description`.
 
 ### Changed
+
 - Improved error handling in the connection module.
+- Enhanced `cursor.description[i][1]` to return `SqlTypeCode` objects that compare equal to both SQL type integers and Python types, improving backwards compatibility while aligning with DB-API 2.0. Note that `SqlTypeCode` instances are intentionally unhashable; code that previously used `cursor.description[i][1]` as a dict or set key should use `int(type_code)` or `type_code.type_code` instead.
 
 ### Fixed
+
 - Bug fix: Resolved issue with connection timeout.
+- Fixed `cursor.description` type handling for better DB-API 2.0 compliance (Issue #352).
 
 ## [1.0.0-alpha] - 2025-02-24
 
 ### Added
+
 - Initial release of the mssql-python driver for SQL Server.
 
 ### Changed
+
 - N/A
 
 ### Fixed
-- N/A
+
+- N/A

mssql_python/__init__.py

@@ -60,6 +60,7 @@
 
 # Cursor Objects
 from .cursor import Cursor
+from .type_code import SqlTypeCode
 
 # Logging Configuration (Simplified single-level DEBUG system)
 from .logging import logger, setup_logging, driver_logger

mssql_python/connection.py

@@ -45,6 +45,8 @@
 from mssql_python.connection_string_builder import _ConnectionStringBuilder
 from mssql_python.constants import _RESERVED_PARAMETERS
 
+from mssql_python.type_code import SqlTypeCode
+
 if TYPE_CHECKING:
     from mssql_python.row import Row
 
@@ -923,7 +925,9 @@ def cursor(self) -> Cursor:
         logger.debug("cursor: Cursor created successfully - total_cursors=%d", len(self._cursors))
         return cursor
 
-    def add_output_converter(self, sqltype: int, func: Callable[[Any], Any]) -> None:
+    def add_output_converter(
+        self, sqltype: Union[int, SqlTypeCode, type], func: Callable[[Any], Any]
+    ) -> None:
         """
         Register an output converter function that will be called whenever a value
         with the given SQL type is read from the database.
@@ -936,32 +940,39 @@ def add_output_converter(self, sqltype: int, func: Callable[[Any], Any]) -> None
         vulnerabilities. This API should never be exposed to untrusted or external input.
 
         Args:
-            sqltype (int): The integer SQL type value to convert, which can be one of the
-                          defined standard constants (e.g. SQL_VARCHAR) or a database-specific
-                          value (e.g. -151 for the SQL Server 2008 geometry data type).
+            sqltype (int, SqlTypeCode, or type): The SQL type value to convert.
+                Also accepts SqlTypeCode objects or Python types for backward compatibility.
             func (callable): The converter function which will be called with a single parameter,
                             the value, and should return the converted value. If the value is NULL
-                            then the parameter passed to the function will be None, otherwise it
-                            will be a bytes object.
+                            then the parameter passed to the function will be None. For string/binary
+                            columns, the value will be bytes (UTF-16LE encoded for strings). For other
+                            types (int, decimal.Decimal, datetime, etc.), the value will be the native
+                            Python object.
 
         Returns:
             None
         """
+        if isinstance(sqltype, SqlTypeCode):
+            sqltype = sqltype.type_code
         with self._converters_lock:
             self._output_converters[sqltype] = func
             # Pass to the underlying connection if native implementation supports it
-            if hasattr(self._conn, "add_output_converter"):
+            # Only forward int type codes to native layer; Python type keys are handled
+            # only in our Python-side dictionary
+            if isinstance(sqltype, int) and hasattr(self._conn, "add_output_converter"):
                 self._conn.add_output_converter(sqltype, func)
         logger.info(f"Added output converter for SQL type {sqltype}")
 
-    def get_output_converter(self, sqltype: Union[int, type]) -> Optional[Callable[[Any], Any]]:
+    def get_output_converter(
+        self, sqltype: Union[int, SqlTypeCode, type]
+    ) -> Optional[Callable[[Any], Any]]:
         """
         Get the output converter function for the specified SQL type.
 
         Thread-safe implementation that protects the converters dictionary with a lock.
 
         Args:
-            sqltype (int or type): The SQL type value or Python type to get the converter for
+            sqltype (int, SqlTypeCode, or type): The SQL type value to get the converter for.
 
         Returns:
             callable or None: The converter function or None if no converter is registered
@@ -970,27 +981,43 @@ def get_output_converter(self, sqltype: Union[int, type]) -> Optional[Callable[[
             ⚠️ The returned converter function will be executed on database values. Only use
             converters from trusted sources.
         """
+        original_sqltype = sqltype
+        if isinstance(sqltype, SqlTypeCode):
+            sqltype = sqltype.type_code
         with self._converters_lock:
-            return self._output_converters.get(sqltype)
+            result = self._output_converters.get(sqltype)
+            # Fallback: try python_type key for backward compatibility
+            if result is None and isinstance(original_sqltype, SqlTypeCode):
+                result = self._output_converters.get(original_sqltype.python_type)
+            return result
 
-    def remove_output_converter(self, sqltype: Union[int, type]) -> None:
+    def remove_output_converter(self, sqltype: Union[int, SqlTypeCode, type]) -> None:
         """
         Remove the output converter function for the specified SQL type.
 
         Thread-safe implementation that protects the converters dictionary with a lock.
 
         Args:
-            sqltype (int or type): The SQL type value to remove the converter for
+            sqltype (int, SqlTypeCode, or type): The SQL type value to remove the converter for.
 
         Returns:
             None
         """
+        python_type_key = None
+        if isinstance(sqltype, SqlTypeCode):
+            python_type_key = sqltype.python_type
+            sqltype = sqltype.type_code
         with self._converters_lock:
             if sqltype in self._output_converters:
                 del self._output_converters[sqltype]
                 # Pass to the underlying connection if native implementation supports it
-                if hasattr(self._conn, "remove_output_converter"):
+                # Only forward int type codes to native layer; Python type keys are handled
+                # only in our Python-side dictionary
+                if isinstance(sqltype, int) and hasattr(self._conn, "remove_output_converter"):
                     self._conn.remove_output_converter(sqltype)
+            # Symmetric with get_output_converter: also remove python_type key if present
+            if python_type_key is not None and python_type_key in self._output_converters:
+                del self._output_converters[python_type_key]
         logger.info(f"Removed output converter for SQL type {sqltype}")
 
     def clear_output_converters(self) -> None:

mssql_python/constants.py

@@ -114,7 +114,12 @@ class ConstantsDDBC(Enum):
     SQL_FETCH_ABSOLUTE = 5
     SQL_FETCH_RELATIVE = 6
     SQL_FETCH_BOOKMARK = 8
+    # NOTE: The following SQL Server-specific type constants MUST stay in sync with
+    # the corresponding values in mssql_python/pybind/ddbc_bindings.cpp
     SQL_DATETIMEOFFSET = -155
+    SQL_SS_TIME2 = -154  # SQL Server TIME(n) type
+    SQL_SS_UDT = -151  # SQL Server User-Defined Types (geometry, geography, hierarchyid)
+    SQL_SS_XML = -152  # SQL Server XML type
     SQL_C_SS_TIMESTAMPOFFSET = 0x4001
     SQL_SCOPE_CURROW = 0
     SQL_BEST_ROWID = 1

mssql_python/cursor.py

@@ -20,6 +20,7 @@
 from mssql_python.helpers import check_error
 from mssql_python.logging import logger
 from mssql_python import ddbc_bindings
+from mssql_python.type_code import SqlTypeCode
 from mssql_python.exceptions import (
     InterfaceError,
     NotSupportedError,
@@ -142,6 +143,9 @@ def __init__(self, connection: "Connection", timeout: int = 0) -> None:
         )
         self.messages = []  # Store diagnostic messages
 
+        # Store raw column metadata for converter lookups
+        self._column_metadata = None
+
     def _is_unicode_string(self, param: str) -> bool:
         """
         Check if a string contains non-ASCII characters.
@@ -724,6 +728,14 @@ def _reset_cursor(self) -> None:
             logger.debug("SQLFreeHandle succeeded")
 
         self._clear_rownumber()
+        self._column_metadata = None
+        self.description = None
+
+        # Clear any result-set-specific caches to avoid stale mappings
+        if hasattr(self, "_cached_column_map"):
+            self._cached_column_map = None
+        if hasattr(self, "_cached_converter_map"):
+            self._cached_converter_map = None
 
         # Reinitialize the statement handle
         self._initialize_cursor()
@@ -756,6 +768,7 @@ def close(self) -> None:
             self.hstmt = None
             logger.debug("SQLFreeHandle succeeded")
         self._clear_rownumber()
+        self._column_metadata = None  # Clear metadata to prevent memory leaks
         self.closed = True
 
     def _check_closed(self) -> None:
@@ -942,8 +955,12 @@ def _initialize_description(self, column_metadata: Optional[Any] = None) -> None
         """Initialize the description attribute from column metadata."""
         if not column_metadata:
             self.description = None
+            self._column_metadata = None  # Clear metadata too
             return
 
+        # Store raw metadata for converter map building
+        self._column_metadata = column_metadata
+
         description = []
         for _, col in enumerate(column_metadata):
             # Get column name - lowercase it if the lowercase flag is set
@@ -954,10 +971,13 @@ def _initialize_description(self, column_metadata: Optional[Any] = None) -> None
                 column_name = column_name.lower()
 
             # Add to description tuple (7 elements as per PEP-249)
+            # Use SqlTypeCode for backwards-compatible type_code that works with both
+            # `desc[1] == str` (pandas) and `desc[1] == -9` (DB-API 2.0)
+            sql_type = col["DataType"]
             description.append(
                 (
                     column_name,  # name
-                    self._map_data_type(col["DataType"]),  # type_code
+                    SqlTypeCode(sql_type),  # type_code - dual compatible
                     None,  # display_size
                     col["ColumnSize"],  # internal_size
                     col["ColumnSize"],  # precision - should match ColumnSize
@@ -975,24 +995,28 @@ def _build_converter_map(self):
         """
         if (
             not self.description
+            or not self._column_metadata
             or not hasattr(self.connection, "_output_converters")
             or not self.connection._output_converters
         ):
             return None
 
         converter_map = []
 
-        for desc in self.description:
-            if desc is None:
-                converter_map.append(None)
-                continue
-            sql_type = desc[1]
+        for col_meta in self._column_metadata:
+            # Use the raw SQL type code from metadata, not the mapped Python type
+            sql_type = col_meta["DataType"]
+            python_type = SqlTypeCode._get_python_type(sql_type)
             converter = self.connection.get_output_converter(sql_type)
-            # If no converter found for the SQL type, try the WVARCHAR converter as a fallback
+
+            # Fallback: If no converter found for SQL type code, try the mapped Python type.
+            # This provides backward compatibility for code that registered converters by Python type.
             if converter is None:
-                from mssql_python.constants import ConstantsDDBC
+                converter = self.connection.get_output_converter(python_type)
 
-                converter = self.connection.get_output_converter(ConstantsDDBC.SQL_WVARCHAR.value)
+            # Fallback: try SQL_WVARCHAR converter for str/bytes columns
+            if converter is None and python_type in (str, bytes):
+                converter = self.connection.get_output_converter(ddbc_sql_const.SQL_WVARCHAR.value)
 
             converter_map.append(converter)
 
@@ -1022,41 +1046,6 @@ def _get_column_and_converter_maps(self):
 
         return column_map, converter_map
 
-    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 = {
-            ddbc_sql_const.SQL_INTEGER.value: int,
-            ddbc_sql_const.SQL_VARCHAR.value: str,
-            ddbc_sql_const.SQL_WVARCHAR.value: str,
-            ddbc_sql_const.SQL_CHAR.value: str,
-            ddbc_sql_const.SQL_WCHAR.value: str,
-            ddbc_sql_const.SQL_FLOAT.value: float,
-            ddbc_sql_const.SQL_DOUBLE.value: float,
-            ddbc_sql_const.SQL_DECIMAL.value: decimal.Decimal,
-            ddbc_sql_const.SQL_NUMERIC.value: decimal.Decimal,
-            ddbc_sql_const.SQL_DATE.value: datetime.date,
-            ddbc_sql_const.SQL_TIMESTAMP.value: datetime.datetime,
-            ddbc_sql_const.SQL_TIME.value: datetime.time,
-            ddbc_sql_const.SQL_BIT.value: bool,
-            ddbc_sql_const.SQL_TINYINT.value: int,
-            ddbc_sql_const.SQL_SMALLINT.value: int,
-            ddbc_sql_const.SQL_BIGINT.value: int,
-            ddbc_sql_const.SQL_BINARY.value: bytes,
-            ddbc_sql_const.SQL_VARBINARY.value: bytes,
-            ddbc_sql_const.SQL_LONGVARBINARY.value: bytes,
-            ddbc_sql_const.SQL_GUID.value: uuid.UUID,
-            # Add more mappings as needed
-        }
-        return sql_to_python_type.get(sql_type, str)
-
     @property
     def rownumber(self) -> int:
         """
@@ -1369,6 +1358,7 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
         except Exception as e:  # pylint: disable=broad-exception-caught
             # If describe fails, it's likely there are no results (e.g., for INSERT)
             self.description = None
+            self._column_metadata = None
 
         # Reset rownumber for new result set (only for SELECT statements)
         if self.description:  # If we have column descriptions, it's likely a SELECT
@@ -1385,15 +1375,6 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
             self._cached_column_map = None
             self._cached_converter_map = None
 
-        # After successful execution, initialize description if there are results
-        column_metadata = []
-        try:
-            ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
-            self._initialize_description(column_metadata)
-        except Exception as e:
-            # If describe fails, it's likely there are no results (e.g., for INSERT)
-            self.description = None
-
         self._reset_inputsizes()  # Reset input sizes after execution
         # Return self for method chaining
         return self
@@ -2425,6 +2406,7 @@ def nextset(self) -> Union[bool, None]:
             logger.debug("nextset: No more result sets available")
             self._clear_rownumber()
             self.description = None
+            self._column_metadata = None
             return False
 
         self._reset_rownumber()
@@ -2444,6 +2426,7 @@ def nextset(self) -> Union[bool, None]:
         except Exception as e:  # pylint: disable=broad-exception-caught
             # If describe fails, there might be no results in this result set
             self.description = None
+            self._column_metadata = None
 
         logger.debug(
             "nextset: Moved to next result set - column_count=%d",
@@ -2788,12 +2771,7 @@ def rollback(self):
         self._connection.rollback()
 
     def __del__(self):
-        """
-        Destructor to ensure the cursor is closed when it is no longer needed.
-        This is a safety net to ensure resources are cleaned up
-        even if close() was not called explicitly.
-        If the cursor is already closed, it will not raise an exception during cleanup.
-        """
+        """Safety net to close cursor if close() was not called explicitly."""
         if "closed" not in self.__dict__ or not self.closed:
             try:
                 self.close()