pyodbcからmssql-pythonへの移行

mssql-pythonドライバーは、MicrosoftのファーストパーティPythonドライバーで、Microsoft SQL向けに開発されています。 Microsoftが管理するドライバーオプションを好む場合、以下が用意されています:

• 外部ODBCドライバー依存性はありません。
• 組み込みの接続プーリング。
• 最新のPython 3.10+サポート。
• ネイティブ Microsoft Entra 認証。

主な違い

基本的な移行ステップ

以下のステップでは、pyodbcアプリケーションをmssql-pythonに移行するための主要な変更点を説明します。

1. インポートの更新

pyodbcインポートを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)と %(name)s (pyformat)パラメータスタイルの両方をサポートしています。 既存の ? クエリは変更なしで動作します:

変更前(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")

出力パラメーター

出力パラメータに頼らず、出力値を取得するためにT-SQL変数を活用してください callproc() :

変更前（pyodbc、callproc を使用）:

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:

conn.autocommit = True
pyodbc.connect(connection_string, autocommit=True)

mssql-python:

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-python:

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-python:

try:
    cursor.execute("SELECT TOP 1 * FROM NonExistentTable_XYZ")
except mssql_python.Error as e:
    # Error message contains SQLSTATE and details
    print(str(e))

コネクションプーリング

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

変更点はインポート文と接続文字列のみ(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ドライバーを展開要件から削除してください。

関連するコンテンツ

• pymssqlからの移行
• Installation
• 接続文字列
• クエリの実行
• ストアド プロシージャ