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})")
인덱스 접근
0 기반 열 인덱스에 따른 접근 가치:
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")
내장형 타입으로 전환하기
행 객체는 다른 라이브러리 및 API와의 통합을 위해 표준 Python 유형으로 변환할 수 있습니다:
튜플로 변환하기
다음 구성자를 사용하여 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'])
nullable 값 작업
드라이버는 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}"
커서.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으로 내보내기
datetime과 십진수 값에 대한 맞춤 타입 처리를 사용하여 행 객체를 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']}")