適用於 SQL Server 的 Microsoft Python 驅動程式 - mssql-python

mssql-python是 Microsoft 用於 SQL Server、Azure SQL Database、Azure SQL 受控執行個體 以及 Microsoft Fabric 中的 SQL 資料庫的 Python 驅動程式。 它使用 Direct Database Connectivity（DDBC），因此你可以在不安裝外部驅動管理員的情況下連接。 驅動程式支援 Python 3.10 或更新版本，並符合 Python 資料庫 API 規範 2.0，同時加入對 Python 友善的日常開發改進。

選擇你的起點

• 要快速啟動本地 SQL Server 範例，請從 Quickstart 開始：連接 mssql-python 驅動程式。
• 要用無密碼認證連接 Azure SQL，請先使用 Microsoft Entra 認證與連線字串。
• 若要互動式探索資料，請從 Jupyter Notebook 或 Rapid Prototyping 開始使用 Connect 。
• 若要有效移動大量資料，請使用 批量複製作業 或批量 複製快速啟動。
• 若要從其他驅動程式遷移，請選擇 從 pyodbc 遷移、從 pymssql 遷移、從 SQLite 遷移，或 從 PostgreSQL 遷移。

Azure SQL 的生產環境基準

使用此範例作為生產導向 Azure SQL 連線的起點。 它從環境中讀取設定，使用管理身份進行認證，並啟用 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, myserver.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;"
    )
    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* 介面，以及符合 Python 風格的擴充功能。
• 直接資料庫連接（DDBC）：無需外部驅動程式管理器。 安裝完 mssql-python 就準備好連接了。
• Microsoft Entra ID 認證：內建對認證模式的支援，包括受管理身份與服務主體。
• SQL Server 與 Windows 驗證：支援平台上的 SQL 登入、Kerberos 及 Windows 單一登入（SSO）。
• 大量複製：支援原生 TDS 協定的 高效能大量插入，適用於大型資料載入。
• 原生資料型態支援： JSON、XML、空間欄位、稀疏欄位、日期時間偏移，以及十進位/貨幣 ，並能精確處理。
• Apache Arrow 整合功能：零複製結果集，可與 pandas、Polars 及 DuckDB 快速交換資料。
• 非同步模式：透過 ThreadPoolExecutor 的變通作法，在 FastAPI 和以 asyncio 為基礎的應用程式中使用此驅動程式。 關於積分模式，請參見 非同步模式 。
• 預設為 TLS： TLS 加密與憑證驗證 預設開啟（透過 ODBC 驅動程式 18）。 當你設定 Encrypt=strict時，可以使用 TDS 8.0 加密。

開始

文章	描述
Installation	安裝mssql-python並驗證你的 Python 環境。
快速入門：使用 mssql-python 進行連線	連接到本地或測試用的 SQL Server 實例，執行你的第一個查詢。
快速入門：從 Jupyter Notebook 連接	在筆記本中使用 mssql-python 進行互動式資料探索。
快速入門：批量複製	利用批量複製 API 將大型資料集移入 SQL Server。
快速入門：快速原型製作	快速建立小型腳本和概念驗證。
快速啟動：可重複部署	打包、配置並交付能與 SQL 溝通的 Python 應用程式。
Apache Arrow 快速入門	將查詢結果匯集為 Apache Arrow 表格，用於分析工作流程。

設定與認證

文章	描述
連接字串	連接字串語法、常用關鍵字與範例。
程式化建置連接字串	從設定和機密資料安全地組成連線字串。
連線管理	乾淨地打開、重複使用並關閉連接。
連接共用	池調音、壽命與重複使用模式。
加密與憑證	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。
空間資料	Python 的 geometry 和 geography 型別。
稀疏欄位	寬表格使用稀疏欄位與欄位集合。
架構探索	檢查資料庫、表格、欄位和索引。

與 Python 工具和框架整合

文章	描述
Apache Arrow 整合	以箭頭表格取得結果，進行零拷貝分析。
Pandas整合	將查詢結果載入 DataFrames 並寫回。
極座標積分	用 Polars 搭配 mssql-python 來處理欄位式工作負載。
DuckDB 整合	查詢 SQL Server 資料與本地 DuckDB 資料表。
FastAPI 整合	將 mssql-python 接入 FastAPI 服務。
Flask 整合	在 Flask 應用程式中使用 mssql-python。
非同步模式	將 mssql-python 與 asyncio 及執行緒集區結合。
資料存取與分析模式	選擇適合的讀取路徑，以便針對 SQL 資料進行游標存取、Arrow 資料擷取，以及使用 pandas、Polars 和 DuckDB 進行分析。
資料載入與移動模式	選擇正確的寫入路徑，用於列插入、批量複製、 MERGE upserts、DataFrame 載入和 CSV 匯入。

部署和操作

文章	描述
集裝箱與在地開發	為連接 SQL 的 Python 應用程式設置 Docker 容器、開發容器和 CI 管線。
性能調校	連線集區調校、預備陳述式、批次大小，以及大量複製。
Troubleshooting	常見錯誤、日誌記錄與憑證診斷。
模組組態	模組層級設定、日誌鉤子與功能標誌。

遷移到 mssql-python

文章	描述
從 pyodbc 移轉	將 pyodbc API 和連線字串映射到 mssql-python。
從 pymssql 遷移	用 mssql-python 取代 pymssql，同時保留行為。
從 SQLite 遷移	將本地 SQLite 工作負載移至 SQL Server 或 Azure SQL。
遷移自 PostgreSQL	Python 開發者從 PostgreSQL 轉換到 SQL Server 搭配 mssql-python 的一站式指南。

參考

文章	描述
支援生命週期	支援的 Python 和 SQL Server 版本，以及更新頻率。
有什麼新鮮事	版本歷史與發佈重點。

相關內容

• GitHub 上的 mssql-python
• PyPI 上的 mssql-python
• 發行說明
• 藍圖
• Azure Python 開發人員中心
• Python 資料庫 API 規範 2.0