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に優しい改善を加えています。

出発点を選択する

• SQL Server、Azure SQL Database、またはMicrosoft FabricのSQLデータベースに接続して最初のクエリを実行するには、まずQuickstartから始めてください:mssql-pythonドライバーで接続してください。
• ローカルSQL Serverコンテナを作成するには、まずContainerとローカル開発から始めてください。
• パスワードレス認証でAzure SQLに接続するには、まずMicrosoft Entra認証とConnection文字列から始めてください。
• インタラクティブにデータを探索するには、Jupyter Notebook から接続または迅速なプロトタイピングから始めてください。
• 大量のデータを効率的に移動するには、 Bulk copy operations や Bulk copy quickstartにアクセスしてください。
• 別のドライバーから移行するには、「 pyodbcからの移行」、 pymssqlからの移行、 SQLiteからの移行、または PostgreSQLからの移行を選んでください。

Azure SQLの運用ベースライン

このサンプルを本番向けAzure SQL接続の出発点として活用してください。 環境から設定を読み込み、マネージドIDで認証し、Tabular Data Stream(TDS)8.0暗号化を有効にします。 また、ログインおよび文ごとのクエリタイムアウトを設定し、一時的な失敗を指数関数的にバックオフ(接続エラー時は新しい接続、デッドロックなどのクエリエラー時には同じ接続)を用いて再試行し、結果をログ化し、コンテキストマネージャーによるリソース解放に依存します。

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

import logging
import os
import time

import mssql_python

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

# Transient errors that require a fresh connection to recover.
CONNECT_RETRY_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
})

# Transient errors that leave the connection usable, such as a deadlock victim
# or a query timeout, so retry on the same connection.
QUERY_RETRY_ERRORS = frozenset({
    "Serialization failure",
    "Timeout expired",
})


def connect_with_retry(conn_str: str, max_attempts: int = 3, login_timeout_s: int = 5) -> mssql_python.Connection:
    """Open a connection, retrying transient failures with exponential backoff."""
    for attempt in range(1, max_attempts + 1):
        try:
            conn = mssql_python.connect(
                conn_str,
                attrs_before={mssql_python.SQL_ATTR_LOGIN_TIMEOUT: login_timeout_s},
            )
            logger.info("connected on attempt %d/%d", attempt, max_attempts)
            return conn
        except mssql_python.OperationalError as exc:
            if exc.driver_error not in CONNECT_RETRY_ERRORS or attempt == max_attempts:
                logger.error("connect failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = 2 ** (attempt - 1)  # 1s, 2s, 4s
            logger.warning(
                "connect attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)


def execute_with_retry(
    conn: mssql_python.Connection,
    sql: str,
    *params,
    max_attempts: int = 3,
    query_timeout_s: int = 10,
) -> mssql_python.Cursor:
    """Run sql on an open connection and return the ready-to-fetch cursor.

    Retries errors that leave the connection usable so callers don't wrap each
    query in its own function. Pass query values as parameters. Retry only
    idempotent statements; wrap writes in an explicit transaction.
    """
    for attempt in range(1, max_attempts + 1):
        cursor = mssql_python.Cursor(conn, timeout=query_timeout_s)
        try:
            cursor.execute(sql, *params)
            if attempt > 1:
                logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return cursor
        except mssql_python.OperationalError as exc:
            cursor.close()
            if exc.driver_error not in QUERY_RETRY_ERRORS or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = 2 ** (attempt - 1)  # 1s, 2s, 4s
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise RuntimeError("unreachable: the retry loop exits by return or raise")


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:
        # Context managers close the cursor and connection automatically.
        with connect_with_retry(conn_str) as conn:
            with execute_with_retry(conn, query) as cursor:
                for business_entity_id, first_name, last_name in cursor.fetchall():
                    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はデフォルトで: TLS暗号化および証明書検証 はデフォルトでオン(ODBCドライバー18経由)。 Encrypt=strict を設定すると、TDS 8.0 の暗号化を利用できます。

概要

設定と認証

データの操作

SQL Serverのデータ型と機能

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

デプロイと運用

mssql-pythonに移行する

機能をリクエストする

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

関連するコンテンツ

• mssql-pythonのライフサイクルサポート
• mssql-pythonの新しいこと
• GitHubのmssql-python
• PyPI上のmssql-python
• リリースノート
• ロードマップ
• Azure Python デベロッパー センター
• PythonデータベースAPI仕様2.0