FastAPI 是一個現代化的 Python 網頁框架，用於建置 API。 結合 mssql-python，你可以建立由 Microsoft SQL 和 Azure SQL Database 支援的高效能 REST API。
先決條件
- Python 3.10 或更新版本。
- 安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。 完整平台細節請參見 安裝 mssql-python。
建立 SQL 資料庫
在以下平台建立或連接 SQL 資料庫：
本文中的範例使用 AdventureWorksLT 範例資料庫，尤其是 SalesLT.Product 資料表。 如果你還沒安裝 AdventureWorksLT，請參考 AdventureWorks 範例資料庫。
專案設定
建立虛擬環境
建立並啟用虛擬環境，讓本專案的套件與其他 Python 安裝保持隔離。 這個步驟也能避免一個常見問題：套件安裝在某個直譯器中，卻用另一個直譯器來執行應用程式或測試。
py -m venv .venv
.\.venv\Scripts\Activate.ps1
啟動環境後，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
專案結構
用獨立的模組來組織你的專案，分別用於資料庫、結構和 CRUD 操作：
my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
└── routers/
└── products.py
資料庫連線管理
FastAPI 使用依賴注入來提供資源，例如資料庫連線以進行路由處理。 本節中的模式會開啟連線、傳回游標，並使用 mssql-python 連線內容管理器在成功時提交、發生例外時回滾，並關閉連線。
建立 database.py
該get_connection_string()函式會根據設定值建立 ODBC 連接字串。 FastAPI 的 Depends() 會對每個請求呼叫 get_db_dependency() 一次，並管理其生命週期。
# database.py
import mssql_python
from collections.abc 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 驗證。
def get_db_dependency() -> Generator:
"""FastAPI dependency for database cursor."""
with mssql_python.connect(get_connection_string()) as conn:
with conn.cursor() as cursor:
yield cursor
擬研模型
態態模型定義了請求與回應資料的形狀與驗證規則。 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:
raise HTTPException(status_code=503, detail="Database unavailable")
執行應用程式
uvicorn main:app --reload --host 0.0.0.0 --port 8000
測試並部署應用程式
請使用配套文章完成申請：
錯誤處理
配套文章涵蓋資料庫例外處理。
全域例外處理程序
請參見 處理資料庫錯誤。
連線池化
配套文章介紹連線池的配置。
強化型資料庫模組
請參見「設定連線池」。
認證中介軟體
請參見 「新增認證相依關係」。
Testing
配套文章涵蓋了整合測試。
測試設置
請參見「測試應用程式」。
部署組態
配套文章涵蓋部署配置與操作。