mssql-python 驱动提供了多种提取方法、行访问模式和光标导航功能，用于获取查询结果。
获取方法
执行 SELECT 查询后，使用提取方法获取结果。
fetchone（）
返回单个行；如果没有更多可用行，则返回 None：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
row = cursor.fetchone()
while row:
print(f"{row.ProductID}: {row.Name} - ${row.ListPrice}")
row = cursor.fetchone()
fetchmany（）
返回行列表。
cursor.arraysize 控制默认批次大小（默认：1）：
cursor.execute("SELECT * FROM Production.Product")
cursor.arraysize = 100 # Fetch 100 rows at a time
while True:
rows = cursor.fetchmany()
if not rows:
break
for row in rows:
print(row.Name)
你也可以直接指定尺寸：
rows = cursor.fetchmany(50) # Fetch up to 50 rows
费查尔（）
返回所有剩余行作为列表：
cursor.execute("SELECT * FROM Production.Product WHERE Color = 'Black'")
rows = cursor.fetchall()
print(f"Found {len(rows)} products")
for row in rows:
print(row.Name)
fetchval（）
返回第一行的第一列，这对标量查询非常有用。
count = cursor.execute("SELECT COUNT(*) FROM Production.Product").fetchval()
print(f"Total products: {count}")
max_price = cursor.execute("SELECT MAX(ListPrice) FROM Production.Product").fetchval()
print(f"Highest price: ${max_price}")
行访问模式
该 Row 类支持多重访问模式。
索引访问
按位置访问列（基于零）：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
product_id = row[0]
name = row[1]
price = row[2]
属性访问
按名称访问列：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
product_id = row.ProductID
name = row.Name
price = row.ListPrice
小写列名称
全局启用小写属性名：
import mssql_python
settings = mssql_python.get_settings()
settings.lowercase = True
cursor.execute("SELECT ProductID, Name FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
print(row.productid, row.name) # Lowercase access
settings.lowercase = False # Restore default
迭代
行支持对值的迭代：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()
for value in row:
print(value)
光标迭代
直接遍历游标以处理各行：
cursor.execute("SELECT * FROM Production.Product")
for row in cursor:
print(row.Name)
这种模式相当于反复叫 fetchone() 。
列元数据
通过 cursor.description 访问列信息：
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
for col in cursor.description:
name, type_code, display_size, internal_size, precision, scale, null_ok = col
print(f"Column: {name}, Type: {type_code}, Nullable: {null_ok}")
行计数
该 cursor.rowcount 属性表示：
- 对于 SELECT：在开始提取之前，
execute()后返回 -1。 一旦开始提取，它会显示迄今为止已提取的累计行数。 - 对于 INSERT/UPDATE/DELETE：受影响的行数。
cursor.execute("SELECT * FROM Production.Product")
print(f"Rows returned: {cursor.rowcount}")
cursor.execute("CREATE TABLE #PriceUpd (Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)")
cursor.execute("INSERT INTO #PriceUpd VALUES ('A',10,1),('B',20,1),('C',30,2)")
cursor.execute("UPDATE #PriceUpd SET Price = Price * 1.1 WHERE CategoryID = 1")
print(f"Rows updated: {cursor.rowcount}")
光标导航
skip（）
跳过行而不获取它们：
cursor.execute("SELECT * FROM Production.Product ORDER BY ProductID")
cursor.skip(10) # Skip first 10 rows
row = cursor.fetchone() # Returns 11th row
scroll()
将光标位置向前移动：
cursor.execute("SELECT * FROM Production.Product ORDER BY ProductID")
# Move forward 5 rows from current position
cursor.scroll(5, mode='relative')
row = cursor.fetchone()
注释
驱动程序仅支持正值的 mode='relative'。 由于驱动程序使用仅向前游标，因此绝对定位和向后滚动会引发 NotSupportedError。
行号
跟踪当前位置：
cursor.execute("SELECT * FROM Production.Product")
print(f"Initial position: {cursor.rownumber}") # -1 (before first fetch)
row = cursor.fetchone()
print(f"After fetchone: {cursor.rownumber}") # 0 (first row fetched)
多个结果集
使用 nextset() 处理多个结果集：
cursor.execute("""
SELECT * FROM Production.Product WHERE Color = 'Black';
SELECT * FROM Production.ProductCategory;
SELECT COUNT(*) FROM Production.Product;
""")
# First result set
products = cursor.fetchall()
print(f"Products: {len(products)}")
# Move to second result set
if cursor.nextset():
categories = cursor.fetchall()
print(f"Categories: {len(categories)}")
# Move to third result set
if cursor.nextset():
count = cursor.fetchval()
print(f"Total count: {count}")
大型结果集
对于大型结果集，请分批处理各行，以控制内存占用：
def process_batch(rows):
# Example: print each row. Replace with your own logic.
for row in rows:
print(row)
cursor.execute("SELECT * FROM LargeTable")
cursor.arraysize = 1000
while True:
rows = cursor.fetchmany()
if not rows:
break
process_batch(rows)
print(f"Processed {cursor.rownumber} rows so far")
上下文管理器
使用上下文管理器进行自动资源清理：
with mssql_python.connect(connection_string) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM Production.Product")
for row in cursor:
print(row.Name)
# Cursor and connection closed automatically
最佳做法
-
处理大型结果集时，请使用
fetchmany()，以避免将所有内容加载到内存中。 - 完成后关闭光标以释放服务器资源。
- 使用列名 （属性访问）以获得更易读的代码。
-
在数据修改语句之后检查
rowcount - 当列可空时，显式处理
None值。