mssql-python 中的列物件

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']}")