この記事を使って、 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())
マルチ結果バッチにおける誤った行数
nextset() の後で rowcount が最初の文のカウントを繰り返す場合は、ドライバーをアップグレードしてください。
nextset() が正常に実行されるたびに新しい結果のカウントが更新されます。 行生成の結果は、フェッチ前の -1 から始まります。 データ修正結果は行ではなくカウントを含むため、取得前に cursor.description 必ず確認してください。
影響を受けた行の結果が必要なときは SET NOCOUNT はオフのままにしてください。 完全な例については、「 各結果をバッチ内で追跡する」を参照してください。
データ型の問題
日付時変換エラー
症状：
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())
タイムスタンプコンストラクターはマイクロ秒引数を必要とします
Timestamp(year, month, day, hour, minute, second)が欠落したTypeError引数に対してmicrosecondを発生した場合、mssql-pythonをアップグレードしてください。 現在のドライバーは6引数 DB-API コンストラクタを受け入れ、 microsecond をゼロにデフォルトしています。 詳細については、DB-API の timestamp コンストラクターを参照してください。
十進精度の問題
症状：
数字は切り詰められたり、誤って丸められたりします。
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を使用している場合は、「デッドロックの分析と防止」をご覧ください。
一括読み込みの問題
パッケージアップグレード後に一括コピーが失敗します
通常のクエリは動作しても、 bulkcopy() や bulkcopy_arrow() がネイティブのバルクコピーモジュールのインポートや読み込みに失敗する場合は、依存関係を持つドライバーをアップグレードしてください。
python -m pip install --upgrade mssql-python
python -m pip check
ドライバーには対応するmssql-python-rsランタイムが必要です。 部分的なアップグレードや --no-deps を使用したインストールは、互換性のないパッケージを残すことがあります。 プライベートパッケージインデックスを使う場合は、両方のコンパニオンパッケージをミラーリングし、ドライバーの宣言した依存ピンを保持してください。 現在のピンについては「 依存要件」を参照してください。
サポートされていない宛先名
bulkcopy()とbulkcopy_arrow()は最大で3つの識別子部分を受け入れます。
server.database.schema.table のような 4 部構成の名前は、メタデータ検索時に、行が送信される前に RuntimeError が発生します。 宛先サーバーに直接接続し、1部構成、2部構成、または3部構成のテーブル名を使用します。 括弧のエスケープや名前の誤りについては、ターゲットテーブル名を参照してください。
一括コピー中にUnicodeまたは秒の小数部が失われる
もし一括コピーでUTF-8テキストが破損したり、互換性のある datetime2 の値を読み込む際に分数秒が切れたりする場合は、ドライバーをアップグレードしてください。 現在の一括コピーランタイムはUTF-8テキストを処理し、ソースタイプと宛先タイプで利用可能な datetime2 の精度を維持します。
Unicode テキストには、nvarchar または varchar の UTF-8 コレーションを使ってください。 Python datetime値とドライバーのArrowフェッチ方式はマイクロ秒単位に制限されています。 すでにArrowのナノ秒タイムスタンプカラムに保持されている100ナノ秒の値を、最初にPython datetime に変換することなく、datetime2(7) に読み込みます。 詳細については、Unicodeと時間的精度およびArrow時間的精度を参照してください。
バルクコピー中の制約違反
症状：
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,
)