"""

Copyright (c) Microsoft Corporation.

Licensed under the MIT license.

This module contains the Cursor class, which represents a database cursor.

Resource Management:

- Cursors are tracked by their parent connection.

- Closing the connection will automatically close all open cursors.

- Do not use a cursor after it is closed, or after its parent connection is closed.

- Use close() to release resources held by the cursor as soon as it is no longer needed.

"""

# pylint: disable=too-many-lines # Large file due to comprehensive DB-API 2.0 implementation

import decimal

import logging

import uuid

import datetime

import warnings

from typing import List, Mapping, Union, Any, Optional, Tuple, Sequence, TYPE_CHECKING, Iterable

from mssql_python.constants import ConstantsDDBC as ddbc_sql_const, SQLTypes

from mssql_python.helpers import check_error, connstr_to_pycore_params

from mssql_python.logging import logger

from mssql_python import ddbc_bindings

from mssql_python.exceptions import (

InterfaceError,

NotSupportedError,

ProgrammingError,

OperationalError,

DatabaseError,

)

from mssql_python.row import Row

from mssql_python import get_settings

from mssql_python.parameter_helper import (

detect_and_convert_parameters,

parse_pyformat_params,

convert_pyformat_to_qmark,

)

if TYPE_CHECKING:

import pyarrow # type: ignore

from mssql_python.connection import Connection

else:

pyarrow = None

# Constants for string handling

MAX_INLINE_CHAR: int = (

4000 # NVARCHAR/VARCHAR inline limit; this triggers NVARCHAR(MAX)/VARCHAR(MAX) + DAE

)

SMALLMONEY_MIN: decimal.Decimal = decimal.Decimal("-214748.3648")

SMALLMONEY_MAX: decimal.Decimal = decimal.Decimal("214748.3647")

MONEY_MIN: decimal.Decimal = decimal.Decimal("-922337203685477.5808")

MONEY_MAX: decimal.Decimal = decimal.Decimal("922337203685477.5807")

# SQL BIGINT is a signed 64-bit integer. Ints outside this range have no BIGINT

# encoding and must be rejected at detect time on both paths (see _map_sql_type).

BIGINT_MIN: int = -(2**63)

BIGINT_MAX: int = 2**63 - 1

def _normalize_time_param(value, c_type):

"""Convert a datetime.time to its isoformat string when bound via text C-types.

Returns the isoformat string if conversion applies, otherwise *None*.

"""

if isinstance(value, datetime.time) and c_type in (

ddbc_sql_const.SQL_C_CHAR.value,

ddbc_sql_const.SQL_C_WCHAR.value,

):

return value.isoformat(timespec="microseconds")

return None

class _ArrowReader:

"""RecordBatchReader-compatible wrapper that makes ``close()`` actually

release server-side resources.

``pyarrow.RecordBatchReader.from_batches(...)`` returns a reader whose

``close()`` only releases the internal ArrowArrayStream — it does **not**

propagate into the underlying Python generator and does **not** stop the

server-side ODBC cursor. This wrapper closes that gap.

Interoperability: this class exposes ``__arrow_c_stream__`` (Arrow

PyCapsule Protocol, pyarrow >= 14), so Arrow-aware consumers

(``pyarrow.RecordBatchReader.from_stream``, ``polars.from_arrow``,

``duckdb.from_arrow``, etc.) can accept it directly without any

``isinstance(x, pyarrow.RecordBatchReader)`` check. Subclassing

``pyarrow.RecordBatchReader`` (a Cython extension type) isn't a viable

alternative because its ``from_batches`` factory returns the base class

regardless of the subclass, ``__class__`` reassignment is rejected on

Cython types, and instances cannot hold arbitrary Python attributes —

so a subclass could not carry the cursor/generator refs this wrapper

needs for cancellation semantics.

Design (optimized):

* The Python generator backing the reader carries its own ``try/finally``

block — so server-side cleanup runs symmetrically whether the user

exhausts the reader, calls ``close()`` mid-iteration, exits a ``with``

block, or just lets the reader be garbage-collected. ``close()``

itself only has to (a) call ``SQLCancel`` to unblock any fetch in

flight on another thread and (b) close the generator; the

``finally`` clause does the rest.

* ``SQLCancel`` is called *before* ``SQLFreeStmt(SQL_CLOSE)`` so a fetch

running on another thread returns cleanly first. ``SQLCancel`` is

the single ODBC entry point (with the diag-record functions) that the

spec marks as safe to call from a different thread than the one

owning the statement.

* Diagnostics are drained *before* the cursor is closed, so records

produced by a cancelled fetch are not lost; a second drain after

close picks up anything ``SQL_CLOSE`` itself emits.

* Cached ``pyarrow.ArrowInvalid`` avoids per-read imports on the

post-close error path.

* ``__del__`` is guarded against interpreter finalization.

* The public method surface is *delegated* to the inner pyarrow reader

via ``__getattr__`` rather than hand-enumerated: any method pyarrow

provides (``read_all``, ``read_pandas``, ``cast``, ``schema``,

``read_next_batch``, and anything added by future pyarrow

versions) transparently forwards. Only the methods the wrapper

genuinely intercepts stay explicit: ``close``, ``closed``,

``__arrow_c_stream__``, ``__iter__``/``__next__``,

``__enter__``/``__exit__``, ``__del__``.

The parent ``Cursor`` is **not** closed; it remains fully usable.

"""

__slots__ = ("_cursor", "_inner", "_generator", "_closed", "_arrow_invalid")

def __init__(

self,

cursor: "Cursor",

inner: "pyarrow.RecordBatchReader",

generator,

arrow_invalid_exc: type,

) -> None:

self._cursor = cursor

self._inner = inner

self._generator = generator

self._closed = False

# Cache the exception class so post-close reads in a hot loop don't

# re-import pyarrow.

self._arrow_invalid = arrow_invalid_exc

# ── Public surface mirroring pyarrow.RecordBatchReader ────────────────

@property

def closed(self) -> bool:

"""True once ``close()`` has been called."""

return self._closed

def __getattr__(self, name):

"""Delegate any attribute we don't explicitly define to the inner

``pyarrow.RecordBatchReader``.

Rationale: enumerating pyarrow's surface by hand was fragile —

methods like ``read_all()``, ``read_pandas()``, and ``cast()`` were

silently missing, breaking existing user code on upgrade, and every

future addition to ``RecordBatchReader`` would repeat the same

regression. ``__getattr__`` is only invoked when normal attribute

lookup fails, so our explicit overrides (``close``, ``closed``,

``__arrow_c_stream__``, iteration and context-manager protocols)

always win; everything else falls through to the wrapped reader.

Private / dunder names (leading ``_``) are refused so that a

partially-constructed instance during ``__del__`` cannot recurse

forever trying to resolve its own slot names via ``self._inner``.

Post-close access raises ``pyarrow.ArrowInvalid`` to match the

behaviour of the explicit ``__next__`` / ``__arrow_c_stream__``

methods — a reader that has been marked closed must not delegate

even if a retry-pending state still holds ``self._inner``.

"""

if name.startswith("_"):

raise AttributeError(name)

if self._closed:

raise self._arrow_invalid("Reader is closed")

return getattr(self._inner, name)

def __arrow_c_stream__(self, requested_schema=None):

"""Arrow PyCapsule Protocol — export as an Arrow C stream.

Implements the Arrow PyCapsule Protocol for streams (pyarrow >= 14),

so this wrapper can be consumed by any Arrow-compatible library

(``pyarrow.RecordBatchReader.from_stream``, ``polars.from_arrow``,

``duckdb.from_arrow``, ``pandas.api.interchange.from_dataframe`` for

streams, etc.) without an ``isinstance(x, pa.RecordBatchReader)``

check. See

https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html

Note: once the capsule has been consumed by the caller, the

underlying pyarrow reader's internal C stream is transferred out;

further calls to ``read_next_batch()`` on this wrapper will fail

with ``ArrowInvalid``. That mirrors pyarrow's own semantics.

"""

if self._closed:

raise self._arrow_invalid("Reader is closed")

# Delegate to the inner pyarrow reader. pyarrow >= 14 exposes

# ``__arrow_c_stream__`` directly on ``RecordBatchReader``; older

# versions do not implement the protocol. Fail explicitly rather

# than silently returning something invalid.

inner_export = getattr(self._inner, "__arrow_c_stream__", None)

if inner_export is None:

raise self._arrow_invalid(

"Arrow PyCapsule Protocol requires pyarrow>=14; "

"the installed pyarrow version does not expose "

"RecordBatchReader.__arrow_c_stream__."

)

return inner_export(requested_schema)

def __iter__(self):

return self

def __next__(self):

if self._closed:

raise self._arrow_invalid("Reader is closed")

return self._inner.read_next_batch()

def __enter__(self):

if self._closed:

raise self._arrow_invalid("Reader is closed")

return self

def __exit__(self, exc_type, exc_val, exc_tb):

self.close()

return False

def __del__(self):

# Best-effort cleanup if the user never called close() (or a previous

# close() attempt failed to release the generator and left cleanup

# incomplete) and the reader is being garbage-collected. Skip during

# interpreter shutdown — the module globals (pyarrow, ddbc_bindings)

# may already be torn down, and touching native code at that point is

# unsafe.

try:

import sys as _sys

if _sys.is_finalizing():

return

# Retry whenever the generator is still referenced — covers both

# "user never called close()" and "earlier close() raised before

# the generator was released".

if getattr(self, "_generator", None) is not None:

self.close()

except Exception: # pylint: disable=broad-exception-caught

pass

# ── Close implementation ──────────────────────────────────────────────

def close(self) -> None:

"""Synchronously stop fetching, release the server-side cursor, and

reset parent-cursor bookkeeping. Idempotent **and retry-safe**:

if a previous call raised before the generator was released (for

example because another thread was still executing it and

``generator.close()`` raised ``ValueError: generator already

executing``), subsequent calls will pick up where the failed call

left off rather than silently no-op'ing.

Most of the actual cleanup work lives in the generator's ``finally``

clause (see ``Cursor.arrow_reader``); this method just unblocks any

in-flight fetch and closes the generator, which triggers that

``finally`` block.

"""

# Fast path: cleanup already completed on a previous call. We use

# the *generator* reference — not ``_closed`` — as the completion

# marker, because ``_closed`` is flipped early (so racing reads

# raise) and must not by itself disable retry of failed cleanup.

if self._generator is None and self._cursor is None:

self._closed = True

return

# Mark closed first so any racing read raises immediately, even if

# the cleanup steps below fail and we end up retried later.

self._closed = True

# SQLCancel (cross-thread safe) — unblocks a fetch running on another

# thread so that the generator's finally clause can then run

# SQLFreeStmt(SQL_CLOSE) without risking the undefined-behaviour

# window of closing an HSTMT mid-fetch. Safe no-op for an idle stmt.

cursor = self._cursor

if cursor is not None and not cursor.closed and cursor.hstmt is not None:

try:

cursor.hstmt._cancel() # pylint: disable=protected-access

except Exception as e: # pylint: disable=broad-exception-caught

logger.debug("arrow_reader.close: SQLCancel raised: %s", e)

# Close the generator — this raises GeneratorExit inside it, which

# runs the try/finally cleanup block (SQLFreeStmt + diag drain +

# cursor bookkeeping reset). If close() raises and the generator is

# still alive (e.g. another thread is currently executing it), keep

# the reference so a subsequent close() / __del__ can retry; only

# drop refs once the generator is actually dead.

gen = self._generator

if gen is not None:

try:

gen.close()

except Exception as e: # pylint: disable=broad-exception-caught

logger.debug("arrow_reader.close: generator.close raised: %s", e)

if getattr(gen, "gi_frame", None) is not None:

# Generator still alive — leave _generator (and _cursor,

# so the next retry can re-issue SQLCancel) intact.

return

self._generator = None

# Drop strong refs so the wrapper does not extend the lifetime of

# the parent Cursor or the inner pyarrow reader.

self._cursor = None

self._inner = None

class Cursor: # pylint: disable=too-many-instance-attributes,too-many-public-methods

"""

Represents a database cursor, which is used to manage the context of a fetch operation.

Attributes:

connection: Database connection object.

description: Sequence of 7-item sequences describing one result column.

rowcount: Number of rows produced or affected by the last execute operation.

arraysize: Number of rows to fetch at a time with fetchmany().

rownumber: Track the current row index in the result set.

Methods:

__init__(connection_str) -> None.

callproc(procname, parameters=None) ->

Modified copy of the input sequence with output parameters.

close() -> None.

execute(operation, parameters=None) -> Cursor.

executemany(operation, seq_of_parameters) -> None.

fetchone() -> Single sequence or None if no more data is available.

fetchmany(size=None) -> Sequence of sequences (e.g. list of tuples).

fetchall() -> Sequence of sequences (e.g. list of tuples).

nextset() -> True if there is another result set, None otherwise.

next() -> Fetch the next row from the cursor.

setinputsizes(sizes) -> None.

setoutputsize(size, column=None) -> None.

"""

# TODO(jathakkar): Thread safety considerations

# The cursor class contains methods that are not thread-safe due to:

# 1. Methods that mutate cursor state (_reset_cursor, self.description, etc.)

# 2. Methods that call ODBC functions with shared handles (self.hstmt)

#

# These methods should be properly synchronized or redesigned when implementing

# async functionality to prevent race conditions and data corruption.

# Consider using locks, redesigning for immutability, or ensuring

# cursor objects are never shared across threads.

def __init__(self, connection: "Connection", timeout: int = 0) -> None:

"""

Initialize the cursor with a database connection.

Args:

connection: Database connection object.

timeout: Query timeout in seconds

"""

# Establish the close() invariant *first*, before any statement that

# can raise (notably ``_initialize_cursor`` below). Setting

# ``closed=False`` (not True) up front means that if ``__init__``

# fails partway — even after ``hstmt`` was allocated — a subsequent

# ``close()`` / ``__del__`` will still see a consistent view and

# correctly release whatever was allocated. Pairing it with

# ``hstmt=None`` keeps ``close()`` safe when allocation fails before

# an HSTMT exists.

self.closed: bool = False

self.hstmt: Optional[Any] = None

self._connection: "Connection" = connection # Store as private attribute

self._timeout: int = timeout

self._inputsizes: Optional[List[Union[int, Tuple[Any, ...]]]] = None

# self.connection.autocommit = False

self._initialize_cursor()

self.description: Optional[

List[

Tuple[

str,

Any,

Optional[int],

Optional[int],

Optional[int],

Optional[int],

Optional[bool],

]

]

] = None

self.rowcount: int = -1

self.arraysize: int = (

1 # Default number of rows to fetch at a time is 1, user can change it

)

self.buffer_length: int = 1024 # Default buffer length for string data

self._result_set_empty: bool = False # Add this initialization

self.last_executed_stmt: str = "" # Stores the last statement executed by this cursor

self.is_stmt_prepared: List[bool] = [

False

] # Indicates if last_executed_stmt was prepared by ddbc shim.

# Is a list instead of a bool coz bools in Python are immutable.

# Hence, we can't pass around bools by reference & modify them.

# Therefore, it must be a list with exactly one bool element.

self._rownumber = -1 # DB-API extension: last returned row index, -1 before first

# Column-name -> index map for the current result set. For catalog/metadata

# result sets this also carries lowercase and friendly aliases (see

# _prepare_metadata_result_set).

self._cached_column_map = None

self._cached_column_map_lower = None

self._cached_converter_map = None

# Raw ODBC SQL type codes (from SQLDescribeCol) per column, parallel to

# self.description. Kept so output-converter dispatch can key on the integer

# ODBC SQL type code (pyodbc-compatible), not just the mapped Python type. See #684.

self._column_sql_types = None

self._uuid_str_indices = None # Pre-computed UUID column indices for str conversion

# Cache the effective native_uuid setting for this cursor's connection.

# Resolution order: connection._native_uuid (if not None) → module-level setting.

self._conn_native_uuid = getattr(self.connection, "_native_uuid", None)

self._next_row_index = 0 # internal: index of the next row the driver will return (0-based)

self._has_result_set = False # Track if we have an active result set

self._skip_increment_for_next_fetch = (

False # Track if we need to skip incrementing the row index

)

self.messages: List[Tuple[str, str]] = [] # Store diagnostic messages

def _is_unicode_string(self, param: str) -> bool:

"""

Check if a string contains non-ASCII characters.

Args:

param: The string to check.

Returns:

True if the string contains non-ASCII characters, False otherwise.

"""

try:

param.encode("ascii")

return False # Can be encoded to ASCII, so not Unicode

except UnicodeEncodeError:

return True # Contains non-ASCII characters, so treat as Unicode

def _parse_date(self, param: str) -> Optional[datetime.date]:

"""

Attempt to parse a string as a date.

Args:

param: The string to parse.

Returns:

A datetime.date object if parsing is successful, else None.

"""

formats = ["%Y-%m-%d"]

for fmt in formats:

try:

return datetime.datetime.strptime(param, fmt).date()

except ValueError:

continue

return None

def _parse_datetime(self, param: str) -> Optional[datetime.datetime]:

"""

Attempt to parse a string as a datetime, smalldatetime, datetime2, timestamp.

Args:

param: The string to parse.

Returns:

A datetime.datetime object if parsing is successful, else None.

"""

formats = [

"%Y-%m-%dT%H:%M:%S.%f", # ISO 8601 datetime with fractional seconds

"%Y-%m-%dT%H:%M:%S", # ISO 8601 datetime

"%Y-%m-%d %H:%M:%S.%f", # Datetime with fractional seconds

"%Y-%m-%d %H:%M:%S", # Datetime without fractional seconds

]

for fmt in formats:

try:

return datetime.datetime.strptime(param, fmt) # Valid datetime

except ValueError:

continue # Try next format

return None # If all formats fail, return None

def _parse_time(self, param: str) -> Optional[datetime.time]:

"""

Attempt to parse a string as a time.

Args:

param: The string to parse.

Returns:

A datetime.time object if parsing is successful, else None.

"""

formats = [

"%H:%M:%S", # Time only

"%H:%M:%S.%f", # Time with fractional seconds

]

for fmt in formats:

try:

return datetime.datetime.strptime(param, fmt).time()

except ValueError:

continue

return None

def _get_numeric_data(self, param: decimal.Decimal) -> Any:

"""

Get the data for a numeric parameter.

Args:

param: The numeric parameter.

Returns:

numeric_data: A NumericData struct containing

the numeric data.

"""

decimal_as_tuple = param.as_tuple()

digits_tuple = decimal_as_tuple.digits

num_digits = len(digits_tuple)

exponent = decimal_as_tuple.exponent

# NaN / sNaN / Infinity report a string exponent ('n', 'N', 'F') instead of an

# int. There is no SQL NUMERIC encoding for them, so refuse here rather than

# falling through to precision=38 and letting the digit packing below emit a

# silent zero. The native detection path raises ValueError for the same input.

if isinstance(exponent, str):

raise ValueError("Cannot bind non-finite Decimal (NaN/Infinity) as SQL NUMERIC")

# Calculate the SQL precision & scale

# precision = no. of significant digits

# scale = no. digits after decimal point

if exponent >= 0:

# digits=314, exp=2 ---> '31400' --> precision=5, scale=0

precision = num_digits + exponent

scale = 0

elif (-1 * exponent) <= num_digits:

# digits=3140, exp=-3 ---> '3.140' --> precision=4, scale=3

precision = num_digits

scale = exponent * -1

else:

# digits=3140, exp=-5 ---> '0.03140' --> precision=5, scale=5

# TODO: double check the precision calculation here with SQL documentation

precision = exponent * -1

scale = exponent * -1

if precision > 38:

raise ValueError(

"Precision of the numeric value is too high - "

+ str(param)

+ ". Should be less than or equal to 38"

)

Numeric_Data = ddbc_bindings.NumericData

numeric_data = Numeric_Data()

numeric_data.scale = scale

numeric_data.precision = precision

numeric_data.sign = 1 if decimal_as_tuple.sign == 0 else 0

# strip decimal point from param & convert the significant digits to integer

# Ex: 12.34 ---> 1234

int_str = "".join(str(d) for d in digits_tuple)

if exponent > 0:

int_str = int_str + ("0" * exponent)

elif exponent < 0:

if -exponent > num_digits:

int_str = ("0" * (-exponent - num_digits)) + int_str

if int_str == "":

int_str = "0"

# Convert decimal base-10 string to python int, then to 16 little-endian bytes

big_int = int(int_str)

byte_array = bytearray(16) # SQL_MAX_NUMERIC_LEN

for i in range(16):

byte_array[i] = big_int & 0xFF

big_int >>= 8

if big_int == 0:

break

numeric_data.val = bytes(byte_array)

return numeric_data

def _get_encoding_settings(self):

"""

Get the encoding settings from the connection.

Returns:

dict: A dictionary with 'encoding' and 'ctype' keys, or default settings if not available

Raises:

OperationalError, DatabaseError: If there are unexpected database connection issues

that indicate a broken connection state. These should not be silently ignored

as they can lead to data corruption or inconsistent behavior.

"""

if hasattr(self._connection, "getencoding"):

try:

return self._connection.getencoding()

except (OperationalError, DatabaseError) as db_error:

# Log the error for debugging but re-raise for fail-fast behavior

# Silently returning defaults can lead to data corruption and hard-to-debug issues

logger.error(

"Failed to get encoding settings from connection due to database error: %s. "

"This indicates a broken connection state that should not be ignored.",

db_error,

)

# Re-raise to fail fast - users should know their connection is broken

raise

except Exception as unexpected_error:

# Handle other unexpected errors (connection closed, programming errors, etc.)

logger.error("Unexpected error getting encoding settings: %s", unexpected_error)

# Re-raise unexpected errors as well

raise

# Return default encoding settings if getencoding is not available

# This is the only case where defaults are appropriate (method doesn't exist)

return {"encoding": "utf-16le", "ctype": ddbc_sql_const.SQL_WCHAR.value}

def _get_decoding_settings(self, sql_type):

"""

Get decoding settings for a specific SQL type.

Args:

sql_type: SQL type constant (SQL_CHAR, SQL_WCHAR, etc.)

Returns:

Dictionary containing the decoding settings.

Raises:

OperationalError, DatabaseError: If there are unexpected database connection issues

that indicate a broken connection state. These should not be silently ignored

as they can lead to data corruption or inconsistent behavior.

"""

try:

# Get decoding settings from connection for this SQL type

return self._connection.getdecoding(sql_type)

except (OperationalError, DatabaseError) as db_error:

# Log the error for debugging but re-raise for fail-fast behavior

# Silently returning defaults can lead to data corruption and hard-to-debug issues

logger.error(

"Failed to get decoding settings for SQL type %s due to database error: %s. "

"This indicates a broken connection state that should not be ignored.",

sql_type,

db_error,

)

# Re-raise to fail fast - users should know their connection is broken

raise

except Exception as unexpected_error:

# Handle other unexpected errors (connection closed, programming errors, etc.)

logger.error(

"Unexpected error getting decoding settings for SQL type %s: %s",

sql_type,

unexpected_error,

)

# Re-raise unexpected errors as well

raise

def _map_sql_type( # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-return-statements,too-many-branches

self,

param: Any,

parameters_list: List[Any],

i: int,

min_val: Optional[Any] = None,

max_val: Optional[Any] = None,

) -> Tuple[int, int, int, int, bool]:

"""

Map a Python data type to the corresponding SQL type,

C type, Column size, and Decimal digits.

Takes:

- param: The parameter to map.

- parameters_list: The list of parameters to bind.

- i: The index of the parameter in the list.

Returns:

- A tuple containing the SQL type, C type, column size, and decimal digits.

"""

logger.debug("_map_sql_type: Mapping param index=%d, type=%s", i, type(param).__name__)

if param is None:

logger.debug("_map_sql_type: NULL parameter - index=%d", i)

# GH-610: Send SQL_UNKNOWN_TYPE to C++ where the describe-cache

# in BindParameters / BindParameterArray resolves the correct

# type via SQLDescribeParam (cached after first call).

return (

ddbc_sql_const.SQL_UNKNOWN_TYPE.value,

ddbc_sql_const.SQL_C_DEFAULT.value,

1,

0,

False,

)

if isinstance(param, bool):

logger.debug("_map_sql_type: BOOL detected - index=%d", i)

return (

ddbc_sql_const.SQL_BIT.value,

ddbc_sql_const.SQL_C_BIT.value,

1,

0,

False,

)

if isinstance(param, int):

# Use min_val/max_val if available

value_to_check = max_val if max_val is not None else param

min_to_check = min_val if min_val is not None else param

logger.debug(

"_map_sql_type: INT detected - index=%d, min=%s, max=%s",

i,

str(min_to_check)[:50],

str(value_to_check)[:50],

)

if 0 <= min_to_check and value_to_check <= 255:

logger.debug("_map_sql_type: INT -> TINYINT - index=%d", i)

return (

ddbc_sql_const.SQL_TINYINT.value,

ddbc_sql_const.SQL_C_TINYINT.value,

3,

0,

False,

)

if -32768 <= min_to_check and value_to_check <= 32767:

logger.debug("_map_sql_type: INT -> SMALLINT - index=%d", i)

return (

ddbc_sql_const.SQL_SMALLINT.value,

ddbc_sql_const.SQL_C_SHORT.value,

5,

0,

False,

)

if -2147483648 <= min_to_check and value_to_check <= 2147483647:

logger.debug("_map_sql_type: INT -> INTEGER - index=%d", i)

return (

ddbc_sql_const.SQL_INTEGER.value,

ddbc_sql_const.SQL_C_LONG.value,

10,

0,

False,

)

# Beyond INTEGER, the only integer SQL type is BIGINT (signed 64-bit). An int

# outside its range cannot bind, so reject here with a clear message rather than

# labelling it BIGINT and failing later at the C++ cast. Mirrors the native path.

if value_to_check > BIGINT_MAX or min_to_check < BIGINT_MIN:

offending = value_to_check if value_to_check > BIGINT_MAX else min_to_check

raise ValueError(

f"integer {offending} is out of range for SQL BIGINT [-2^63, 2^63-1]"

)

logger.debug("_map_sql_type: INT -> BIGINT - index=%d", i)

return (

ddbc_sql_const.SQL_BIGINT.value,

ddbc_sql_const.SQL_C_SBIGINT.value,

19,

0,

False,

)

if isinstance(param, float):

logger.debug("_map_sql_type: FLOAT detected - index=%d", i)

return (

ddbc_sql_const.SQL_DOUBLE.value,

ddbc_sql_const.SQL_C_DOUBLE.value,

15,

0,

False,

)

if isinstance(param, decimal.Decimal):

logger.debug("_map_sql_type: DECIMAL detected - index=%d", i)

# First check precision limit for all decimal values

decimal_as_tuple = param.as_tuple()

digits_tuple = decimal_as_tuple.digits

num_digits = len(digits_tuple)

exponent = decimal_as_tuple.exponent

# NaN / sNaN / Infinity report a string exponent ('n', 'N', 'F'). Reject them

# before the MONEY range comparison below, which would otherwise raise

# decimal.InvalidOperation for NaN, and before _get_numeric_data, which used to

# fail with TypeError for Infinity. The native detection path raises ValueError

# for the same input, so both paths now agree on type and message.

if isinstance(exponent, str):

logger.debug(

"_map_sql_type: DECIMAL non-finite value - index=%d, exponent=%s", i, exponent

)

raise ValueError("Cannot bind non-finite Decimal (NaN/Infinity) as SQL NUMERIC")

# Calculate the SQL precision (same logic as _get_numeric_data)

if exponent >= 0:

precision = num_digits + exponent

elif (-1 * exponent) <= num_digits:

precision = num_digits

else:

precision = exponent * -1

logger.debug(

"_map_sql_type: DECIMAL precision calculated - index=%d, precision=%d",

i,

precision,

)

if precision > 38:

logger.debug(

"_map_sql_type: DECIMAL precision too high - index=%d, precision=%d",

i,

precision,

)

raise ValueError(

f"Precision of the numeric value is too high. "

f"The maximum precision supported by SQL Server is 38, but got {precision}."

)

# Detect MONEY / SMALLMONEY range

if SMALLMONEY_MIN <= param <= SMALLMONEY_MAX:

logger.debug("_map_sql_type: DECIMAL -> SMALLMONEY - index=%d", i)

# smallmoney

parameters_list[i] = format(param, "f")

return (

ddbc_sql_const.SQL_VARCHAR.value,

ddbc_sql_const.SQL_C_CHAR.value,

len(parameters_list[i]),

0,

False,

)

if MONEY_MIN <= param <= MONEY_MAX:

logger.debug("_map_sql_type: DECIMAL -> MONEY - index=%d", i)

# money

parameters_list[i] = format(param, "f")

return (

ddbc_sql_const.SQL_VARCHAR.value,

ddbc_sql_const.SQL_C_CHAR.value,

len(parameters_list[i]),

0,

False,

)

# fallback to generic numeric binding

logger.debug("_map_sql_type: DECIMAL -> NUMERIC - index=%d", i)

parameters_list[i] = self._get_numeric_data(param)

logger.debug(

"_map_sql_type: NUMERIC created - index=%d, precision=%d, scale=%d",

i,

parameters_list[i].precision,

parameters_list[i].scale,

)

return (

ddbc_sql_const.SQL_NUMERIC.value,

ddbc_sql_const.SQL_C_NUMERIC.value,

parameters_list[i].precision,

parameters_list[i].scale,

False,

)

if isinstance(param, uuid.UUID):

logger.debug("_map_sql_type: UUID detected - index=%d", i)

parameters_list[i] = param.bytes_le

return (

ddbc_sql_const.SQL_GUID.value,

ddbc_sql_const.SQL_C_GUID.value,

16,

0,

False,

)

if isinstance(param, str):

logger.debug("_map_sql_type: STR detected - index=%d, length=%d", i, len(param))

if (

param.startswith("POINT")

or param.startswith("LINESTRING")

or param.startswith("POLYGON")

):

logger.debug("_map_sql_type: STR is geometry type - index=%d", i)

return (

ddbc_sql_const.SQL_WVARCHAR.value,

ddbc_sql_const.SQL_C_WCHAR.value,

len(param),

0,

False,

)

# String mapping logic here

is_unicode = self._is_unicode_string(param)

# Computes UTF-16 code units (handles surrogate pairs)

utf16_len = sum(2 if ord(c) > 0xFFFF else 1 for c in param)

logger.debug(

"_map_sql_type: STR analysis - index=%d, is_unicode=%s, utf16_len=%d",

i,

str(is_unicode),

utf16_len,

)

if utf16_len > MAX_INLINE_CHAR: # Long strings -> DAE

logger.debug("_map_sql_type: STR exceeds MAX_INLINE_CHAR, using DAE - index=%d", i)

if is_unicode:

return (

ddbc_sql_const.SQL_WVARCHAR.value,

ddbc_sql_const.SQL_C_WCHAR.value,

0,

0,

True,

)

return (

ddbc_sql_const.SQL_VARCHAR.value,

ddbc_sql_const.SQL_C_CHAR.value,

0,

0,

True,

)

# Short strings

if is_unicode:

return (

ddbc_sql_const.SQL_WVARCHAR.value,

ddbc_sql_const.SQL_C_WCHAR.value,

utf16_len,

0,

False,

)

return (

ddbc_sql_const.SQL_VARCHAR.value,

ddbc_sql_const.SQL_C_CHAR.value,

len(param),

0,

False,

)

if isinstance(param, (bytes, bytearray)):

length = len(param)

if length > 8000: # Use VARBINARY(MAX) for large blobs

return (

ddbc_sql_const.SQL_VARBINARY.value,

ddbc_sql_const.SQL_C_BINARY.value,

0,

0,

True,

)

# Small blobs → direct binding

return (

ddbc_sql_const.SQL_VARBINARY.value,

ddbc_sql_const.SQL_C_BINARY.value,

max(length, 1),

0,

False,

)

if isinstance(param, datetime.datetime):

if param.tzinfo is not None:

# Timezone-aware datetime -> DATETIMEOFFSET

return (

ddbc_sql_const.SQL_DATETIMEOFFSET.value,

ddbc_sql_const.SQL_C_SS_TIMESTAMPOFFSET.value,

34,

7,

False,

)

# Naive datetime -> TIMESTAMP

return (

ddbc_sql_const.SQL_TIMESTAMP.value,

ddbc_sql_const.SQL_C_TYPE_TIMESTAMP.value,

26,

6,

False,

)

if isinstance(param, datetime.date):

return (

ddbc_sql_const.SQL_DATE.value,

ddbc_sql_const.SQL_C_TYPE_DATE.value,

10,

0,

False,

)

if isinstance(param, datetime.time):

return (

ddbc_sql_const.SQL_TYPE_TIME.value,

ddbc_sql_const.SQL_C_CHAR.value,

16,

6,

False,

)

# For safety: unknown/unhandled Python types should not silently go to SQL

raise TypeError(

"Unsupported parameter type: The driver cannot safely convert it to a SQL type."

)

def _initialize_cursor(self) -> None:

"""

Initialize the DDBC statement handle.

"""

self._allocate_statement_handle()

self._set_timeout()

def _allocate_statement_handle(self) -> None:

"""

Allocate the DDBC statement handle.

"""

self.hstmt = self._connection._conn.alloc_statement_handle()

def _set_timeout(self) -> None:

"""

Set the query timeout attribute on the statement handle.

This is called once when the cursor is created and after any handle reallocation.

Following pyodbc's approach for better performance.

"""

if self._timeout > 0:

logger.debug("_set_timeout: Setting query timeout=%d seconds", self._timeout)

try:

timeout_value = int(self._timeout)

ret = ddbc_bindings.DDBCSQLSetStmtAttr(

self.hstmt,

ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value,

timeout_value,

)

check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

logger.debug("Query timeout set to %d seconds", timeout_value)

except Exception as e: # pylint: disable=broad-exception-caught