mssql-python 드라이버는 쿼리 실행, 다중 결과 집합 처리, 메모리 효율적인 관리를 위한 커서 객체를 제공합니다.
커서 기본
커서를 생성하고 사용
커서를 생성하기 위해 호출 conn.cursor() 한 후, 쿼리를 실행하고 결과를 가져오기 메서드를 사용합니다 execute() :
import mssql_python
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes"
)
# Create cursor
cursor = conn.cursor()
# Execute query
cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
# Process results
for row in cursor:
print(row.Name)
# Close cursor when done
cursor.close()
컨텍스트 매니저 패턴
이 문장을 사용하여 with 자동 정리를 위한 컨텍스트 관리자를 구현하세요:
with mssql_python.connect(connection_string) as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
products = cursor.fetchall()
# Cursor automatically closed on exit
# Connection automatically closed on exit
여러 커서
중요합니다
mssql-python 드라이버는 다중 활성 결과 집합(MARS)을 지원하지 않습니다. 한 연결에 여러 커서를 만들 수는 있지만, 한 번에 활성 쿼리를 가질 수 있는 커서는 하나뿐입니다. 같은 연결에서 다른 커서에서 실행하기 전에 항상 커서에서 모든 결과를 가져오세요.
conn = mssql_python.connect(connection_string)
# Multiple cursors on same connection
cursor1 = conn.cursor()
cursor2 = conn.cursor()
# Fetch results completely from cursor1 before using cursor2
cursor1.execute("SELECT TOP 5 ProductID, Name FROM Production.Product")
products = cursor1.fetchall()
cursor2.execute("SELECT TOP 5 ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor2.fetchall()
cursor1.close()
cursor2.close()
쿼리를 동시에 실행해야 한다면, 별도의 연결을 사용하세요:
conn1 = mssql_python.connect(connection_string)
conn2 = mssql_python.connect(connection_string)
cursor1 = conn1.cursor()
cursor2 = conn2.cursor()
cursor1.execute("SELECT TOP 5 ProductID, Name FROM Production.Product")
cursor2.execute("SELECT TOP 5 ProductCategoryID, Name FROM Production.ProductCategory")
products = cursor1.fetchall()
categories = cursor2.fetchall()
cursor1.close()
cursor2.close()
conn1.close()
conn2.close()
가져오기 방식
전체 가져오기와 반복 가져오기 비교
fetchall() 전체 결과 세트를 한 번에 메모리에 불러오거나, 커서를 반복하여 버퍼링 없이 한 행씩 처리할 수 있습니다.
# Fetch all at once - loads entire result into memory
cursor.execute("SELECT * FROM Production.Product")
all_products = cursor.fetchall()
print(f"Loaded {len(all_products)} products")
# Iterative fetch - memory efficient
cursor.execute("SELECT * FROM Production.Product")
count = 0
for row in cursor:
count += 1
print(f"Processed {count} products")
묶음으로 가져오기
배치 크기와 함께 fetchmany()를 사용하여 전체를 메모리에 로드하지 않고 큰 결과 집합을 청크 단위로 처리하세요.
def process_batch(rows):
# Example: print each row. Replace with your own logic.
for row in rows:
print(row)
def fetch_in_batches(cursor, batch_size: int = 1000):
"""Fetch results in batches to manage memory."""
while True:
batch = cursor.fetchmany(batch_size)
if not batch:
break
yield batch
cursor.execute("SELECT * FROM LargeTable")
for batch in fetch_in_batches(cursor, batch_size=5000):
process_batch(batch)
print(f"Processed batch of {len(batch)} rows")
단일 값에는 fetchval을 사용하세요
단일 값을 반환하는 스칼라 쿼리에 사용됩니다 fetchval() . 첫 번째 행의 첫 번째 열을 반환합니다.
# Efficient for scalar queries
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval() # Returns single value directly
cursor.execute("SELECT MAX(ListPrice) FROM Production.Product")
max_price = cursor.fetchval()
다중 결과 집합
여러 결과 집합을 처리합니다
nextset() 이전 세트의 모든 행을 가져온 후 현재 결과 세트를 넘어 다음 결과로 진행할 수 있습니다.
# Query returns multiple results
cursor.execute("""
SELECT TOP 3 CustomerID, AccountNumber FROM Sales.Customer;
SELECT TOP 3 SalesOrderID, OrderDate FROM Sales.SalesOrderHeader;
SELECT TOP 3 ProductID, Name FROM Production.Product;
""")
# First result set
print("Customers:")
customers = cursor.fetchall()
for c in customers:
print(f" {c.AccountNumber}")
# Move to second result set
if cursor.nextset():
print("Orders:")
orders = cursor.fetchall()
for o in orders:
print(f" Order #{o.SalesOrderID}")
# Move to third result set
if cursor.nextset():
print("Products:")
products = cursor.fetchall()
for p in products:
print(f" {p.Name}")
모든 결과 집합을 반복합니다
단일 실행 호출에서 모든 결과 집합을 소모할 때까지 nextset()False 루프를 반복합니다:
def process_all_result_sets(cursor):
"""Process all result sets from a query."""
result_sets = []
while True:
# Fetch current result set
rows = cursor.fetchall()
result_sets.append(rows)
# Try to move to next result set
if not cursor.nextset():
break
return result_sets
cursor.execute("""
SELECT TOP 3 ProductID, Name FROM Production.Product ORDER BY ProductID;
SELECT TOP 3 SalesOrderID, TotalDue FROM Sales.SalesOrderHeader ORDER BY SalesOrderID;
""")
all_results = process_all_result_sets(cursor)
print(f"Retrieved {len(all_results)} result sets")
더 많은 결과 집합이 있는지 확인해
루프에서 nextset()의 반환 값을 확인하여, 결과 집합이 몇 개인지 미리 알지 못하더라도 모든 결과 집합을 처리합니다:
cursor.execute("""
SELECT COUNT(*) AS ProductCount FROM Production.Product;
SELECT COUNT(*) AS PersonCount FROM Person.Person;
""")
result_num = 1
while True:
count = cursor.fetchval()
print(f"Result set {result_num}: {count}")
result_num += 1
if not cursor.nextset():
break
커서 설명
열 메타데이터 접근
쿼리를 실행한 후 cursor.description에는 열마다 하나씩, 이름, 타입 코드, 표시 크기, 내부 크기, 정밀도, 스케일, NULL 허용 여부로 이루어진 7개 항목 튜플의 시퀀스가 포함됩니다.
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
# Get column information
for col in cursor.description:
print(f"Column: {col[0]}, Type: {col[1]}")
# description structure: (name, type_code, display_size, internal_size,
# precision, scale, null_ok)
동적 결과 핸들러 구축
런타임에 cursor.description에서 열 목록을 구성하여 모든 쿼리에서 동작하는 결과 핸들러를 구축하세요:
def query_to_dicts(cursor) -> list[dict]:
"""Convert query results 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 = query_to_dicts(cursor)
for p in products:
print(p["Name"])
결과 없는 쿼리 처리
cursor.description는 INSERT, UPDATE, DELETE와 같은 비-SELECT 문 뒤에 None입니다. fetch 메서드를 호출하기 전에 반드시 확인하세요:
cursor.execute("CREATE TABLE #UpdDemo (Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)")
cursor.execute("INSERT INTO #UpdDemo VALUES ('Widget', 10.0, 5), ('Gadget', 20.0, 5)")
cursor.execute("UPDATE #UpdDemo SET Price = Price * 1.1 WHERE CategoryID = 5")
# description is None for non-SELECT statements
if cursor.description is None:
print(f"Updated {cursor.rowcount} rows")
else:
results = cursor.fetchall()
행 수
영향을 받는 행 추적
INSERT, UPDATE 또는 DELETE 후에 cursor.rowcount는 해당 문에 의해 영향을 받은 행 수를 반환합니다:
cursor.execute("CREATE TABLE #RowDemo (Name NVARCHAR(50), Stock INT)")
cursor.execute("INSERT INTO #RowDemo VALUES ('A', 0), ('B', 5), ('C', 0)")
cursor.execute("UPDATE #RowDemo SET Stock = -1 WHERE Stock = 0")
print(f"Rows affected: {cursor.rowcount}")
cursor.execute("DELETE FROM #RowDemo WHERE Stock = -1")
print(f"Deleted {cursor.rowcount} rows")
알 수 없는 행 수 처리
# Some operations might not return row count
cursor.execute("EXEC dbo.uspGetEmployeeManagers @BusinessEntityID = 5")
if cursor.rowcount == -1:
print("Row count not available")
else:
print(f"Affected {cursor.rowcount} rows")
행 건너뛰기
페이지네이션 대안으로 skip 사용
cursor.skip() 행을 가져오지 않고 커서 위치를 이동시킵니다. 대규모 데이터셋의 경우, 더 나은 성능을 위해 SQL 수준의 OFFSET-FETCH 페이지네이션을 선호합니다:
def get_page_using_skip(cursor, page: int, page_size: int):
"""Get a page of results using skip."""
cursor.execute("SELECT * FROM Production.Product ORDER BY ProductID")
# Skip rows from previous pages
cursor.skip((page - 1) * page_size)
# Fetch this page
return cursor.fetchmany(page_size)
# Get page 3
page_3 = get_page_using_skip(cursor, page=3, page_size=20)
비고
대규모 데이터셋의 경우, 클라이언트 측 스킵 대신 SQL 수준의 페이지네이션(OFFSET-FETCH)을 사용하세요. 이 방법이 더 효율적입니다.
진단 메시지
cursor.messages에 액세스합니다
이 속성은 messagesPEP 249에 설명된 대로 SQL 문장 실행 중 생성된 정보 메시지를 저장합니다. 이러한 메시지에는 PRINT 문의 출력과 심각도 수준이 11 미만인 RAISERROR가 포함됩니다.
속성은 각 튜플이 메시지 유형 코드와 메시지 텍스트를 포함하는 튜플 목록입니다:
conn = mssql_python.connect(connection_string, autocommit=True)
cursor = conn.cursor()
cursor.execute("PRINT 'Hello world!'")
print(cursor.messages)
Output:
[('[01000] (0)', '[Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Hello world!')]
메시지 텍스트에는 드라이버 접두사 정보가 포함되어 있는데, 드라이버가 진단 기록 SQLGetDiagRec으로 메시지를 가져오기 때문입니다.
저장 프로시저에서 메시지 캡처
실행 후 cursor.messages을 읽어 이전 구문의 PRINT 출력 또는 정보용 서버 메시지를 검색합니다:
cursor.execute("EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = 5")
results = cursor.fetchall()
# Check for any informational messages
if cursor.messages:
for msg_type, msg_text in cursor.messages:
print(f"Server message: {msg_text}")
메모리 관리
대규모 결과를 효율적으로 처리합니다
한 번에 메모리에 로드하기에는 너무 큰 테이블을 처리하려면 fetchmany()를 사용해 배치 단위로 가져옵니다:
def process_large_table(cursor, batch_size: int = 10000):
"""Process large result set without loading all into memory."""
cursor.execute("SELECT * FROM VeryLargeTable")
total_processed = 0
while True:
rows = cursor.fetchmany(batch_size)
if not rows:
break
for row in rows:
process_row(row)
total_processed += len(rows)
print(f"Progress: {total_processed} rows processed")
return total_processed
생성기 기반 처리
생성기에서 랩 배치 가져오기를 통해 결과 세트 크기와 상관없이 메모리 사용량을 일정하게 유지하면서 한 행씩 처리합니다:
def row_generator(cursor, batch_size: int = 1000):
"""Generate rows from cursor without loading all."""
while True:
rows = cursor.fetchmany(batch_size)
if not rows:
break
for row in rows:
yield row
cursor.execute("SELECT * FROM LargeTable")
for row in row_generator(cursor, batch_size=5000):
# Process one row at a time
print(row) # Replace with your own row-handling logic
즉시 커서를 닫으세요
예외가 발생하더라도 항상 블록 내 finally 커서를 닫아 서버 측 자원을 해제하세요:
def get_product(conn, product_id: int):
"""Get product and properly close cursor."""
cursor = conn.cursor()
try:
cursor.execute(
"SELECT * FROM Production.Product WHERE ProductID = %(id)s",
{"id": product_id}
)
return cursor.fetchone()
finally:
cursor.close()
커서 상태 관리
커서에 데이터가 있는지 확인해 보세요
fetchone()가 None를 반환하는지 확인하여 쿼리가 행을 하나라도 반환했는지 테스트합니다:
cursor.execute("SELECT ProductID, Name FROM Production.Product WHERE ProductID = 999")
row = cursor.fetchone()
if row is None:
print("Product not found")
else:
print(f"Found: {row.Name}")
커서 재사용
하나의 커서로 여러 쿼리를 순차적으로 실행할 수 있습니다. 각 execute() 호출은 이전 결과 세트를 대체합니다:
cursor = conn.cursor()
# Execute multiple queries with same cursor
cursor.execute("SELECT TOP 5 * FROM Sales.Customer")
customers = cursor.fetchall()
cursor.execute("SELECT TOP 5 * FROM Production.Product")
products = cursor.fetchall()
cursor.execute("SELECT TOP 5 * FROM Sales.SalesOrderHeader")
orders = cursor.fetchall()
cursor.close()
모범 사례
패턴: 커서 도우미 클래스
커서 수명 주기 관리를 헬퍼 클래스에 캡슐화하여 애플리케이션 전반에 걸쳐 보일러플레이트를 줄이세요:
class CursorManager:
"""Helper for managing cursor lifecycle."""
def __init__(self, connection):
self.conn = connection
def execute_and_fetch(self, query: str, params: dict = None) -> list:
"""Execute query and return all results."""
cursor = self.conn.cursor()
try:
cursor.execute(query, params or {})
return cursor.fetchall()
finally:
cursor.close()
def execute_scalar(self, query: str, params: dict = None):
"""Execute query and return single value."""
cursor = self.conn.cursor()
try:
cursor.execute(query, params or {})
return cursor.fetchval()
finally:
cursor.close()
def execute_non_query(self, query: str, params: dict = None) -> int:
"""Execute non-SELECT and return row count."""
cursor = self.conn.cursor()
try:
cursor.execute(query, params or {})
return cursor.rowcount
finally:
cursor.close()
# Usage
db = CursorManager(conn)
products = db.execute_and_fetch("SELECT TOP 5 Name FROM Production.Product")
count = db.execute_scalar("SELECT COUNT(*) FROM Production.Product")
db.execute_non_query("CREATE TABLE #Logs (LogID INT, Age INT)")
db.execute_non_query("INSERT INTO #Logs VALUES (1, 45), (2, 20), (3, 60)")
affected = db.execute_non_query("DELETE FROM #Logs WHERE Age > 30")
커서를 열어두지 마세요
명시적으로 닫혀 있지 않은 커서는 연결이 종료될 때까지 서버 측 자원을 유지합니다. 청소를 보장하기 위해 사용 try/finally :
# Bad: cursor left open
def get_data_bad(conn):
cursor = conn.cursor()
cursor.execute("SELECT * FROM Data")
return cursor.fetchall()
# Cursor never closed!
# Good: always close cursor
def get_data_good(conn):
cursor = conn.cursor()
try:
cursor.execute("SELECT * FROM Data")
return cursor.fetchall()
finally:
cursor.close()
커서 수명을 동작과 일치시키기
단일 연산을 위한 단기 커서를 생성하세요. 동일한 커서를 관련 연산 시퀀스에만 재사용하세요:
# Short-lived cursor for simple query
def get_user_count(conn) -> int:
cursor = conn.cursor()
try:
cursor.execute("SELECT COUNT(*) FROM Person.Person")
return cursor.fetchval()
finally:
cursor.close()
# Reuse cursor for related operations
def update_inventory(conn, items: list):
cursor = conn.cursor()
try:
for item in items:
cursor.execute(
"UPDATE Inventory SET Quantity = %(qty)s WHERE ProductID = %(id)s",
item
)
conn.commit()
finally:
cursor.close()