-
Notifications
You must be signed in to change notification settings - Fork 46
Expand file tree
/
Copy pathrow.py
More file actions
230 lines (191 loc) · 8.86 KB
/
row.py
File metadata and controls
230 lines (191 loc) · 8.86 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
"""
Copyright (c) Microsoft Corporation.
Licensed under the MIT license.
This module contains the Row class, which represents a single row of data
from a cursor fetch operation.
"""
import decimal
import uuid as _uuid
from typing import Any
from mssql_python.helpers import get_settings
from mssql_python.logging import logger
class Row:
"""
A row of data from a cursor fetch operation. Provides both tuple-like indexing
and attribute access to column values.
Column attribute access behavior depends on the global 'lowercase' setting:
- When enabled: Case-insensitive attribute access
- When disabled (default): Case-sensitive attribute access matching original column names
Example:
row = cursor.fetchone()
print(row[0]) # Access by index
print(row.column_name) # Access by column name (case sensitivity varies)
"""
def __init__(self, values, column_map, cursor=None, converter_map=None, uuid_str_indices=None):
"""
Initialize a Row object with values and pre-built column map.
Args:
values: List of values for this row
column_map: Pre-built column name to index mapping (shared across rows)
cursor: Optional cursor reference (for backward compatibility and lowercase access)
converter_map: Pre-computed converter map (shared across rows for performance)
uuid_str_indices: Tuple of column indices whose uuid.UUID values should be
converted to str. Pre-computed once per result set when native_uuid=False.
None means no conversion (native_uuid=True, the default).
"""
# Apply output converters if available using pre-computed converter map
if converter_map:
self._values = self._apply_output_converters_optimized(values, converter_map)
elif (
cursor
and hasattr(cursor.connection, "_output_converters")
and cursor.connection._output_converters
):
# Fallback to original method for backward compatibility
self._values = self._apply_output_converters(values, cursor)
else:
self._values = values
# Convert UUID columns to str when native_uuid=False.
# uuid_str_indices is pre-computed once at execute() time, so this is
# O(num_uuid_columns) per row — zero cost when native_uuid=True (the default).
if uuid_str_indices:
self._stringify_uuids(uuid_str_indices)
self._column_map = column_map
self._cursor = cursor
def _stringify_uuids(self, indices):
"""
Convert uuid.UUID values at the given column indices to uppercase str in-place.
This is only called when native_uuid=False. It operates directly on
self._values to avoid creating an extra list copy.
"""
vals = self._values
# If values are still the original list (no converters), we need a mutable copy
if not isinstance(vals, list):
vals = list(vals)
self._values = vals
for i in indices:
v = vals[i]
if v is not None and isinstance(v, _uuid.UUID):
vals[i] = str(v).upper()
def _apply_output_converters(self, values, cursor):
"""
Apply output converters to raw values.
Args:
values: Raw values from the database
cursor: Cursor object with connection and description
Returns:
List of converted values
"""
if not cursor.description:
return values
converted_values = list(values)
for i, (value, desc) in enumerate(zip(values, cursor.description)):
if desc is None or value is None:
continue
# Get SQL type from description
sql_type = desc[1] # type_code is at index 1 in description tuple
# Try to get a converter for this type
converter = cursor.connection.get_output_converter(sql_type)
# If no converter found for the SQL type but the value is a string or bytes,
# try the WVARCHAR converter as a fallback
if converter is None and isinstance(value, (str, bytes)):
from mssql_python.constants import ConstantsDDBC
converter = cursor.connection.get_output_converter(ConstantsDDBC.SQL_WVARCHAR.value)
# If we found a converter, apply it
if converter:
try:
# If value is already a Python type (str, int, etc.),
# we need to convert it to bytes for our converters
if isinstance(value, str):
# Encode as UTF-16LE for string values (SQL_WVARCHAR format)
value_bytes = value.encode("utf-16-le")
converted_values[i] = converter(value_bytes)
else:
converted_values[i] = converter(value)
except Exception:
logger.debug("Exception occurred in output converter", exc_info=True)
# If conversion fails, keep the original value
pass
return converted_values
def _apply_output_converters_optimized(self, values, converter_map):
"""
Apply output converters using pre-computed converter map for optimal performance.
Args:
values: Raw values from the database
converter_map: Pre-computed list of converters (one per column, None if no converter)
Returns:
List of converted values
"""
converted_values = list(values)
for i, (value, converter) in enumerate(zip(values, converter_map)):
if converter and value is not None:
try:
if isinstance(value, str):
value_bytes = value.encode("utf-16-le")
converted_values[i] = converter(value_bytes)
else:
converted_values[i] = converter(value)
except Exception:
pass
return converted_values
def __getitem__(self, index: int) -> Any:
"""Allow accessing by numeric index: row[0]"""
return self._values[index]
def __getattr__(self, name: str) -> Any:
"""
Allow accessing by column name as attribute: row.column_name
Note: Case sensitivity depends on the global 'lowercase' setting:
- When lowercase=True: Column names are stored in lowercase, enabling
case-insensitive attribute access (e.g., row.NAME, row.name, row.Name all work).
- When lowercase=False (default): Column names preserve original casing,
requiring exact case matching for attribute access.
"""
# Handle lowercase attribute access - if lowercase is enabled,
# try to match attribute names case-insensitively
if name in self._column_map:
return self._values[self._column_map[name]]
# If lowercase is enabled on the cursor, try case-insensitive lookup
if hasattr(self._cursor, "lowercase") and self._cursor.lowercase:
name_lower = name.lower()
for col_name in self._column_map:
if col_name.lower() == name_lower:
return self._values[self._column_map[col_name]]
raise AttributeError(f"Row has no attribute '{name}'")
def __eq__(self, other: Any) -> bool:
"""
Support comparison with lists for test compatibility.
This is the key change needed to fix the tests.
"""
if isinstance(other, list):
return self._values == other
if isinstance(other, Row):
return self._values == other._values
return super().__eq__(other)
def __len__(self) -> int:
"""Return the number of values in the row"""
return len(self._values)
def __iter__(self) -> Any:
"""Allow iteration through values"""
return iter(self._values)
def __str__(self) -> str:
"""Return string representation of the row"""
# Local import to avoid circular dependency
from mssql_python import getDecimalSeparator
parts = []
for value in self:
if isinstance(value, decimal.Decimal):
# Apply custom decimal separator for display
sep = getDecimalSeparator()
if sep != "." and value is not None:
s = str(value)
if "." in s:
s = s.replace(".", sep)
parts.append(s)
else:
parts.append(str(value))
else:
parts.append(repr(value))
return "(" + ", ".join(parts) + ")"
def __repr__(self) -> str:
"""Return a detailed string representation for debugging"""
return repr(tuple(self._values))