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は大文字の環境変数名を自動的に読み込みます。
Note
ActiveDirectoryDefault 複数の認証情報提供者を順番に試します。 本番環境では、配布されたアイデンティティの認証モード(管理IDの ActiveDirectoryMSI など)を指定し、認証チェーンを移動しないようにします。 利用可能なモードについては、Microsoft Entra with 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
接続コンテキストマネージャーはリクエスト処理が成功するとコミットし、リクエスト処理が例外を発生させるとロールバックし、接続を閉じます。 接続を閉じると、プールに戻されます。 プール キー、サイジング、ID の分離、枯渇に関するガイダンスについては、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ウェブトークン(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なしで動かしましょう。