mssql-python으로 FastAPI 애플리케이션을 구축한 후에는 배포, 연결 재사용, 오류 처리, 인증, 자동 테스트 등을 구성하세요.
Prerequisites
FastAPI에서 mssql-python 사용을 완료하거나, AdventureWorksLT 샘플 데이터베이스를 사용하는 이에 상응하는 FastAPI 애플리케이션을 준비하세요. 이 문서의 인증 종속성은
SalesLT.Customer를 쿼리합니다.프로덕션 및 테스트 의존성을 설치하세요:
pip install pydantic-settings pyjwt pytest httpx2
배포 설정 구성
Pydantic 설정을 사용해 환경 변수에서 배포 전용 값을 불러오세요. 이 방법은 소스 코드에 비밀을 숨기고 각 환경에 고유한 데이터베이스, 풀, 인증 구성을 제공합니다.
프로젝트 루트에서 config.py 옆에 database.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는 대문자 환경 변수 이름을 자동으로 읽습니다.
로컬 테스트를 위해서는 데이터베이스 자리 표시자를 교체하고 활성화된 환경에서 변수를 설정하세요. 같은 터미널에서 나중에 출발하세요 pytest .
$env:DATABASE_SERVER = "<server>.database.windows.net"
$env:DATABASE_NAME = "<database>"
$env:JWT_SECRET = python -c "import secrets; print(secrets.token_urlsafe(32))"
메모
ActiveDirectoryDefault 여러 인증 제공자를 순서대로 시도합니다. 운영 환경에서는 배포된 신원에 대해 관리 신원과 같은 ActiveDirectoryMSI 인증 모드를 지정하여 자격 증명 체인을 거치는 것을 피합니다. 사용 가능한 모드에 대해서는 mssql-python을 이용한 Microsoft Entra 인증을 참조하세요.
연결 풀링을 구성합니다
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 웹 토큰(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의 경우, 최소 32바이트의 랜덤 바이트로 설정 JWT_SECRET 하세요. 저장소나 이미지에 생산 서명 비밀을 저장하지 마세요.
애플리케이션 테스트
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,JWT_SECRET,DATABASE_NAME를 배포 플랫폼의 비밀 저장소 및 구성 저장소를 통해 설정합니다. - 최소한의 데이터베이스 권한만 부여한 전용 Microsoft Entra 아이덴티티를 사용하세요.
- 풀 크기를 데이터베이스의 연결 한도 이하로 설정하고, 관리 접근 및 기타 작업 부하를 위한 용량을 남겨두세요.
- 격리된 테스트 데이터베이스에 대해 데이터베이스 통합 테스트를 실행하세요.
- 데이터베이스 예외, 요청 지연, 풀 소진에 대비해 보호된 텔레메트리를 구성하세요.
- 배포 환경에서는
--reload없이 Uvicorn을 실행하세요.