使用 mssql-python 測試並部署 FastAPI 應用程式

當你用 mssql-python 建置 FastAPI 應用程式後，先設定它以部署、連線重用、錯誤處理、認證和自動化測試。

先決條件

完整使用 mssql-python 搭配 FastAPI，或使用類似的 FastAPI 應用程式，使用 AdventureWorksLT 範例資料庫。本文中的驗證相依性會查詢 SalesLT.Customer。
安裝正式環境與測試所需的相依套件：
```
pip install pydantic-settings pyjwt pytest httpx
```

設定部署設定

使用 Pydantic Settings 從環境變數載入部署專屬值。此方法將秘密排除在原始碼之外，並為每個環境提供專屬的資料庫、池與認證設定。

創建 config.py：

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_server: str
    database_name: str
    pool_size: int = 20
    pool_idle_timeout: int = 300
    jwt_secret: str


settings = Settings()


def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

在部署環境中設定 DATABASE_SERVER、DATABASE_NAME 和 JWT_SECRET。 Pydantic Settings 會自動讀取大寫的環境變數名稱。

Note

ActiveDirectoryDefault 會依序嘗試多個憑證提供者。在生產環境中，請為已部署的身分識別指定驗證模式，例如針對受控識別指定 ActiveDirectoryMSI，以避免逐一檢查認證鏈。關於可用模式，請參見 Microsoft Entra 與 mssql-python 的認證。

設定連線集區

MSSQL-Python 預設啟用連線池。在應用程式建立第一個連線前，先設定一次池子。根據應用程式預期的並行資料庫工作負載和資料庫服務層級來調整集區大小。

更新 database.py 以使用部署設定：

from collections.abc import Generator

import mssql_python

from config import get_connection_string, settings


mssql_python.pooling(
    max_size=settings.pool_size,
    idle_timeout=settings.pool_idle_timeout,
)


def get_db_dependency() -> Generator:
    with mssql_python.connect(get_connection_string()) as conn:
        with conn.cursor() as cursor:
            yield cursor

連線上下文管理器在成功處理請求後提交，當請求處理提出例外時回滾，並關閉連線。關閉連線後，該訊號會回到泳池。關於池金鑰、大小、身份隔離與耗盡指引，請參見 mssql-python 的連線池化。

處理資料庫錯誤

註冊例外處理器，使資料庫失敗時能回傳一致的回應，且不會暴露連線細節、查詢或伺服器錯誤文字。

在 main.py 中於 app = FastAPI(...) 之後加入處理常式：

import mssql_python
from fastapi import Request
from fastapi.responses import JSONResponse


@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(
    request: Request,
    exc: mssql_python.IntegrityError,
):
    return JSONResponse(
        status_code=409,
        content={
            "detail": "The request conflicts with existing data.",
            "type": "integrity_error",
        },
    )


@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(
    request: Request,
    exc: mssql_python.DatabaseError,
):
    return JSONResponse(
        status_code=500,
        content={
            "detail": "A database operation failed.",
            "type": "database_error",
        },
    )

在回傳回應前，請先將該例外記錄到應用程式的受保護遙測管線。關於例外階層與 SQLSTATE 處理，請參閱 mssql-python 的錯誤處理與 SQLSTATE 程式碼。

新增驗證相依性

透過 FastAPI 相依串聯來驗證 JSON Web Token （JWT）、載入匹配的 AdventureWorksLT 客戶，並讓該客戶對受保護路由開放。在取得資料庫連線前，請先驗證權杖，以免無效的權杖占用連線池中的連線。

創建 auth.py：

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from config import settings
from database import get_db_dependency


security = HTTPBearer()


def get_customer_id(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> int:
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.jwt_secret,
            algorithms=["HS256"],
        )
        customer_id = int(payload["sub"])
    except (KeyError, TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    return customer_id


def get_current_customer(
    customer_id: int = Depends(get_customer_id),
    cursor = Depends(get_db_dependency),
):
    cursor.execute(
        """
        SELECT CustomerID, FirstName, LastName
        FROM SalesLT.Customer
        WHERE CustomerID = %(id)s
        """,
        {"id": customer_id},
    )
    customer = cursor.fetchone()
    if customer is None:
        raise HTTPException(status_code=401, detail="Customer not found")

    return {
        "id": customer.CustomerID,
        "first_name": customer.FirstName,
        "last_name": customer.LastName,
    }

匯入依賴並新增受保護路由至 main.py：

from auth import get_current_customer


@app.get("/me")
def get_me(current_customer: dict = Depends(get_current_customer)):
    return current_customer

使用身份提供者來發行並輪換簽署金鑰。對於 HS256，將 JWT_SECRET 設為至少 32 個隨機位元組。不要在倉庫或映像檔中儲存生產簽章秘密。

測試應用程式

FastAPI TestClient 在不啟動 HTTP 伺服器的情況下，向應用程式發送請求。以下整合測試使用已設定的資料庫。

創建 test_api.py：

import uuid

from fastapi.testclient import TestClient

from main import app


client = TestClient(app)


def test_list_products():
    response = client.get("/products")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data


def test_create_product():
    suffix = uuid.uuid4().hex[:8]
    response = client.post(
        "/products",
        json={
            "name": f"Test Product {suffix}",
            "product_number": f"TEST-{suffix}",
            "price": 19.99,
            "color": "Red",
            "size": "M",
            "category_id": 1,
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["product_number"] == f"TEST-{suffix}"
    assert data["price"] == 19.99


def test_get_product_not_found():
    response = client.get("/products/99999")
    assert response.status_code == 404


def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

從專案根執行測試：

pytest

這些測試使用已設定的資料庫，並在 test_create_product 中插入一列。SalesLT.Product 使用專用的測試資料庫，並在測試執行間重置其資料。

部署檢查清單

透過部署平台的秘密和組態儲存區來設定 DATABASE_SERVER、DATABASE_NAME 和 JWT_SECRET。
使用專用的 Microsoft Entra 身份，並取得最低所需的資料庫權限。
將池大小設於資料庫的連線限制以下，並保留管理存取及其他工作負載容量。
對一個獨立的測試資料庫執行資料庫整合測試。
配置受保護的遙測資料以應對資料庫例外、請求延遲及池池耗盡。
在部署環境中執行 Uvicorn 時，不使用 --reload。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-09-23