使用 mssql-python 搭配 FastAPI

FastAPI 是一個現代化的 Python 網頁框架，用於建置 API。結合 mssql-python，你可以建立由 Microsoft SQL 和 Azure SQL Database 支援的高效能 REST API。

先決條件

Python 3.10 或更新版本。
mssql-python、fastapi、uvicorn、pydantic和PyJWT套件。使用 pip install fastapi uvicorn mssql-python pydantic pyjwt 安裝全部。

安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。完整平台細節請參見安裝 mssql-python。

apk add libtool krb5-libs krb5-dev

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

dnf install -y libtool-ltdl krb5-libs

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

zypper install -y libltdl7

brew install openssl

建立 SQL 資料庫

在以下平台建立或連接 SQL 資料庫：

使用 Azure 入口網站在幾分鐘內建立 SQL 資料庫

本文中的範例使用 AdventureWorksLT 範例資料庫，尤其是 SalesLT.Product 資料表。如果你還沒安裝 AdventureWorksLT，請參考 AdventureWorks 範例資料庫。

專案設定

建立虛擬環境

建立並啟用虛擬環境，讓本專案的套件與其他 Python 安裝保持隔離。這個步驟也能避免一個常見問題：套件安裝在某個直譯器中，卻用另一個直譯器來執行應用程式或測試。

Windows
Linux/macOS

py -m venv .venv
.\.venv\Scripts\Activate.ps1

python3 -m venv .venv
source .venv/bin/activate

啟動環境後，python、pip 和 pytest 都會對應到同一個直譯器。請從已啟用的環境執行本文剩餘的指令。

Note

在 Windows on Arm 上，請使用 Arm64 版本的 Python 建立環境，讓 mssql-python 及其相依套件能從預先建置的 wheel 套件安裝。在有多個 Python 版本的機器上，py -m venv可能會選擇與預期不同的版本或架構，啟用後請確認python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())"。如果 pip 嘗試從原始碼建置 cryptography （Rust 和 OpenSSL 工具鏈錯誤），先安裝一個輪子備份版本，包含 pip install --only-binary=:all: cryptography，然後再安裝其他部分。

安裝依賴項

使用 pip 安裝所需套件：

pip install fastapi uvicorn mssql-python pydantic pyjwt

專案結構

用獨立的模組來組織你的專案，分別用於資料庫、結構和 CRUD 操作：

my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── test_api.py
└── routers/
    └── products.py

資料庫連線管理

FastAPI 使用依賴注入來提供資源，例如資料庫連線以進行路由處理。本節的模式建立了一個上下文管理器，能開啟連線、產生游標，並自動處理提交/回滾/關閉。

建立 database.py

該get_connection_string()函式會根據設定值建立 ODBC 連接字串。 get_db()上下文管理器和get_db_dependency()產生器都遵循相同的模式：開啟連線、讓出游標、成功時提交、錯誤時回滾，完成後總是關閉。 FastAPI 的 Depends() 會對每個請求呼叫 get_db_dependency() 一次，並管理其生命週期。

# database.py
import mssql_python
from contextlib import contextmanager
from typing import Generator

# Configuration
DATABASE_CONFIG = {
    "server": "<server>.database.windows.net",
    "database": "<database>",
}

def get_connection_string() -> str:
    """Build connection string from config."""
    return (
        f"Server={DATABASE_CONFIG['server']};"
        f"Database={DATABASE_CONFIG['database']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Note

ActiveDirectoryDefault 使用 DefaultAzureCredential，其會依序嘗試多個憑證提供者。第一次連線可能會比較慢，因為 SDK 會一直走鏈條直到找到可用的供應商。在生產環境中，如果您知道您的環境使用哪一種憑證類型，請直接指定該憑證類型（例如，針對受控識別可指定 ActiveDirectoryMSI），以避免逐一嘗試整個憑證鏈。如需詳細資訊，請參閱 Microsoft Entra 驗證。

@contextmanager
def get_db() -> Generator:
    """Database connection context manager for FastAPI dependency injection."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

def get_db_dependency():
    """FastAPI dependency for database cursor."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

擬研模型

態態模型定義了請求與回應資料的形狀與驗證規則。 FastAPI 利用這些模型解析輸入的 JSON、驗證欄位約束，並自動產生 OpenAPI 文件。

建立 schemas.py 檔案

將結構 Base分為、 Create、 Update、和響應變體。該 Base 架構保留共享欄位， Create 並繼承其作為插入操作，並 Update 使所有欄位在部分更新時為可選。

# schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Optional
from datetime import datetime

# Product schemas
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    product_number: str = Field(..., min_length=1, max_length=25)
    price: float = Field(..., gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class ProductCreate(ProductBase):
    pass

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    product_number: Optional[str] = Field(None, min_length=1, max_length=25)
    price: Optional[float] = Field(None, gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class Product(ProductBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

# Pagination
class PaginatedResponse(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    pages: int

CRUD作業

將資料庫查詢封裝在專用類別中，以保持路由處理程序的精簡。每個靜態方法會接收一個游標（由 FastAPI 注入），並以參數化查詢（%(name)s 帶有值字典的佔位符）處理一次操作，以防止 SQL 注入。這種分離讓商業邏輯更容易測試和重用。

建立 crud.py

# crud.py
from typing import Optional, List
from schemas import ProductCreate, ProductUpdate, Product

class ProductCRUD:
    """CRUD operations for products."""
    
    @staticmethod
    def get(cursor, product_id: int) -> Optional[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE ProductID = %(id)s
        """, {"id": product_id})
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def get_all(cursor, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]
    
    @staticmethod
    def count(cursor) -> int:
        cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
        return cursor.fetchval()
    
    @staticmethod
    def create(cursor, product: ProductCreate) -> dict:
        cursor.execute("""
            INSERT INTO SalesLT.Product (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
        """, {
            "name": product.name,
            "product_number": product.product_number,
            "price": product.price,
            "color": product.color,
            "size": product.size,
            "category_id": product.category_id
        })
        
        row = cursor.fetchone()
        return {
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        }
    
    @staticmethod
    def update(cursor, product_id: int, product: ProductUpdate) -> Optional[dict]:
        # Build dynamic update
        updates = []
        params = {"id": product_id}
        
        if product.name is not None:
            updates.append("Name = %(name)s")
            params["name"] = product.name
        if product.product_number is not None:
            updates.append("ProductNumber = %(product_number)s")
            params["product_number"] = product.product_number
        if product.price is not None:
            updates.append("ListPrice = %(price)s")
            params["price"] = product.price
        if product.category_id is not None:
            updates.append("ProductCategoryID = %(category_id)s")
            params["category_id"] = product.category_id
        
        if not updates:
            return ProductCRUD.get(cursor, product_id)
        
        cursor.execute(f"""
            UPDATE SalesLT.Product SET {', '.join(updates)}
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            WHERE ProductID = %(id)s
        """, params)
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def delete(cursor, product_id: int) -> bool:
        cursor.execute("""
            DELETE FROM SalesLT.Product WHERE ProductID = %(id)s
        """, {"id": product_id})
        return cursor.rowcount > 0
    
    @staticmethod
    def search(cursor, query: str, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE Name LIKE %(query)s OR ProductNumber LIKE %(query)s
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"query": f"%{query}%", "skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]

FastAPI 應用

建立 main.py

主模組將所有東西接線連接起來。每條路由都會 cursor = Depends(get_db_dependency)宣告，告訴 FastAPI 呼叫產生器，將產生的游標傳給處理器，然後再進行清理。 FastAPI 也會在處理器執行前，先驗證請求體與你的 Pydantic 架構。

# main.py
from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List
from database import get_db_dependency
from schemas import Product, ProductCreate, ProductUpdate, PaginatedResponse
from crud import ProductCRUD

app = FastAPI(
    title="Product API",
    description="REST API for products using mssql-python",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Product API", "docs": "/docs"}

@app.get("/products", response_model=PaginatedResponse)
def list_products(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """List all products with pagination."""
    skip = (page - 1) * page_size
    items = ProductCRUD.get_all(cursor, skip=skip, limit=page_size)
    total = ProductCRUD.count(cursor)
    
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    }

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Get a specific product by ID."""
    product = ProductCRUD.get(cursor, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate, cursor = Depends(get_db_dependency)):
    """Create a new product."""
    return ProductCRUD.create(cursor, product)

@app.put("/products/{product_id}", response_model=Product)
def update_product(
    product_id: int,
    product: ProductUpdate,
    cursor = Depends(get_db_dependency)
):
    """Update an existing product."""
    updated = ProductCRUD.update(cursor, product_id, product)
    if not updated:
        raise HTTPException(status_code=404, detail="Product not found")
    return updated

@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Delete a product."""
    if not ProductCRUD.delete(cursor, product_id):
        raise HTTPException(status_code=404, detail="Product not found")

@app.get("/products/search/", response_model=List[Product])
def search_products(
    q: str = Query(..., min_length=1),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """Search products by name or product number."""
    skip = (page - 1) * page_size
    return ProductCRUD.search(cursor, q, skip=skip, limit=page_size)

# Health check endpoint
@app.get("/health")
def health_check(cursor = Depends(get_db_dependency)):
    """Check database connectivity."""
    try:
        cursor.execute("SELECT 1")
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database unhealthy: {str(e)}")

執行應用程式

uvicorn main:app --reload --host 0.0.0.0 --port 8000

錯誤處理

FastAPI 允許你為特定例外類型註冊全域例外處理器。當你捕捉到 mssql_python.DatabaseError 和 mssql_python.IntegrityError 時，FastAPI 會回傳結構化的 JSON 錯誤以及適當的 HTTP 狀態碼，而不是泛用的 500 錯誤回應。

全域例外處理程序

在 app = FastAPI(...) 這一行後面，將這些處理常式加入至 main.py。 FastAPI 會在路由產生該異常類型時執行匹配處理器，因此不需要在每條路由中都設置 try/except 區塊。

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse
import mssql_python

@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(request: Request, exc: mssql_python.DatabaseError):
    """Handle database errors globally."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Database error occurred", "type": "database_error"}
    )

@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(request: Request, exc: mssql_python.IntegrityError):
    """Handle integrity constraint violations."""
    error_msg = str(exc)
    
    if "UNIQUE" in error_msg:
        return JSONResponse(
            status_code=409,
            content={"detail": "Resource already exists", "type": "duplicate_error"}
        )
    elif "FOREIGN KEY" in error_msg:
        return JSONResponse(
            status_code=400,
            content={"detail": "Referenced resource not found", "type": "reference_error"}
        )
    
    return JSONResponse(
        status_code=400,
        content={"detail": "Data integrity error", "type": "integrity_error"}
    )

Note

刪除仍被其他資料列參照的產品時，會因外鍵約束而引發 mssql_python.IntegrityError，而處理常式會回傳 400，而不是移除該資料列。在 AdventureWorksLT 範例中，SalesLT.Product 中的大多數產品都會由 SalesLT.SalesOrderDetail 參考，因此 DELETE 依設計會對這些產品失敗。若要測試成功刪除，請建立一個包含 POST /products 的產品，然後刪除該產品；或先移除引用它的資料列。

連線池化

若沒有連線池，每個請求都會開啟或關閉 Microsoft SQL 的 TCP 連線，這會增加延遲。連線池會讓一組閒置連線隨時待命，方便重複使用。在啟動時呼叫一次 mssql_python.pooling()。啟用池化時，在 get_db_dependency() 中，conn.close() 會將連線返回連線集區，而不是實際將其關閉。

強化型資料庫模組

啟動時呼叫 mssql_python.pooling() 以啟用集區，並使用適當的大小上限和逾時設定進行設定：

# database.py with connection pooling
import mssql_python
from contextlib import contextmanager
import os

# Configure pool
mssql_python.pooling(max_size=20, idle_timeout=300)

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

def get_db_dependency():
    """FastAPI dependency with connection pooling."""
    conn = mssql_python.connect(DATABASE_URL)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()  # Returns to pool

認證中介軟體

你可以結合資料庫存取和認證，透過連結 FastAPI 相依關係。以下範例驗證 JWT 承載標記，查詢 AdventureWorksLT 範例資料庫中的匹配人物紀錄，並將結果提供給受保護路由。

# auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    cursor = Depends(get_db_dependency)
):
    """Validate JWT and return the matching AdventureWorksLT person."""
    try:
        token = credentials.credentials
        # Replace with a strong secret loaded from environment variables
        payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
        person_id = int(payload.get("sub"))
        
        if not person_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        cursor.execute("""
            SELECT BusinessEntityID, FirstName, LastName
            FROM Person.Person
            WHERE BusinessEntityID = %(id)s
        """, {"id": person_id})
        
        person = cursor.fetchone()
        if not person:
            raise HTTPException(status_code=401, detail="User not found")
        
        return {
            "id": person.BusinessEntityID,
            "first_name": person.FirstName,
            "last_name": person.LastName
        }
        
    except (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")

# Protected endpoint
@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Testing

FastAPI 提供了一個 TestClient 內建 httpx 的程式，可以在不啟動真實 HTTP 伺服器的情況下，直接向你的應用程式發送請求。撰寫測試 pytest 以驗證路由、狀態碼和回應形狀。

在執行本節中的測試之前，請先安裝測試相依性套件：

pip install pytest httpx

Note

如果你使用的是最新版 Starlette，或正在設定新環境，請優先使用 httpx2，而非 httpx。最近的 Starlette 版本使用 httpx2 作為 TestClient，且在僅安裝 httpx 時會發出棄用警告。使用 pip install pytest httpx2 安裝。

測試設置

建立一個測試檔案，用 TestClient 來驗證路由行為與回應結構：

# test_api.py
from fastapi.testclient import TestClient
from main import app
import uuid
import pytest

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]
    name = f"Test Product {suffix}"
    product_data = {
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "color": "Red",
        "size": "M",
        "category_id": 1
    }
    response = client.post("/products", json=product_data)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == name
    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 執行測試，也就是 main.py 所在的同一個目錄：

pytest

這些測試是針對你的即時資料庫執行，而非模擬資料庫，因此test_create_product會插入一個實數列。SalesLT.Product 在 AdventureWorksLT 中，Name 和 ProductNumber 都具有唯一約束，因此每次執行測試時，都會為兩者各自產生唯一值。如果你改為將那些值硬編碼，除非先刪除那一列，否則測試在第二次執行時會因衝突而失敗。

部署組態

使用 Pydantic BaseSettings 來從環境變數和 .env 檔案載入設定。這種方法能將原始碼的祕密排除在外，並讓環境間切換變得容易。使用 pip install pydantic-settings 安裝設定套件。

環境變數

建立一個設定模組，從環境變數載入設定，讓你能管理程式碼外的秘密和部署專屬值：

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_server: str = "<server>.database.windows.net"
    database_name: str = "<database>"
    pool_size: int = 10

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

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

然後更新 database.py，改為從 config 匯入 get_connection_string，而不是自行定義一份副本。移除重複的功能後，你確保應用程式能從單一來源讀取連線設定。

# database.py
from config import get_connection_string

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22

使用 mssql-python 搭配 FastAPI

先決條件

建立 SQL 資料庫

專案設定

建立虛擬環境

安裝依賴項

專案結構

資料庫連線管理

建立 database.py

擬研模型

建立 schemas.py 檔案

CRUD作業

建立 crud.py

FastAPI 應用

建立 main.py

執行應用程式

錯誤處理

全域例外處理程序

連線池化

強化型資料庫模組

認證中介軟體

Testing

測試設置

部署組態

環境變數

相關內容

意見反應

其他資源