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

mssql-python 드라이버는 Microsoft SQL용으로 Microsoft가 자체 제공하는 최초의 공식 Python 드라이버입니다. Microsoft가 관리하는 드라이버 옵션을 선호한다면, 다음과 같은 기능을 제공합니다:

  • FreeTDS 의존성은 없습니다.
  • 연결마다 여러 커서가 동시에 존재합니다.
  • 기본 제공 연결 풀링.
  • 최신 Python 3.10+ 지원.
  • Microsoft Entra 기본 인증.
  • 기본적으로 속성 접근 권한을 가진 행 객체들.

주요 차이점

특징 pymssql mssql-python
매개변수 스타일 format (%s, %d) qmark (?) 그리고 pyformat (%(name)s)
네이티브 라이브러리 FreeTDS DDBC(포함)
연결 풀링 (Connection Pooling) 외부 기본 제공
연결별 커서 1 Multiple
최소 Python 버전 3.6 3.10
callproc() 지원됨 구현되지 않음
as_dict 커서 Extension 행 객체 (기본값)
대량 복사 conn.bulk_copy() cursor.bulkcopy()
자동 커밋 기본 설정 Off Off

기본 마이그레이션 단계

다음 단계에서는 pymssql 애플리케이션을 mssql-python으로 마이그레이션하는 데 가장 흔히 필요한 변경 사항을 안내합니다.

1. 임포트 업데이트

pymssql import를 mssql_python로 대체합니다:

이전 (pymssql):

import pymssql

(mssql-python) 이후에:

import mssql_python

2. 연결 호출 업데이트

pymssql은 위치 인수를 사용합니다. mssql-python 드라이버는 연결 문자열 또는 키워드 인수를 사용합니다:

이전 (pymssql, 위치 인수):

conn = pymssql.connect("<server>", "<user>", "<password>", "<database>")

이전 (pymssql, 키워드 인수):

conn = pymssql.connect(
    host=r"<server>\<instance>",
    user="<login>",
    password="<password>",
    database="<database>"
)

(mssql-python, 연결 문자열) 이후에:

conn = mssql_python.connect(
    "Server=<server>;"
    "Database=<database>;"
    "UID=<username>;"
    "PWD=<password>;"
    "Encrypt=yes;"
)

(mssql-python, Microsoft Entra 권장) 이후:

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

3. 매개변수 마커 업데이트

pymssql은 %s%d 형식 자리 표시자를 사용합니다. mssql-python 드라이버는 (qmark) 또는 ? (pyformat)를 사용합니다 %(name)s :

이전 (pymssql):

cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = %d AND FirstName = %s", (user_id, name))

(mssql-python, qmark 스타일) 이후:

user_id, name = 1, "Ken"
cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = ? AND FirstName = ?", (user_id, name))
print(cursor.fetchone())

cursor.execute(
    "SELECT * FROM Person.Person WHERE BusinessEntityID = %(id)s AND FirstName = %(name)s",
    {"id": user_id, "name": name}
)
print(cursor.fetchone())

4. executemany 업데이트

SQL 자리 표시자를 %s/%d에서 ? 또는 %(name)s로 업데이트하세요:

이전 (pymssql):

cursor.execute("CREATE TABLE #Persons (ID INT, Name NVARCHAR(50), Department NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Persons VALUES (%d, %s, %s)",
    [(1, "John", "Sales"), (2, "Jane", "Marketing")]
)

(mssql-python, qmark 스타일) 이후:

cursor.execute("IF OBJECT_ID('#Persons') IS NOT NULL DROP TABLE #Persons")
cursor.execute("CREATE TABLE #Persons (ID INT, Name NVARCHAR(50), Department NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Persons VALUES (?, ?, ?)",
    [(1, "John", "Sales"), (2, "Jane", "Marketing")]
)
cursor.execute("SELECT * FROM #Persons")
for row in cursor:
    print(row)

5. as_dict 커서 대신 행 속성을 사용하세요

pymssql에서 열 이름으로 열에 접근하려면 as_dict=True이(가) 필요합니다. mssql-python 드라이버는 기본적으로 속성과 인덱스 접근을 모두 지원하는 객체를 반환 Row 합니다:

이전 (pymssql):

cursor = conn.cursor(as_dict=True)
cursor.execute("SELECT BusinessEntityID, FirstName FROM Person.Person WHERE FirstName = %s", ("John",))
for row in cursor:
    print("ID=%d, Name=%s" % (row["BusinessEntityID"], row["FirstName"]))

이후(mssql-python, 기본적으로 속성 액세스):

cursor = conn.cursor()
cursor.execute("SELECT BusinessEntityID, FirstName FROM Person.Person WHERE FirstName = ?", ("John",))
for row in cursor:
    print(f"ID={row.BusinessEntityID}, Name={row.FirstName}")
    # Index access also works: row[0], row[1]

저장 프로시저 마이그레이션

mssql-python 드라이버는 callproc()를 구현하지 않습니다. 대신 EXECUTE 문을 사용하세요.

저장 프로시저에는 EXECUTE를 사용하세요

pymssql은 callproc()를 지원하지만, mssql-python 드라이버는 지원하지 않습니다. 대신 다음을 사용합니다 EXECUTE .

변경 전 (pymssql):

cursor.callproc("uspGetEmployeeManagers", (5,))
for row in cursor:
    print(row)

(mssql-python) 이후에:

cursor.execute("EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = ?", (5,))
for row in cursor:
    print(row)

출력 매개 변수

출력 매개변수에 callproc() 의존하지 않고 출력 값을 캡처하기 위해 T-SQL 변수를 사용하세요:

이전 (pymssql):

cursor.callproc("GetProductCount", (category_id,))
count = cursor.fetchval()

(mssql-python, T-SQL 변수) 이후:

cursor.execute("""
    DECLARE @count INT;
    SELECT @count = COUNT(*) FROM Production.Product
    WHERE ProductSubcategoryID = ?;
    SELECT @count AS ProductCount;
""", (1,))
product_count = cursor.fetchval()
print(f"Product count: {product_count}")

대량 복사 이전

pymssql은 연결에서 bulk_copy()을 호출합니다. mssql-python 드라이버는 더 많은 옵션과 함께 커서에서 bulkcopy()를 호출합니다:

이전 (pymssql):

conn.bulk_copy("##BulkDemo", [(1, 2)] * 1000)
conn.commit()

(mssql-python) 이후에:

cursor = conn.cursor()
cursor.execute("CREATE TABLE ##BulkDemo (Col1 INT, Col2 INT)")
conn.commit()
result = cursor.bulkcopy("##BulkDemo", [(1, 2)] * 1000)
print(f"Copied {result['rows_copied']} rows")
conn.commit()
cursor.execute("DROP TABLE ##BulkDemo")
conn.commit()

mssql-python bulkcopy() 메서드는 batch_size, timeout, column_mappings, keep_identity, check_constraints, table_lock, keep_nulls, fire_triggersuse_internal_transaction를 지원합니다. 자세한 내용은 대량 사본 을 참조하세요.

여러 커서

pymssql은 연결당 하나의 활성 커서만 허용합니다. mssql-python 드라이버는 여러 개의 동시 커서를 지원합니다:

이전 (pymssql):

c1 = conn.cursor()
c1.execute("SELECT TOP 5 * FROM Person.Person")
c2 = conn.cursor()
c2.execute("SELECT TOP 5 * FROM Sales.SalesOrderHeader")
c1.fetchall()

(mssql-python) 이후에:

c1 = conn.cursor()
c1.execute("SELECT TOP 5 BusinessEntityID, FirstName FROM Person.Person")
persons = c1.fetchall()

c2 = conn.cursor()
c2.execute("SELECT TOP 5 SalesOrderID FROM Sales.SalesOrderHeader")
orders = c2.fetchall()
print(f"Persons: {len(persons)}, Orders: {len(orders)}")

연결 풀링 (Connection Pooling)

pymssql에는 내장된 풀링이 없습니다. mssql-python 드라이버는 자동으로 다음과 같이 포함합니다:

이전 (pymssql, 외부 풀 필요):

from dbutils.pooled_db import PooledDB
pool = PooledDB(pymssql, host="server", user="user", password="pwd", database="db")
conn = pool.connection()

이후(mssql-python, 풀링이 자동으로 수행됨):

conn = mssql_python.connect(connection_string)
conn.close()

오류 처리

mssql-python 드라이버는 pymssql과 동일한 예외 계층 구조를 사용하기 때문에 대부분의 예외 핸들러는 모듈 이름 변경만 하면 됩니다:

이전 (pymssql):

try:
    cursor.execute(query)
except pymssql.OperationalError as e:
    print(f"Operation failed: {e}")
except pymssql.InterfaceError as e:
    print(f"Interface error: {e}")

(mssql-python) 이후에:

try:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 1 * FROM Production.Product")
    row = cursor.fetchone()
    print(row)
except mssql_python.OperationalError as e:
    print(f"Operation failed: {e}")
except mssql_python.InterfaceError as e:
    print(f"Interface error: {e}")

마이그레이션 완료 예제

다음은 pymssql로 작성한 동일한 함수를 mssql-python으로 다시 작성한 모습을 보여줍니다.

이전 (pymssql)

이 버전은 위치 연결 인자와 as_dict=True매개변수 %d 마커를 사용합니다:

import pymssql

def get_orders(customer_id: int):
    conn = pymssql.connect("<server>", "<user>", "<password>", "<database>")
    cursor = conn.cursor(as_dict=True)

    cursor.execute("""
        SELECT TOP 10 SalesOrderID, OrderDate, TotalDue
        FROM Sales.SalesOrderHeader
        WHERE CustomerID = %d
        ORDER BY OrderDate DESC
    """, (customer_id,))

    orders = []
    for row in cursor:
        orders.append({
            "id": row["SalesOrderID"],
            "date": row["OrderDate"],
            "total": row["TotalDue"]
        })

    cursor.close()
    conn.close()
    return orders

이후 (mssql-python)

주요 구조적 변경 사항은 매개변수 마커, 연결 스타일, 그리고 행 접근입니다:

import mssql_python

def get_orders(customer_id: int):
    conn = mssql_python.connect(
        "Server=<server>;"
        "Database=<database>;"
        "UID=<username>;"
        "PWD=<password>;"
    )
    cursor = conn.cursor()

    cursor.execute("""
        SELECT TOP 10 SalesOrderID, OrderDate, TotalDue
        FROM Sales.SalesOrderHeader
        WHERE CustomerID = ?
        ORDER BY OrderDate DESC
    """, (customer_id,))

    orders = []
    for row in cursor:
        orders.append({
            "id": row.SalesOrderID,
            "date": row.OrderDate,
            "total": row.TotalDue
        })

    cursor.close()
    conn.close()
    return orders

구조적 변화는 다음과 같습니다:

  1. import pymssqlimport mssql_python.
  2. 위치 기반 connect 인수 → 키워드가 포함된 연결 문자열.
  3. %d 파라미터 마커 → ?.
  4. cursor(as_dict=True)cursor() 속성 접근을 사용한(row["SalesOrderID"] 대신 row.SalesOrderID).

Checklist

  • [ ] 가져오기를 pymssql에서 mssql_python(으)로 업데이트합니다.
  • [ ] 위치 인자에서 연결 호출을 연결 문자열로 변환합니다.
  • [ ] %s/%d 매개변수 마커를 ? 또는 %(name)s로 변환하세요.
  • [ ] 저장 프로시저 호출에는 EXECUTE 문을 사용합니다.
  • [ ] as_dict=True 커서 대신 Row 속성 액세스를 사용하세요.
  • [ ] conn.bulk_copy()을(를) cursor.bulkcopy()(으)로 마이그레이션.
  • [ ] 외부 연결 풀링 구성을 제거하세요.
  • [ ] 배포 요건에서 FreeTDS를 제거하세요.
  • [ ] 클래스 이름 처리 예외 업데이트.
  • [ ] 모든 쿼리와 저장 프로시저를 테스트하세요.