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')}
字典轉換的輔助函式
建立一個可重用的輔助函式,將所有擷取的列轉換為字典:
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.description
存取欄位元資料與列資料並存:
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]}")
常見模式
以下是實際應用中處理 Row 物件的實務模式:
具有命名存取權限的程序列
使用屬性存取來處理可讀且可維護的程式碼列:
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,並自訂日期時間與十進位值的型別處理:
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']}")