pyodbc에서 mssql-python으로 마이그레이션하기

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=AdventureWorks2022;"
    "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=AdventureWorks2022;"
        "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=AdventureWorks2022;"
        "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 드라이버를 배포 요구사항에서 제거하세요.