mssql-pythonのクエリ、データ、操作の問題をトラブルシューティングしてください

この記事を使って、 mssql-python ドライバのクエリ実行、データ型、パフォーマンス、トランザクション、一括コピーの問題を診断してください。

クエリ実行の問題

テーブルまたはオブジェクトが見つからない

症状：

ProgrammingError: [42S02] (208) Invalid object name 'TableName'.

考えられる原因と解決策:

誤ったデータベースコンテキスト

cursor.execute("SELECT DB_NAME()")
print(cursor.fetchone()[0])

スキーマは指定されていません

cursor.execute("SELECT * FROM dbo.TableName")

テーブルは存在しません

cursor.execute("""
    SELECT TABLE_NAME
    FROM INFORMATION_SCHEMA.TABLES
    WHERE TABLE_NAME = 'TableName'
""")

構文エラー

症状：

ProgrammingError: [42000] (102) Incorrect syntax near '...'.

ソリューション:

SQL Server Management Studio(SSMS)でSQL文をテストして構文を検証します。

文字列補間の代わりにパラメータ付きクエリを使う:

# Don't use string interpolation for query parameters.
cursor.execute(f"SELECT * FROM Production.Product WHERE Name = '{name}'")

# Use parameters.
cursor.execute(
    "SELECT * FROM Production.Product WHERE Name = %(name)s",
    {"name": name},
)

パラメータ誤差

症状：

ProgrammingError: [07001] Wrong number of parameters

ソリューション:

プレースホルダーやパラメータを数えましょう。カウントは一致しなければなりません。

正しいパラメータスタイルを選択してください:

# Qmark style: positional parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = ? AND Name LIKE ?",
    (1, "Adjustable%"),
)
print(cursor.fetchone())

# Pyformat style: named parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = %(id)s AND Name LIKE %(name)s",
    {"id": 1, "name": "Adjustable%"},
)
print(cursor.fetchone())

データ型の問題

日付時変換エラー

症状：

DataError: [22007] Invalid datetime format

Solution:

文字列の代わりにPython datetimeオブジェクトを使いましょう。

from datetime import datetime

cursor.execute("CREATE TABLE #Events (EventDate DATETIME)")

# This value raises an error because the date is invalid.
try:
    cursor.execute(
        "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
        {"event_date": "2024-13-45"},
    )
except Exception as e:
    print(f"Expected error: {e}")

# Use a Python datetime object.
cursor.execute(
    "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
    {"event_date": datetime(2024, 3, 15)},
)
cursor.execute("SELECT EventDate FROM #Events")
print(cursor.fetchone())

十進精度の問題

症状：

数字は切り詰められたり、誤って丸められたりします。

Solution:

正確な数値表示には decimal.Decimal を使います:

from decimal import Decimal

cursor.execute("CREATE TABLE #PriceDemo (ListPrice DECIMAL(10, 2))")
cursor.execute(
    "INSERT INTO #PriceDemo (ListPrice) VALUES (%(list_price)s)",
    {"list_price": Decimal("19.99")},
)

Unicodeエンコーディングの問題点

症状：

特殊文字は乱れたりエラーを生んだりします。

ソリューション:

データベース内のUnicodeデータには nvarchar 列を使いましょう。

弦を直接渡す。ドライバーはエンコーディングを担当します:

cursor.execute("CREATE TABLE #UnicodeDemo (Name NVARCHAR(50))")
cursor.execute(
    "INSERT INTO #UnicodeDemo (Name) VALUES (%(name)s)",
    {"name": "日本語"},
)
cursor.execute("SELECT Name FROM #UnicodeDemo")
print(cursor.fetchone())

パフォーマンスの問題

クエリ実行の遅さ

考えられる原因と解決策:

欠けているインデックス: SSMSのクエリ実行計画を確認してください。

大規模な結果セット:fetchall()の代わりにfetchmany()を使う:

cursor.arraysize = 1000
while True:
    rows = cursor.fetchmany()
    if not rows:
        break
    process_rows(rows)

接続プーリング無効: プーリングを有効にする:

import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

大きな結果に対する記憶の問題

症状：

Pythonプロセスはメモリ不足で動作します。

ソリューション:

すべての行をメモリに読み込むのではなく、結果をストリーミングします。
```
cursor.execute("SELECT * FROM LargeTable")
for row in cursor:
    process_row(row)
```

サーバーサイドページネーションを使いましょう。

page_size = 1000
offset = 0

while True:
    cursor.execute(
        "SELECT * FROM LargeTable ORDER BY ID "
        "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
        (offset, page_size),
    )
    rows = cursor.fetchall()
    if not rows:
        break
    process_rows(rows)
    offset += page_size

取引に関する問題

オートコミットによる一時テーブルのスコーピング

トランザクション内で作成したセッションの一時テーブル(#tablename)は、トランザクションがロールバックされると消えます。この動作は、デフォルトでオートコミットがオフである場合に混乱を引き起こすことが多いです。

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

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")

# An explicit rollback or an error removes #TempData.
conn.rollback()

# This statement fails with "Invalid object name '#TempData'".
cursor.execute("SELECT * FROM #TempData")

一時テーブルを作成したらすぐにコミットするか、オートコミットモードを使うことができます。

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
conn.commit()

cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")
conn.commit()

CREATE DATABASEのように自動コミットモードが必要なDDL文は、オープントランザクション内で失敗します。実行前にオートコミットを設定する:

conn.autocommit = True
cursor.execute("CREATE DATABASE TestDB")
conn.autocommit = False

トランザクションはコミットされていません

症状：

接続を閉じた後もデータ変更は持続しません。

Solution:

デフォルトである autocommit=Falseでは、 commit()と呼びます:

cursor.execute("CREATE TABLE #Products (Name NVARCHAR(100))")
cursor.execute(
    "INSERT INTO #Products (Name) VALUES (%(name)s)",
    {"name": "Widget"},
)
conn.commit()

あるいは、オートコミットモードを使う方法もあります:

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

デッドロックエラー

症状：

OperationalError: [40001] (1205) Transaction ... was deadlocked on lock resources with another process

Solution:

リトライロジックは即時の失敗を処理しますが、繰り返されるデッドロックは設計上の問題を示しています。デッドロックグラフを取得し、文やロックタイプを分析します。一般的な修正方法には以下のような変更があります:

操作を順序付け替えさせ、競合するトランザクション同士が同じ順序でロックを取得するようにします。
取引範囲を縮小しましょう。
ロック期間を短縮するために適切なインデックスを追加しましょう。

デッドロック解析の詳細な解説については、Deadlocksガイドをご覧ください。 Azure SQL Databaseを使用している場合は、「デッドロックの分析と防止」をご覧ください。

一括読み込みの問題

バルクコピー中の制約違反

症状：

RuntimeError: CHECK constraint ... Conflict occurred in database ...
RuntimeError: Cannot insert duplicate key ... violation of PRIMARY KEY constraint

原因:

バッチ内のデータは、プライマリキー、ユニーク、チェック、外部キー制約などのテーブル制約に違反します。

Solution:

データを読み込む前に検証してください。大規模なデータセットの場合は、データをステージングテーブルに読み込み、その後ターゲットに統合します。

cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(100))")
cursor.bulkcopy("##Staging", rows)

# Check for duplicate rows before the merge.
cursor.execute("""
    SELECT s.ID
    FROM ##Staging AS s
    INNER JOIN dbo.Target AS t
        ON s.ID = t.ID
""")
dupes = cursor.fetchall()
if dupes:
    print(f"Skipping {len(dupes)} duplicate rows")

# Insert rows that don't exist in the target.
cursor.execute("""
    INSERT INTO dbo.Target (ID, Name)
    SELECT s.ID, s.Name
    FROM ##Staging AS s
    WHERE NOT EXISTS (
        SELECT 1
        FROM dbo.Target AS t
        WHERE t.ID = s.ID
    )
""")
conn.commit()

ステージングテーブル付きのアップサートパターンについては、データロードおよび移動パターンを参照してください。

列マッピングエラー

症状：

RuntimeError: Bulk copy failure - column count mismatch

原因:

データの列数が目標テーブルの列数と一致していないか、列の順序が間違っている場合もあります。

Solution:

データがテーブルスキーマと順番および数に一致していることを確認してください:

from decimal import Decimal

cursor.execute("""
    SELECT COLUMN_NAME, DATA_TYPE
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_NAME = 'MyTable'
    ORDER BY ORDINAL_POSITION
""")
for col in cursor.fetchall():
    print(col)

rows = [
    (1, "Widget", Decimal("19.99")),
    (2, "Gadget", Decimal("29.99")),
]
cursor.bulkcopy("dbo.MyTable", rows)

一括コピー時のタイプミスマッチ

症状：

データは読み込まれますが、値が切り詰められたり、丸められたり、誤りです。

原因:

Pythonの値はターゲットのカラムタイプにきれいにマッピングされません。一般的な例としては、float 列に読み込まれた値では精度が失われる可能性があることや、固定長列に長すぎる文字列が読み込まれることが挙げられます。

Solution:

スキーマに合ったPython型を使う:

from decimal import Decimal

rows = [
    # Use Decimal for decimal and numeric columns.
    (1, "Widget", Decimal("19.99")),
    # Avoid float values because they can lose precision.
    # (1, "Widget", 19.99),
]
cursor.bulkcopy("dbo.Products", rows)

NumPy型のバインディング失敗

症状：

NumPyの整数型やfloat型を使うと、パラメータは静かに失敗したりデータ型エラーを引き起こしたりします。

原因:

numpy.int64やnumpy.int32のようなNumPyタイプはNumPy 2.xではisinstance(x, int)パスできません。ドライバーの型推論がそれらを認識せず、予期せぬ挙動を引き起こします。

Solution:

NumPyの値をバインドする前にネイティブのPython型に変換してください:

import numpy as np

cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductID = %(product_id)s",
    {"product_id": int(np.int64(42))},
)

for _, row in df.iterrows():
    cursor.execute(
        "INSERT INTO #Orders (ProductID, Qty) "
        "VALUES (%(product_id)s, %(qty)s)",
        {
            "product_id": int(row["ProductID"]),
            "qty": int(row["Qty"]),
        },
    )

より大きなデータセットの場合は、 Arrow や pandas の統合パスをご利用ください。これらのパスは内部で型変換を処理します。

一時テーブルを使用した一括コピー

症状：

cursor.bulkcopy("#TempTable", data) は RuntimeError: Invalid object name '#TempTable'を発生させます。

原因:

bulkcopy() メタデータ検索の制限でセッション一時テーブル(#tablename)を解決できません。グローバル一時テーブル(##tablename)と常設テーブルは動作します。

Solution:

グローバル温度表または通常のステージング表を使いましょう:

# A global temp table is visible to all sessions and is dropped
# when the last session disconnects.
cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("##Staging", rows)

# Alternatively, use a permanent staging table.
cursor.execute("CREATE TABLE dbo.Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("dbo.Staging", rows)

小規模なデータセットでセッション一時テーブルを好む場合は、 executemany()を使いましょう。

cursor.execute("CREATE TABLE #Staging (ID INT, Name NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Staging (ID, Name) VALUES (?, ?)",
    rows,
)

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-09-21

mssql-pythonのクエリ、データ、操作の問題をトラブルシューティングしてください

クエリ実行の問題

テーブルまたはオブジェクトが見つからない

構文エラー

パラメータ誤差

データ型の問題

日付時変換エラー

十進精度の問題

Unicodeエンコーディングの問題点

パフォーマンスの問題

クエリ実行の遅さ

大きな結果に対する記憶の問題

取引に関する問題

オートコミットによる一時テーブルのスコーピング

トランザクションはコミットされていません

デッドロック エラー

一括読み込みの問題

バルクコピー中の制約違反

列マッピング エラー

一括コピー時のタイプミスマッチ

NumPy型のバインディング失敗

一時テーブルを使用した一括コピー

関連するコンテンツ

フィードバック

その他のリソース

デッドロックエラー

列マッピングエラー