mssql-python 드라이버는 Microsoft SQL용으로 Microsoft가 자체 제공하는 최초의 공식 Python 드라이버입니다. Microsoft가 관리하는 드라이버 옵션을 선호한다면, 다음과 같은 기능을 제공합니다:
- 외부 ODBC 드라이버 의존성도 없습니다.
- 기본 제공 연결 풀링.
- 최신 Python 3.10+ 지원.
- Microsoft Entra 기본 인증.
주요 차이점
| 특징 | pyodbc | mssql-python |
|---|---|---|
| 매개변수 스타일 |
qmark(?) |
qmark (?) 그리고 pyformat (%(name)s) |
| ODBC 드라이버 필요 | Yes | 아니오 |
| 연결 풀링 (Connection Pooling) | 외부 | 기본 제공 |
| 최소 Python 버전 | 3.6 | 3.10 |
callproc() |
지원됨 | 구현되지 않음 |
| 자동 커밋 기본 설정 | Off | Off |
기본 마이그레이션 단계
다음 단계들은 pyodbc 애플리케이션을 mssql-python으로 마이그레이션하기 위한 주요 변경 사항을 다룹니다.
1. 임포트 업데이트
pyodbc import를 mssql_python로 대체합니다:
이전 (pyodbc):
import pyodbc
(mssql-python) 이후에:
import mssql_python
2. 연결 문자열 업데이트
키워드를 DRIVER= 제거하고 인증 방법을 업데이트하세요:
이전(pyodbc, ODBC 드라이버 필요):
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=localhost;"
"DATABASE=AdventureWorks2025;"
"Trusted_Connection=yes;"
)
(mssql-python, 드라이버 필요 없음, Microsoft Entra 인증 사용) 후:
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
3. 쿼리를 있는 그대로 유지하세요
mssql-python 드라이버는 (qmark)와 ? (pyformat) 매개변수 스타일 모두 %(name)s 를 지원합니다. 기존 ? 쿼리는 변경 없이 작동합니다:
이전 (pyodbc):
cursor.execute("SELECT * FROM Production.Product WHERE ProductID = ? AND Color = ?", (1, "Red"))
이후(mssql-python, 동일한 쿼리):
cursor.execute("SELECT * FROM Production.Product WHERE ProductID = ? AND Color = ?", (1, "Red"))
4. executemany를 그대로 유지하세요
튜플과 executemany 마커가 있는 기존 ? 호출은 변경 없이 작동합니다:
이전 (pyodbc):
cursor.execute("CREATE TABLE #MigrateDemo (ID INT, Name NVARCHAR(50))")
data = [(1, "Alice"), (2, "Bob"), (3, "Carol")]
cursor.executemany("INSERT INTO #MigrateDemo (ID, Name) VALUES (?, ?)", data)
그 다음에 (mssql-python, 동일한 코드):
cursor.execute("DROP TABLE IF EXISTS #MigrateDemo")
cursor.execute("CREATE TABLE #MigrateDemo (ID INT, Name NVARCHAR(50))")
data = [(1, "Alice"), (2, "Bob"), (3, "Carol")]
cursor.executemany("INSERT INTO #MigrateDemo (ID, Name) VALUES (?, ?)", data)
저장 프로시저 마이그레이션
mssql-python 드라이버는 callproc()를 구현하지 않습니다. 다음 섹션에서는 대신 사용하는 EXECUTE 방법을 보여줍니다.
저장 프로시저에는 EXECUTE를 사용하세요
pyodbc 드라이버는 callproc()를 지원하지만, mssql-python 드라이버는 지원하지 않습니다. 대신 다음을 사용합니다 EXECUTE .
이전 (pyodbc):
cursor.callproc("dbo.uspGetEmployeeManagers", (5,))
results = cursor.fetchall()
(mssql-python) 이후에:
cursor.execute(
"EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(id)s",
{"id": 5}
)
results = cursor.fetchall()
print(f"Got {len(results)} rows")
출력 매개 변수
출력 매개변수에 callproc() 의존하지 않고 출력 값을 캡처하기 위해 T-SQL 변수를 사용하세요:
이전(pyodbc, 콜프로크 사용):
params = (category_id, pyodbc.SQL_INTEGER)
cursor.callproc("dbo.GetProductCount", params)
count = params[1].value
다음에 (mssql-python, T-SQL 변수 사용):
cursor.execute(
"""
DECLARE @count INT;
SELECT @count = COUNT(*) FROM Production.Product
WHERE ProductSubcategoryID = %(cat_id)s;
SELECT @count AS ProductCount;
""",
{"cat_id": 1}
)
product_count = cursor.fetchval()
print(f"Product count: {product_count}")
특징별 마이그레이션
다음 섹션에서는 pyodbc의 특정 기능과 mssql-python에 대응하는 기능을 다룹니다.
연결 문자열
| pyodbc 키워드 | MSSQL-python 키워드 | 비고 |
|---|---|---|
DRIVER={...} |
필요하지 않음 | ODBC 드라이버는 패키지 의존성으로 자동으로 설치됩니다. |
SERVER= |
Server= |
동작이 변경되지 않습니다. |
DATABASE= |
Database= |
동작이 변경되지 않습니다. |
Trusted_Connection= |
Trusted_Connection= |
동작이 변경되지 않습니다. |
UID= / PWD= |
UID= / PWD= |
동작이 변경되지 않습니다. |
Authentication= |
Authentication= |
같은 가치를 받아들입니다. |
자동 커밋
자동 커밋 동작은 두 드라이버 모두에서 동일합니다:
pyodbc:
conn.autocommit = True
pyodbc.connect(connection_string, autocommit=True)
MSSQL-파이썬:
conn.autocommit = True
대량 삽입
대규모 INSERT 배치의 속도를 높이기 위해 pyodbc 사용자는 fast_executemany = True를 설정합니다. mssql-python 드라이버는 매개변수화된 배치에 대해 executemany를 이미 최적화하므로, 별도의 플래그가 필요하지 않습니다. 대량 로드의 경우, 각 행마다 bulkcopy() 구문 하나를 보내는 대신 벌크 복사 프로토콜을 통해 행을 스트리밍하는 INSERT를 사용하세요. 전체 워크플로우는 '대량 복사 사용(Use bulk copy)'을 참조하세요.
pyodbc:
cursor.fast_executemany = True
cursor.executemany(query, data)
(mssql-python) 이후, executemany를 사용한 중간 규모 배치:
cursor.execute("DROP TABLE IF EXISTS #BulkTarget")
cursor.execute("CREATE TABLE #BulkTarget (ID INT, Name NVARCHAR(50))")
data = [(i, f"Item {i}") for i in range(100)]
cursor.executemany("INSERT INTO #BulkTarget (ID, Name) VALUES (?, ?)", data)
conn.commit()
(mssql-python) 이후, 대용량 로드 시 bulkcopy (선호됨):
cursor.execute("IF OBJECT_ID('##BulkTarget') IS NOT NULL DROP TABLE ##BulkTarget")
cursor.execute("CREATE TABLE ##BulkTarget (ID INT, Name NVARCHAR(50))")
conn.commit() # Commit DDL before bulkcopy
data = [(i, f"Item {i}") for i in range(100)]
result = cursor.bulkcopy("##BulkTarget", data)
print(f"Bulk copied {result['rows_copied']} rows")
cursor.execute("DROP TABLE ##BulkTarget")
conn.commit()
로우 팩토리
mssql-python 드라이버는 커스텀 행 팩토리 없이 기본적으로 속성 접근을 지원하는 객체를 반환 Row 합니다:
pyodbc(사용자 지정 행 생성기):
def namedtuple_row_factory(cursor):
from collections import namedtuple
columns = [col[0] for col in cursor.description]
Row = namedtuple("Row", columns)
return Row
MSSQL-Python (기본값으로 속성 접근):
cursor.execute("SELECT Name, ListPrice FROM Production.Product")
row = cursor.fetchone()
print(row.Name) # Attribute access works directly
print(row[0]) # Index access also works
오류 처리
mssql-python 드라이버는 pyodbc와 동일한 예외 계층 구조를 사용하기 때문에 대부분의 예외 핸들러는 모듈 이름 변경만 하면 됩니다.
예외 계층 구조
예외 클래스 이름은 드라이버 간에 직접 매핑됩니다:
pyodbc:
try:
cursor.execute(query)
except pyodbc.Error as e:
pass
except pyodbc.DatabaseError as e:
pass
except pyodbc.OperationalError as e:
pass
MSSQL-파이썬:
try:
cursor.execute("SELECT TOP 1 * FROM Production.Product")
print(cursor.fetchone())
except mssql_python.Error as e:
pass
except mssql_python.DatabaseError as e:
pass
except mssql_python.OperationalError as e:
pass
오류 세부 정보
두 드라이버 모두 예외 인수를 통해 오류 세부 정보를 노출합니다:
pyodbc:
try:
cursor.execute(query)
except pyodbc.Error as e:
sqlstate = e.args[0]
message = e.args[1]
MSSQL-파이썬:
try:
cursor.execute("SELECT TOP 1 * FROM NonExistentTable_XYZ")
except mssql_python.Error as e:
# Error message contains SQLSTATE and details
print(str(e))
연결 풀링 (Connection Pooling)
mssql-python 드라이버는 기본적으로 연결 풀링을 포함하고 있어 외부 풀링 라이브러리가 더 이상 필요하지 않습니다.
외부 풀링 제거
pyodbc와 함께 외부 풀링을 사용했다면, mssql-python 드라이버에 내장되어 있습니다:
이전(pyodbc 외부 풀):
from dbutils.pooled_db import PooledDB
pool = PooledDB(pyodbc, 5, driver="{ODBC Driver 18 for SQL Server}",
server="your_server", database="your_database",
uid="your_username", pwd="your_password")
conn = pool.connection()
적용 후(mssql-python의 내장 풀링):
conn = mssql_python.connect(connection_string)
conn.close()
풀 구성
mssql_python.pooling()으로 기본 풀 크기와 제한 시간을 재정의합니다:
import mssql_python
mssql_python.pooling()
마이그레이션 완료 예제
다음은 pyodbc로 작성한 동일한 함수를 mssql-python으로 다시 작성한 모습입니다.
변경 전 (pyodbc)
이 버전은 DRIVER 키워드를 사용하는 pyodbc 연결 문자열을 사용합니다:
import pyodbc
from datetime import date
def get_orders(customer_id: int, start_date: date):
conn = pyodbc.connect(
"DRIVER={ODBC Driver 18 for SQL Server};"
"SERVER=localhost;"
"DATABASE=AdventureWorks2025;"
"Trusted_Connection=yes;"
)
cursor = conn.cursor()
cursor.execute("""
SELECT SalesOrderID, OrderDate, TotalDue
FROM Sales.SalesOrderHeader
WHERE CustomerID = ? AND OrderDate >= ?
ORDER BY OrderDate DESC
""", (customer_id, start_date))
orders = []
for row in cursor:
orders.append({
"id": row.SalesOrderID,
"date": row.OrderDate,
"total": row.TotalDue
})
cursor.close()
conn.close()
return orders
이후 (mssql-python)
이 버전에서는 키워드가 제거되었습니다 DRIVER . 모든 쿼리, 매개변수, 행 접근 패턴은 동일하게 유지됩니다:
import mssql_python
from datetime import date
def get_orders(customer_id: int, start_date: date):
conn = mssql_python.connect(
"Server=localhost;"
"Database=AdventureWorks2025;"
"Trusted_Connection=yes;"
)
cursor = conn.cursor()
cursor.execute("""
SELECT SalesOrderID, OrderDate, TotalDue
FROM Sales.SalesOrderHeader
WHERE CustomerID = ? AND OrderDate >= ?
ORDER BY OrderDate DESC
""", (customer_id, start_date))
orders = []
for row in cursor:
orders.append({
"id": row.SalesOrderID,
"date": row.OrderDate,
"total": row.TotalDue
})
cursor.close()
conn.close()
return orders
변경 사항은 import 문과 연결 문자열(키워드 필요 없음DRIVER)뿐입니다. 모든 쿼리, 매개변수, 페치 패턴, 행 접근은 동일하게 유지됩니다.
마이그레이션 테스트
마이그레이션을 완료하기 전에 두 드라이버에 대해 동일한 쿼리를 실행하고 결과를 비교하여 동일한 동작을 확인하세요.
동등한 동작을 검증하기
동일한 쿼리를 두 드라이버에 대해 실행하고 결과가 일치한다고 주장하는 비교 함수를 사용하세요:
import pyodbc
import mssql_python
def compare_results(pyodbc_conn_str: str, mssql_conn_str: str, query: str):
"""Compare results from both drivers."""
# pyodbc query
pyodbc_conn = pyodbc.connect(pyodbc_conn_str)
pyodbc_cursor = pyodbc_conn.cursor()
pyodbc_cursor.execute(query)
pyodbc_results = pyodbc_cursor.fetchall()
pyodbc_conn.close()
# mssql-python query
mssql_conn = mssql_python.connect(mssql_conn_str)
mssql_cursor = mssql_conn.cursor()
mssql_cursor.execute(query)
mssql_results = mssql_cursor.fetchall()
mssql_conn.close()
# Compare
assert len(pyodbc_results) == len(mssql_results)
for p_row, m_row in zip(pyodbc_results, mssql_results):
assert tuple(p_row) == tuple(m_row)
print(f"Results match: {len(pyodbc_results)} rows")
Checklist
- [ ] 가져오기를
pyodbc에서mssql_python(으)로 업데이트합니다. - [ ] 연결 문자열에서
DRIVER=를 제거하세요. - [ ] 기존
?매개변수 쿼리를 유지하세요 (as-is작동합니다). - [ ] 저장 프로시저 호출에는
EXECUTE문을 사용합니다. - [ ] 외부 연결 풀링 구성을 제거하세요.
- [ ] 클래스 이름 처리 예외 업데이트.
- [ ] 모든 쿼리와 저장 프로시저를 테스트하세요.
- [ ] 데이터 타입 처리(특히 소수점과 날짜)를 검증하세요.
- [ ] ODBC 드라이버를 배포 요구사항에서 제거하세요.