mssql-pythonドライバーは、fetch操作から行データを 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()を使って、1行の列数を取得します:
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")
組み込み型への変換
行オブジェクトは他のライブラリやAPIとの統合のために標準的なPython型に変換可能です:
タプルへの変換
tuple() constructorを使って行をタプルに変換します:
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'])
nullable値の扱い
ドライバーはNULL値を以下の通り返Python None:
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にシリアライズし、datetimeと10進数の値のカスタムタイプ処理を行します:
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']}")