在用 mssql-python 构建 FastAPI 应用后，配置其部署、连接重用、错误处理、认证和自动测试。
Prerequisites
完全 使用 mssql-python 配合 FastAPI，或拥有使用 AdventureWorksLT 示例数据库的等效 FastAPI 应用。 本文中的身份验证依赖项会查询
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 会自动读取大写环境变量名称。
注释
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 令牌（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 身份，并授予其所需的最低数据库权限。
- 将池大小设在数据库连接限制以下，并保留管理访问和其他工作负载容量。
- 对一个隔离的测试数据库运行数据库集成测试。
- 配置受保护的遥测数据以应对数据库异常、请求延迟和池耗尽。
- 在部署环境中运行不带
--reload的 Uvicorn。