pymssqlからmssql-pythonへの移行

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

• FreeTDSに依存していません。
• 接続ごとに複数の同時カーソルがあります。
• 組み込みの接続プーリング。
• 最新のPython 3.10+サポート。
• ネイティブ Microsoft Entra 認証。
• デフォルトで属性アクセスを持つ行オブジェクト。

主な違い

基本的な移行ステップ

以下のステップでは、pymssqlアプリケーションをmssql-pythonに移行する際に必要な最も一般的な変更点を解説します。

1. インポートの更新

pymssqlインポートを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)または %(name)s (pyformat)を使用します:

以前(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 カーソルの代わりに Row 属性を使う

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)

出力パラメーター

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

以前(pymssql):

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

その後 (mssql-python, T-SQL variables):

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_triggers、 use_internal_transactionをサポートしています。 詳細は 「一括コピー 」を参照してください。

複数のカーソル

pymsSQLは接続ごとにアクティブカーソルを1つだけ許可しています。 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)}")

コネクションプーリング

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 pymssql → import mssql_python。
2. 位置指定の接続引数 → キーワード付きの接続文字列。
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を展開要件から外せ。
• [ ] クラス名の処理に関する例外の更新。
• [ ] すべてのクエリとストアドプロシージャをテストしてください。

関連するコンテンツ

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