不用密碼連 Azure SQL:用 Token 驗證搭配 fast_executemany 解決效能瓶頸

前言

在將本地資料同步到 Azure SQL Database 的專案中,我遇到了兩個問題:

  1. 安全性:連線字串裡不想寫死帳號密碼
  2. 效能:用 pyodbc 逐筆 UPDATE 速度慢到不行

這篇文章紀錄我怎麼用 Azure Managed Identity + Access Token 取代密碼驗證,並透過 fast_executemany 大幅提升批次寫入效能。同時也整理了幾個常見疑問:有沒有 Token 差在哪、ActiveDirectoryMsi 與手動 Token 有何不同、兩者效能是否有差。


環境

  • Python 3.11+
  • pyodbc + ODBC Driver 18 for SQL Server
  • azure-identity
  • Azure SQL Database(已啟用 Entra ID / Managed Identity)
pip install pyodbc azure-identity

問題一:不想在程式裡放密碼

有無 Access Token 的差別

傳統帳密驗證的連線字串長這樣:

# 傳統帳密驗證
conn_str = (
    "Driver={ODBC Driver 18 for SQL Server};"
    "Server=your-server.database.windows.net;"
    "Database=your-db;"
    "UID=myuser;"
    "PWD=mypassword;"
)
conn = pyodbc.connect(conn_str)

Access Token 驗證則是拿掉帳密,改用 attrs_before 注入 Token:

# Access Token 驗證
conn_str = (
    "Driver={ODBC Driver 18 for SQL Server};"
    "Server=your-server.database.windows.net;"
    "Database=your-db;"
    "Encrypt=yes;"
    # 沒有 UID / PWD
)
conn = pyodbc.connect(conn_str, attrs_before={1256: token_struct})

兩者的核心差異:

帳密Access Token
驗證來源SQL Server 本地帳號Azure Entra ID (AAD)
密碼管理要自己管、要輪換由 Azure 管,短效 Token 自動刷新
程式碼安全密碼容易外洩無密碼,較安全
CI/CD / Managed Identity需要把密碼塞進環境變數直接用 Managed Identity,不需任何密碼
本機開發帳密寫死或讀 .envaz login 後自動生效

一句話總結:帳密是「你知道什麼」,Token 是「你是誰(身份)」,在雲端環境推薦用 Token,省去密碼管理的麻煩。


解法:用 Azure Identity 取得 Access Token

azure-identity 套件提供的 DefaultAzureCredential 可以依序嘗試多種驗證方式(環境變數、Managed Identity、Azure CLI 登入等),在本機開發和正式環境都能無縫切換。

import struct
import pyodbc
from azure.identity import DefaultAzureCredential

# 在程式啟動時初始化一次,讓 Token 可以被快取重複使用
azure_credential = DefaultAzureCredential()

def get_mssql_connection(server, database, driver):
    # 每次建立連線時呼叫 get_token(),有快取就回快取,快到期會自動刷新
    token_object = azure_credential.get_token("https://database.windows.net/.default")

    # pyodbc 要求 Token 必須以特定格式打包
    token_bytes = token_object.token.encode("utf-16-le")
    token_struct = struct.pack(f"<I{len(token_bytes)}s", len(token_bytes), token_bytes)

    # 連線字串不帶 UID/PWD,改用 attrs_before 注入 Token
    # SQL_COPT_SS_ACCESS_TOKEN = 1256
    conn_str = f"Driver={driver};Server={server};Database={database};Encrypt=yes;TrustServerCertificate=no;"
    conn = pyodbc.connect(conn_str, attrs_before={1256: token_struct})
    return conn

關鍵細節:

  • Token 必須先用 utf-16-le 編碼,再用 struct.pack 包成 <I{n}s 格式
  • attrs_before={1256: token_struct} 是 pyodbc 注入 SQL_COPT_SS_ACCESS_TOKEN 的方式
  • 連線字串裡不能Authentication=...,否則會與 Token 驗證衝突

補充:Authentication=ActiveDirectoryMsi vs 手動 Access Token

Azure SQL 還有另一種無密碼寫法:在連線字串直接加 Authentication=ActiveDirectoryMsi,讓 ODBC Driver 自己去拿 Token:

# ActiveDirectoryMsi 寫法:Driver 自己處理 Token
conn_str = (
    "Driver={ODBC Driver 18 for SQL Server};"
    "Server=your-server.database.windows.net;"
    "Database=your-db;"
    "Authentication=ActiveDirectoryMsi;"
    "Encrypt=yes;"
)
conn = pyodbc.connect(conn_str)

兩種方式的比較:

ActiveDirectoryMsi手動 Access Token
程式碼複雜度簡單,Driver 全包需要自己打包 struct
套件依賴只需 pyodbc需要 azure-identity
Token 快取Driver 自己管,不透明自己控制(全域 credential 物件)
彈性僅限 MSI可用任何 Credential(CLI、SP、環境變數…)
本機開發❌ 只能在 Azure 環境跑az login 本機也能用
User-Assigned MSI需加 UID=client-idManagedIdentityCredential(client_id=...)

效能有差嗎?

連線建立後的 SQL 執行效能:完全一樣。 差別只在取 Token 的開銷,且僅影響建立連線那一刻:

ActiveDirectoryMsi手動 DefaultAzureCredential
誰去打 IMDSODBC Driver(每次 connect() 都可能觸發)azure-identity(有記憶體快取)
Token 快取Driver 內部管理,不透明,無法控制全域 credential 物件自動快取,明確可控
快取失效前重連可能再打一次 IMDS直接回傳快取 Token,不打網路

該選哪個?

實務上幾乎不用思考,直接選手動管理就對了——只要記住一個原則:

每次 connect() 都重新呼叫 get_token(),不要把打包好的 token_struct 存起來重複用。

# ❌ 危險:token_struct 是靜態 bytes,1 小時後 Token 過期,重連時會爆
token_struct = pack_token(azure_credential.get_token(...))
# ... 之後某處重新 connect() 但沿用舊的 token_struct

# ✅ 正確:每次建立連線時才呼叫 get_token()
def get_conn():
    token = azure_credential.get_token(...)  # 有快取就回快取,快到期自動刷新
    return pyodbc.connect(conn_str, attrs_before={1256: pack_token(token)})

azure-identityget_token() 本身有快取,不用擔心每次都打網路,讓它去判斷就好。


問題二:逐筆 UPDATE 太慢

初版程式是這樣跑的:

for row in rows:
    cursor.execute("UPDATE users SET name = ?, email = ? WHERE account = ?", row)
conn.commit()

幾百筆資料就要花好幾秒,每一筆都是一次來回的 network round-trip。

解法:executemany + fast_executemany = True

cursor = conn.cursor()

# 啟用 fast_executemany:改用 ODBC 的批次參數傳輸,大幅減少 round-trip
cursor.fast_executemany = True

sql = "UPDATE users SET name = ?, email = ? WHERE account = ?"
cursor.executemany(sql, rows)  # rows 是 list of tuples

conn.commit()

fast_executemany 是 pyodbc 的一個優化開關,啟用後會將所有參數打包成一個批次送出,而不是一筆一筆傳送,對幾百到幾千筆的批次操作效果非常明顯。


完整範例

import sqlite3
import pyodbc
import os
import struct
from azure.identity import DefaultAzureCredential

SERVER   = "your-server.database.windows.net"
DATABASE = "your-database"
DRIVER   = "{ODBC Driver 18 for SQL Server}"

# 全域初始化一次,Token 快取由 azure-identity 管理
azure_credential = DefaultAzureCredential()


def get_mssql_connection():
    # 每次連線都呼叫 get_token(),讓 azure-identity 決定要不要刷新
    token_object = azure_credential.get_token("https://database.windows.net/.default")
    token_bytes  = token_object.token.encode("utf-16-le")
    token_struct = struct.pack(f"<I{len(token_bytes)}s", len(token_bytes), token_bytes)
    conn_str = (
        f"Driver={DRIVER};Server={SERVER};Database={DATABASE};"
        "Encrypt=yes;TrustServerCertificate=no;"
    )
    return pyodbc.connect(conn_str, attrs_before={1256: token_struct})


def sync_sqlite_to_mssql(sqlite_path: str):
    if not os.path.exists(sqlite_path):
        print(f"SQLite not found: {sqlite_path}")
        return

    with sqlite3.connect(sqlite_path) as sqlite_conn:
        rows = sqlite_conn.execute(
            "SELECT name, email, account FROM employee WHERE status = 'active'"
        ).fetchall()

    if not rows:
        print("No rows to sync.")
        return

    print(f"Syncing {len(rows)} rows...")

    with get_mssql_connection() as mssql_conn:
        cursor = mssql_conn.cursor()
        cursor.fast_executemany = True
        cursor.executemany(
            "UPDATE users SET name = ?, email = ? WHERE account = ?",
            rows,
        )
        mssql_conn.commit()

    print("Done.")


if __name__ == "__main__":
    sync_sqlite_to_mssql("portal.db")

前置設定:讓 Managed Identity 能存取 Azure SQL

在 Azure SQL 裡執行以下 T-SQL,把你的 Managed Identity 或 Entra ID 帳號加入:

-- 使用 Azure AD 帳號登入後執行
CREATE USER [your-managed-identity-name] FROM EXTERNAL PROVIDER;
ALTER ROLE db_datareader ADD MEMBER [your-managed-identity-name];
ALTER ROLE db_datawriter ADD MEMBER [your-managed-identity-name];

本機開發時,用 Azure CLI 登入後 DefaultAzureCredential 就會自動使用你的帳號:

az login

小結

問題解法
連線字串含密碼DefaultAzureCredential + Access Token (SQL_COPT_SS_ACCESS_TOKEN)
逐筆 UPDATE 慢cursor.fast_executemany = True + executemany()
本機 / 正式環境切換麻煩DefaultAzureCredential 自動偵測驗證來源
ActiveDirectoryMsi vs 手動 Token手動管理更靈活,本機開發也能用,直接選它
Token 過期問題每次 connect() 都呼叫 get_token(),不要快取 token_struct

兩個改動加起來不超過 10 行,卻同時解決了安全性與效能問題,值得加進你的 Azure SQL 工具箱。

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *