Microsoft Python SQL Server ドライバー - mssql-python

mssql-pythonMicrosoftのSQL Server、Azure SQL Database、Azure SQL Managed Instance、およびMicrosoft FabricのSQLデータベース向けのPythonドライバーです。 Direct Database Connectivity(DDBC)を使っているので、外部ドライバーマネージャーをインストールせずに接続できます。 このドライバーはPython 3.10以降をサポートし、Python Database API Specification 2.0に準拠しつつ、日常開発のためにPythonに優しい改善を加えています。

出発点を選択する

Azure SQLの運用ベースライン

このサンプルは、管理型アイデンティティとTabular Data Stream(TDS)8.0暗号化を用いたAzure SQLに接続しています。 環境変数から接続設定を読み取り、ログインタイムアウトとクエリタイムアウトを別々に設定します。 リトライごとに、失敗した接続オブジェクトを再利用する代わりに、connect()が再度呼び出されます。 コンテキストマネージャーは、各試み後に接続とカーソルを解放します。

このヘルパーは、繰り返しても安全な冪等な読み取りクエリにのみ使用してください。 個々の書き込みや部分的に完了したトランザクションの再生には使わないでください。

接続文字列内のConnectRetryCountおよびConnectRetryIntervalキーワードにより、SQL Server のアイドル接続の回復性が有効になります。つまり、ドライバーは切断されたアイドル接続を透過的に再接続します。 これは、このサンプルのアプリケーションレベルの再試行とは異なります。リトライは、デッドロックやクエリタイムアウトなどの一時的なエラーで失敗した クエリ を再試行します。 この二つは補完的なものなので、両方保持してください。

Azureのエラーチェックは、診断でエラー番号が省略されていても標準の英語エラー40501メッセージと一致します。 また、ddbc_error が存在する場合は、既知の番号もチェックします。 マッチングはベストエフォートです。ローカライズされたメッセージやフォーマットが異なるメッセージが一致しないことがあります。 詳細は「一時的な接続エラー」を参照してください。

import logging
import os
import re
import time

import mssql_python

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("app")

TRANSIENT_DRIVER_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
    "Serialization failure",
    "Statement completion unknown",
})

AZURE_THROTTLING_ERRORS = frozenset({
    40197, 40501, 40540, 40613, 40680, 49918, 49919, 49920, 10928, 10929,
})

_ERROR_NUMBER_RE = re.compile(r"\b(?:Error|Msg)\s+(\d+)\b")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


def is_azure_throttling(error: BaseException) -> bool:
    """Best-effort detection of Azure SQL throttling in ProgrammingError text."""
    if not isinstance(error, mssql_python.ProgrammingError):
        return False
    ddbc_text = getattr(error, "ddbc_error", "") or ""
    return (
        _AZURE_40501_MESSAGE in ddbc_text
        or any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))
    )


def query_with_retry(
    conn_str: str,
    sql: str,
    *params,
    max_attempts: int = 3,
    login_timeout_s: int = 5,
    query_timeout_s: int = 10,
) -> list:
    """Fetch an idempotent read query, reconnecting before each retry."""
    for attempt in range(1, max_attempts + 1):
        try:
            with mssql_python.connect(conn_str, timeout=login_timeout_s) as conn:
                conn.timeout = query_timeout_s
                with conn.cursor() as cursor:
                    cursor.execute(sql, *params)
                    rows = cursor.fetchall()
            logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return rows
        except mssql_python.Error as exc:
            azure_throttling = is_azure_throttling(exc)
            transient = (
                isinstance(exc, mssql_python.OperationalError)
                and exc.driver_error in TRANSIENT_DRIVER_ERRORS
            )
            if not (transient or azure_throttling) or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = (10 if azure_throttling else 1) * (2 ** (attempt - 1))
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise ValueError("max_attempts must be at least 1")


def main() -> None:
    # Read configuration from the environment; never hard-code secrets.
    server = os.environ["SQL_SERVER"]      # for example, <server>.database.windows.net
    database = os.environ["SQL_DATABASE"]  # for example, AdventureWorks
    client_id = os.getenv("AZURE_CLIENT_ID")  # set for a user-assigned managed identity

    # Authenticate with the workload's managed identity over TDS 8.0 encryption.
    # ConnectRetryCount/ConnectRetryInterval transparently reconnect a dropped
    # idle connection; they don't replay a failed query.
    conn_str = (
        f"Server={server};"
        f"Database={database};"
        "Authentication=ActiveDirectoryMsi;"
        "Encrypt=strict;"
        "ConnectRetryCount=3;"
        "ConnectRetryInterval=10;"
        # Parallel dials to all resolved IPs; safe on single-IP targets.
        "MultiSubnetFailover=Yes;"
    )
    if client_id:
        conn_str += f"UID={client_id};"

    query = """
        SELECT TOP 10
            p.BusinessEntityID,
            p.FirstName,
            p.LastName
        FROM Person.Person AS p
        ORDER BY p.BusinessEntityID;
    """

    try:
        rows = query_with_retry(conn_str, query)
        for business_entity_id, first_name, last_name in rows:
            print(f"{business_entity_id}\t{first_name}\t{last_name}")
    except mssql_python.Error:
        logger.exception("query failed")
        raise


if __name__ == "__main__":
    main()

このサンプルの各懸念に関する詳細なガイダンスについては、Microsoft Entra認証、接続プーリング、暗号化と証明書、再試行ロジック、エラー処理を参照してください。

主要な機能

  • PEP 249準拠:標準 connect、 cursor、 execute、 fetch* インターフェースに加え、Pythonic拡張機能も利用できます。
  • 直接データベース接続(DDBC):外部ドライバーマネージャーは不要。 mssql-pythonをインストールしれば接続の準備が整います。
  • Microsoft Entra ID認証:管理型アイデンティティやサービスプリンシパルを含む認証モードの組み込みサポート。
  • SQL ServerおよびWindows 認証:対応プラットフォーム上でのSQLログイン、Kerberos、Windowsシングルサインオン(SSO)。
  • バルクコピー:TDSプロトコルをネイティブサポートし、大規模なデータ負荷向けの 高性能バルクインサート 。
  • ネイティブのデータ型サポート: JSON、XML、空間、スパースカラム、datetimeoffset、10進数/マネー と精密な処理が可能です。
  • Apache Arrow統合: pandas、Polars、DuckDBとの高速なデータ交換を実現するゼロコピーの結果セット。
  • 非同期パターン: asyncioベースのアプリケーションやThreadPoolExecutorのFastAPIの回避策でドライバーを使用してください。 積分パターンについては 非同期パターン を参照してください。
  • TLS (既定): ネイティブ ODBC プロバイダーを使用する場合、TLS 暗号化と証明書の検証が既定でオンになります。 Encrypt=strict を設定すると、TDS 8.0 の暗号化を利用できます。

概要

[アーティクル] 説明
Installation mssql-pythonをインストールしてPython環境を確認してください。
クイックスタート:mssql-pythonと接続しましょう ローカルまたはテスト用のSQL Serverインスタンスに接続し、最初のクエリを実行してください。
クイックスタート:Jupyter Notebookから接続 mssql-pythonをノートブック内で使い、インタラクティブなデータ探索をしましょう。
クイックスタート:一括コピー 大量データセットをSQL ServerにまとめコピーAPIで移動させます。
クイックスタート:ラピッドプロトタイピング 小さなスクリプトや概念実証を素早く作りましょう。
クイックスタート:繰り返し展開 SQLと連携するPythonアプリケーションをパッケージ化、設定、出荷しましょう。
Apache Arrow クイックスタート 分析ワークフロー用のApache Arrowテーブルとしてクエリ結果を取得すること。

設定と認証

[アーティクル] 説明
mssql-odbc provider (Preview) プレビューネイティブプロバイダーを選択し、サポートされている機能や既知の問題を確認しましょう。
接続文字列 接続文字列の構文、よく使われるキーワード、そして例。
接続文字列をプログラム的に構築する 設定や秘密から安全に接続文字列を構成してください。
接続管理 接続をきれいに開けて、再利用し、閉じてください。
接続プール プールの調整、寿命、再利用パターン。
暗号化と証明書 TLS暗号化モード、証明書検証、TDS 8.0。
Microsoft Entra 認証 管理されたアイデンティティ、サービスプリンシパル、インタラクティブ、デバイスコードフローを備えたAzure SQLのパスワードレス認証。
セキュリティのベスト プラクティス パラメーター化、シークレット管理、最小特権、暗号化。
可用性グループ Always Onの可用性グループや読み取り専用レプリカに接続してください。

データの操作

[アーティクル] 説明
クエリの実行 execute、executemany、複数ステートメント バッチ、および結果セット。
データの取得 fetchone、 fetchmany、 fetchall、そしてストリーミングパターン。
パラメーター化されたクエリ パラメータを安全にバインドしてSQLインジェクションを防ぎましょう。
ストアド プロシージャ プロシージャの呼び出し、出力パラメータの読み取り、結果セットの処理。
カーソル管理 カーソルの寿命、スクロール、そして配列サイズの調整。
行オブジェクト インデックス、名前、またはマッピングで行にアクセスできます。
トランザクション管理 コミット、ロールバック、セーブポイント、そしてアイソレーションレベル。
ページネーション 大きな結果セットに対するキーセットおよびオフセット ページング パターン。
エラー処理 mssql_python.Error、DatabaseError、SQL Server誤差構造。
再試行ロジック 過渡的な誤差を検出し、指数的なバックオフで再試行します。

SQL Serverのデータ型と機能

[アーティクル] 説明
データ型のマッピング SQL ServerからPythonへのテーブルと変換ルール。
デートタイムハンドリング datetime、 datetime2、 datetimeoffset、そしてタイムゾーンの考慮事項。
十進法と貨幣の種類 正確な数値タイプと decimal.Decimal 精度。
文字列およびUnicodeデータ varchar、 nvarchar、コレーション、コードページ。
NULL処理 三値論理、センチネル、パンダの相互運用性。
バイナリ データ varbinary、image、および大きなオブジェクトのストリーミング。
カスタムタイプコンバーター カスタムタイプ用のレジスタ入出力コンバータ。
一括コピー作業 一括コピー API による高スループットの挿入。
JSONデータ FOR JSONやOPENJSONを使用して、JSONの保存、クエリの実行、分解を行えます。
XMLデータ xmlデータ型、XPath、XQueryを扱いましょう。
空間データ geometry と geography の Python の型。
スパース列 広いテーブルにはスパース カラムとカラム セットを使います。
スキーマ検出 データベース、テーブル、カラム、インデックスを点検してください。

Pythonツールやフレームワークとの統合

[アーティクル] 説明
Apache Arrow との統合 ゼロコピー分析のためにArrowテーブルで結果を取得しましょう。
パンダス統合 クエリ結果をDataFrameに読み込み、書き戻します。
極子積分 列形式のワークロードにはPolarsをmssql-pythonと組み合わせて使います。
DuckDB統合 ローカルのDuckDBテーブルと並べてSQL Serverデータをクエリします。
FastAPI統合 mssql-pythonをFastAPIサービスに接続します。
FastAPIアプリケーションのテストと展開 FastAPIサービスに対してプーリング、エラー、認証、テスト、デプロイ設定の設定を設定します。
Flask 連携 Flaskアプリケーションではmssql-pythonを使いましょう。
非同期パターン mssql-pythonと asyncio およびスレッドプールを組み合わせましょう。
データアクセスと分析パターン SQLデータに対するカーソルアクセス、Arrow抽出、pandas、Polars、DuckDBでの分析向けに、最適な読み取りパスを選択してください。
データのロードと移動パターン 行挿入、一括コピー、 MERGE アップサート、DataFrameの読み込み、CSV取り込みのために適切な書き込みパスを選択してください。

デプロイと運用

[アーティクル] 説明
コンテナーとローカル開発 SQLに接続するPythonアプリケーション用にDockerコンテナ、devcontainer、CIパイプラインを設定しましょう。
パフォーマンスチューニング プールの調整、Batch サイズ、準備済みのステートメント、一括コピー。
Troubleshooting 症状やエラーメッセージで指針を見つけましょう。
設置と接続のトラブルシューティング インストール、接続、コンテナ、CIの問題を解決します。
クエリ、データ、操作のトラブルシューティング クエリ、データ型、パフォーマンス、トランザクション、大量コピーの問題を解決します。
モジュール構成 モジュールレベルの設定、ログフック、機能フラグなどがあります。

mssql-pythonに移行する

[アーティクル] 説明
pyodbcからの移行 pyodbcのAPIや接続文字列をmssql-pythonにマッピングします。
pymssqlからの移行 pymssqlをmssql-pythonに置き換えつつ、動作は保持してください。
SQLiteからの移行 ローカルのSQLiteワークロードをSQL ServerまたはAzure SQLに移してください。
PostgreSQL からの移行 Python開発者がPostgreSQLからSQL Serverに移行するためのワンストップガイド、mssql-python。

機能をリクエストする

機能をリクエストするには、mssql-pythonのGitHubリポジトリでイシューを開いてください。

関連するコンテンツ

  • 最終更新: