FastAPI는 API를 구축하기 위한 현대적인 Python 웹 프레임워크입니다. mssql-python과 결합하면 Microsoft SQL과 Azure SQL Database를 기반으로 한 고성능 REST API를 구축할 수 있습니다.
사전 요구 사항
- Python 3.10 이상.
- 일회성 운영 체제 관련 필수 구성 요소를 설치합니다. 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
프로젝트 구조
데이터베이스, 스키마, 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"
)
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
메모
ActiveDirectoryDefault는 여러 자격 증명 제공자를 순서대로 시도하는 DefaultAzureCredential를 사용합니다. 첫 번째 연결은 SDK가 작동하는 공급자를 찾을 때까지 체인을 따라 걸어 다니기 때문에 느릴 수 있습니다. 운영 환경에서는 어떤 자격 증명 유형을 사용하는지 알면, 체인 워크를 피하기 위해 직접 지정하세요(예: ActiveDirectoryMSI 관리 신원). 자세한 내용은 Microsoft Entra 인증을 참조하세요.
피단틱 모델
피단틱 모델은 요청 및 응답 데이터의 형태와 검증 규칙을 정의합니다. 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:
raise HTTPException(status_code=503, detail="Database unavailable")
애플리케이션 실행
uvicorn main:app --reload --host 0.0.0.0 --port 8000
서버는 http://localhost:8000에서 수신 대기합니다. API를 사용하는 동안 이 터미널을 계속 작동시키세요.
API 활용
브라우저에서 엽니다 http://localhost:8000/docs . FastAPI는 모든 경로에 대해 대화형 문서를 표시합니다.
-
GET /health를 펼친 후 'Try it out'을 선택한 다음, '실행'을 선택하세요. 응답에 상태 코드
200가 있고 데이터베이스 연결이 정상인지 확인하세요. -
GET /products를 펼치고 Try it out을 선택한 다음,
5를page_size(으)로 설정한 후 Execute를 선택하세요. 답변에는 다섯 가지 제품과 페이지 정보가 포함되어 있습니다. - 응답에서 값을 복사하세요
id. GET /products/{product_id}를 펼치고, 'Try it out'을 선택한 뒤, 복사한 값을product_id입력한 후 '실행'을 선택하세요. -
GET /products/search/를 펼치고, Try it out을 선택한 뒤, for
q와 같은bike검색어를 입력한 후 Execute를 선택하세요.
애플리케이션 테스트 및 배포
오류 처리, 연결 풀링, 인증, 테스트 및 배포에 관한 지침은 mssql-python을 이용한 FastAPI 애플리케이션 테스트 및 배포를 참조하세요.