FastAPI는 API를 구축하기 위한 현대적인 Python 웹 프레임워크입니다. mssql-python과 결합하면 Microsoft SQL과 Azure SQL Database를 기반으로 한 고성능 REST API를 구축할 수 있습니다.
사전 요구 사항
- Python 3.10 이상.
-
mssql-pythonfastapiuvicornpydantic, , 그리고PyJWT패키지들.pip install fastapi uvicorn mssql-python pydantic pyjwt로 모두 설치하세요. - 일회성 운영 체제 관련 필수 구성 요소를 설치합니다. Windows 사용자는 이 단계를 건너뛸 수 있습니다. 전체 플랫폼 세부사항은 Install mssql-python을 참조하세요.
SQL 데이터베이스 만들기
다음 플랫폼 중 하나에서 SQL 데이터베이스를 생성하거나 연결하세요:
이 글의 예시들은 AdventureWorksLT 샘플 데이터베이스, 특히 표를 SalesLT.Product 사용합니다. AdventureWorksLT가 설치되어 있지 않다면, AdventureWorks 샘플 데이터베이스를 참고하세요.
프로젝트 설정
가상 환경 만들기
이 프로젝트의 패키지가 다른 Python 설치와 격리되도록 가상 환경을 만들고 활성화하세요. 이 단계는 또한 한 인터프리터에 패키지를 설치하는 문제도 방지하며, 앱을 실행하거나 다른 인터프리터로 테스트를 실행하는 문제를 예방합니다.
py -m venv .venv
.\.venv\Scripts\Activate.ps1
환경을 활성화한 후, python, pip, pytest 모두 동일한 인터프리터로 해석됩니다. 이 글의 나머지 명령어들을 활성화된 환경에서 실행해 보세요.
비고
Windows on Arm에서는 Arm64 빌드인 Python so mssql-python 빌드로 환경을 만들고, 프리빌드 휠에서 그 의존성을 설치하세요. 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로 wheel 패키지 버전을 설치한 다음 나머지를 설치하세요.
종속성 설치
필요한 패키지를 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"
)
비고
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로 주입됨)를 받아 SQL 주입을 방지하기 위해 매개변수화된 쿼리 (%(name)s 값 사전이 포함된 자리 표시자)를 사용하여 하나의 연산을 처리합니다. 이러한 분리는 비즈니스 로직을 테스트하고 재사용하기를 더 쉽게 만듭니다.
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를 catch하면 FastAPI는 일반적인 HTTP 500 응답 대신 적절한 HTTP 상태 코드가 포함된 구조화된 JSON 오류를 반환합니다.
글로벌 예외 핸들러
이 핸들러들을 main.py에, app = FastAPI(...) 줄 바로 뒤에 추가하세요. 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"}
)
비고
다른 행들이 여전히 참조하고 있는 제품을 삭제하면 외래 키 제약 조건으로 인해 mssql_python.IntegrityError 오류가 발생하고, 핸들러는 해당 행을 제거하는 대신 400 응답을 반환합니다. AdventureWorksLT 샘플에서는 SalesLT.Product의 대부분의 제품이 SalesLT.SalesOrderDetail에서 참조되므로, 해당 제품들에 대해서는 DELETE가 설계상 실패합니다. 성공적인 삭제를 테스트하려면 해당 제품을 POST /products 생성하고 삭제하거나, 먼저 참조 행을 제거하세요.
연결 풀링 (Connection Pooling)
연결 풀링이 없으면 각 요청이 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는 실제 HTTP 서버를 시작하지 않고도 애플리케이션에 요청을 보내는, httpx를 기반으로 구축된 TestClient을 제공합니다. 경로, 상태 코드, 응답 형태를 검증하는 테스트를 pytest 작성하세요.
이 섹션의 테스트를 실행하기 전에 테스트 의존성을 설치하세요:
pip install pytest httpx
비고
최신 버전의 Starlette를 사용 중이거나 새 환경을 설정하는 경우, httpx보다 httpx2를 사용하는 것이 좋습니다. 최근 Starlette 버전은 TestClient에 httpx2를 사용하며, 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"
프로젝트 루트, 즉 main.py와 같은 디렉터리에서 pytest를 사용해 테스트를 실행하세요:
pytest
이 테스트는 모의 객체가 아니라 실제 데이터베이스를 대상으로 실행되므로, test_create_productSalesLT.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"
)
그런 다음 자체 사본을 정의하는 대신 config에서 get_connection_string을 가져오도록 database.py를 업데이트하세요. 중복된 기능을 제거하면 앱이 단일 출처에서 연결 설정을 읽도록 할 수 있습니다.
# database.py
from config import get_connection_string