mssql-python 드라이버는 fetch 작업을 통해 행 데이터를 객체로 Row 반환합니다. 이 객체들은 유연한 접근 패턴을 제공합니다:
-
속성 접근 (
row.ColumnName)은 이름 있는 열에 대해 가장 읽기 쉬운 선택입니다. 쿼리가 알려진 안정적인 열 목록이 있을 때 사용하세요. -
문자열 키 접근 (
row['ColumnName'])은 이름별로 열을 사전 스타일로 접근할 수 있게 해주며, 프로그램 컬럼 조회가 필요하거나 열명에 공백이나 특수 문자가 포함될 때 유용합니다. -
인덱스 접근 (
row[0])은 열 이름이 개발 시점이나 결과를 처리할SELECT *때 동적 쿼리에 유용합니다. -
튜플 언패킹 (
a, b, c = row)은 열이 고정된 소수의 루프 몸체에 가장 간결한 선택입니다. - 매핑 액세스(
row._mapping)는 열명을 값으로 읽기 전용 매핑하는 것을 반환합니다. 이 기능을 사용해 행을 사전으로 변환하거나, 매핑을 기대하는 코드에 행을 전달할 수 있습니다.
속성 접근
열의 값을 이름으로 직접 접근:
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으로 직렬화해야 할 때 행을 사전 형식으로 변환하거나, 템플릿 엔진에 넘기거나, 다른 데이터와 병합해야 할 때 사용할 수 있습니다. 이 속성은 row._mapping 열 이름과 결과 집합 열 순서로 값으로 읽기 전용 매핑을 반환합니다. 일반 사전을 얻으려면 다음 dict() 단계를 통과하세요:
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()
row_dict = dict(row._mapping)
print(row_dict) # {'ProductID': 1, 'Name': 'Adjustable Race', 'ListPrice': Decimal('0.0000')}
이 부동산은 _mapping 다음과 같은 행동을 보입니다:
- 매핑은 읽기 전용입니다. 주요 인사
TypeError를 할당하면 . 값을 바꿔야 할 때는 first로dict()변환하세요. - 키는 에서 열
cursor.description명을 얻은 것입니다. 키 조회는 대소문자 구분입니다. 모듈 설정을 활성화lowercase하면 키가 소문자이고, 매핑이 그 소문자 키와 일치합니다. 행 자체의 속성 및 문자열 키 접근은 켜져 있으면lowercase대소문자 구분이 없지만,row._mapping조회는 그렇지 않습니다. - 결과 집합에 중복된 열 이름이 있을 경우, 매핑은 각 이름의 마지막 값을 유지합니다. 각 열에 고유한 이름을 부여하거나 인덱스로 값을 읽기 위해 Transact-SQL(T-SQL)
AS별칭을 사용하세요. - 이 속성은 결과
row._mapping열보다_mapping우선하므로 항상 사상을 반환합니다. 그 칼럼을row['_mapping']읽어보세요.
dict 변환을 위한 헬퍼 함수
모든 불러온 행을 사전으로 변환하는 재사용 보조 함수를 만드세요:
def rows_to_dicts(cursor):
"""Convert fetched rows to list of dictionaries."""
return [dict(row._mapping) 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):
rows = [dict(row._mapping) 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']}")