mssql-python驱动通过取指操作返回行数据作为 Row 对象。 这些对象提供了灵活的访问模式：
-
属性访问 （
row.ColumnName）是命名列中最易读的选项。 当你的查询有一个已知且稳定的列列表时，才使用它。 -
字符串键访问 （
row['ColumnName']）提供按列表名称访问字典风格的访问，适合需要程序列查找或列名包含空格或特殊字符时。 -
索引访问 （
row[0]）适用于动态查询，因为开发时列名未知，或处理SELECT *结果时。 - 对于列数较少且固定的环体，元组解包（
a, b, c = row）是最简洁的选择。
属性访问
按名称直接访问列的值：
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()
print(row.ProductID) # 1
print(row.Name) # 'Adjustable Race'
print(row.ListPrice) # 0.00
事例敏感性
列名访问区分大小写，且与 SQL Server 返回的列名一致。 如果你的数据库使用不一致的外壳，可以使用SQL AS 别名来规范名称，或者启用 lowercase 模块设置（参见 模块配置）：
cursor.execute("SELECT FirstName, LastName, EmailPromotion FROM Person.Person WHERE BusinessEntityID < 10")
row = cursor.fetchone()
print(row.FirstName) # Works
print(row.LastName) # Works
print(row.EmailPromotion) # Works
print(row.firstname) # AttributeError - wrong case
列别名
使用SQL别名创建友好属性名称：
cursor.execute("""
SELECT
p.ProductID,
p.Name,
c.Name AS Category
FROM Production.Product p
JOIN Production.ProductSubcategory c ON p.ProductSubcategoryID = c.ProductSubcategoryID
WHERE p.ProductSubcategoryID IS NOT NULL
""")
for row in cursor:
print(f"{row.Name} ({row.Category})")
索引访问
按零为基础的列索引访问值：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
print(row[0]) # ProductID
print(row[1]) # Name
print(row[2]) # ListPrice
负索引
该驱动程序支持类似 Python 的负索引：
cursor.execute("""
SELECT Demo.A, Demo.B, Demo.C, Demo.D
FROM (VALUES (1, 2, 3, 4)) AS Demo(A, B, C, D)
""")
row = cursor.fetchone()
print(row[-1]) # Last column (D)
print(row[-2]) # Second to last (C)
字符串键访问
使用字典式语法按名称访问列值：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()
print(row['ProductID']) # 1
print(row['Name']) # 'Adjustable Race'
print(row['ListPrice']) # Decimal('0.00')
当列名包含空格或特殊字符，或需要程序访问列时，这非常有用：
column_name = 'ListPrice'
value = row[column_name] # Programmatic column access
切片
使用切片提取多个值：
cursor.execute("""
SELECT Demo.A, Demo.B, Demo.C, Demo.D, Demo.E
FROM (VALUES (1, 2, 3, 4, 5)) AS Demo(A, B, C, D, E)
""")
row = cursor.fetchone()
print(row[1:4]) # Columns B, C, D (indices 1, 2, 3)
print(row[:2]) # First two columns (A, B)
print(row[2:]) # From C to end
元组解包
将行值直接解压为变量：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
for product_id, name, price in cursor:
print(f"#{product_id}: {name} - ${price}")
部分拆包
用于 * 捕捉剩余值：
cursor.execute("""
SELECT TOP 2 ProductID, Name, ProductNumber, Color, Size, Weight
FROM Production.Product
WHERE ProductNumber IS NOT NULL
ORDER BY ProductID
""")
for product_id, name, *rest in cursor:
print(f"{product_id}: {name}, extra columns: {rest}")
行长与迭代
处理行维度并遍历列值：
获取列数
用来 len() 计算一行的列数：
cursor.execute("SELECT * FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
print(len(row)) # Number of columns
对数值进行迭代
按顺序循环列值：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
for value in row:
print(value)
检查是否存在该列
用于 hasattr() 检测列名是否存在：
# Use hasattr to check for column name
if hasattr(row, 'DiscountPrice'):
print(f"Discount: {row.DiscountPrice}")
else:
print("No discount available")
转换为内置类型
行对象可以转换为标准 Python 类型，以便与其他库和 API 集成：
转换为元组
使用 tuple() 构造函数将一行转换为元组：
cursor.execute("SELECT ProductID, Name FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()
row_tuple = tuple(row)
print(row_tuple) # (1, 'Adjustable Race')
转换为列表
使用 list() 构造子将一行转换为列表：
row_list = list(row)
print(row_list) # [1, 'Adjustable Race']
转换为字典
当你需要将行序列化成JSON、传递到模板引擎或与其他数据合并时，将行转换为字典。 构建字典 和 行 cursor.description 值：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()
# Create dict from description and values
columns = [col[0] for col in cursor.description]
row_dict = dict(zip(columns, row))
print(row_dict) # {'ProductID': 1, 'Name': 'Adjustable Race', 'ListPrice': Decimal('0.00')}
dict转换的辅助函数
创建一个可重复使用的辅助函数，将所有获取的行转换为字典：
def rows_to_dicts(cursor):
"""Convert fetched rows to list of dictionaries."""
columns = [col[0] for col in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
products = rows_to_dicts(cursor)
for p in products:
print(p['Name'])
处理可空值
驱动程序以 Python None格式返回 NULL 值：
cursor.execute("SELECT FirstName, MiddleName, LastName FROM Person.Person WHERE BusinessEntityID = 1")
row = cursor.fetchone()
if row.MiddleName is None:
full_name = f"{row.FirstName} {row.LastName}"
else:
full_name = f"{row.FirstName} {row.MiddleName} {row.LastName}"
使用光标。描述
访问列元数据与行数据并列：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
# Column information
for col in cursor.description:
print(f"Name: {col[0]}, Type: {col[1]}")
# Fetch with metadata
row = cursor.fetchone()
for i, col in enumerate(cursor.description):
print(f"{col[0]}: {row[i]}")
常见模式
以下是在现实应用中使用行对象的实用模式：
具有命名访问权限的进程行
使用属性访问处理行，以获得可读且可维护的代码：
def process_orders(conn):
cursor = conn.cursor()
cursor.execute("""
SELECT SalesOrderID, CustomerID, OrderDate, TotalDue
FROM Sales.SalesOrderHeader
WHERE Status = 5
""")
for order in cursor:
print(f"Order #{order.SalesOrderID}")
print(f" Customer: {order.CustomerID}")
print(f" Date: {order.OrderDate}")
print(f" Total: ${order.TotalDue:.2f}")
从行构建对象
对于带有域模型的应用，可以将行映射到数据类或类型对象。 这种映射为IDE提供了自动补全、类型检查功能，以及数据库行与应用逻辑之间的明确界限。
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
@dataclass
class Product:
id: int
name: str
price: Decimal
created: date
def get_products(conn) -> list[Product]:
cursor = conn.cursor()
cursor.execute("SELECT ProductID, Name, ListPrice, SellStartDate FROM Production.Product WHERE ProductID < 10")
return [
Product(
id=row.ProductID,
name=row.Name,
price=row.ListPrice,
created=row.SellStartDate
)
for row in cursor
]
导出为JSON
将行对象序列化为JSON，并对datetime和十进制值进行自定义类型处理：
import json
from datetime import date, datetime
from decimal import Decimal
def json_serializer(obj):
"""Custom serializer for non-JSON types."""
if isinstance(obj, (date, datetime)):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
raise TypeError(f"Type {type(obj)} not serializable")
def export_to_json(cursor, filename):
columns = [col[0] for col in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
with open(filename, 'w') as f:
json.dump(rows, f, default=json_serializer, indent=2)
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
export_to_json(cursor, "products.json")
聚合成组
按列值分组行，并将其收集到词典中进行分析或显示：
from collections import defaultdict
cursor.execute("""
SELECT c.Name AS CategoryName, p.Name AS ProductName, p.ListPrice
FROM Production.Product p
JOIN Production.ProductSubcategory c ON p.ProductSubcategoryID = c.ProductSubcategoryID
WHERE p.ProductSubcategoryID IS NOT NULL
ORDER BY c.Name
""")
products_by_category = defaultdict(list)
for row in cursor:
products_by_category[row.CategoryName].append({
'name': row.ProductName,
'price': row.ListPrice
})
for category, products in products_by_category.items():
print(f"\n{category}:")
for p in products:
print(f" - {p['name']}: ${p['price']}")