REST API 만들기

FastAPI로 완전한 RESTful API를 설계하고 구현하는 방법을 배워봅시다.

REST API란?

REST (Representational State Transfer)는 웹 서비스를 설계하기 위한 아키텍처 스타일입니다. RESTful API는 HTTP 프로토콜을 활용하여 리소스를 관리합니다.

REST의 핵심 원칙

1. 클라이언트-서버 분리 - 독립적인 개발과 확장
2. 무상태성 (Stateless) - 각 요청은 독립적
3. 캐시 가능 - 응답은 캐시 가능 여부 명시
4. 계층화 시스템 - 중간 서버 사용 가능
5. 인터페이스 일관성 - 통일된 인터페이스

HTTP 메서드

메서드	용도	예시
GET	리소스 조회	GET /users
POST	리소스 생성	POST /users
PUT	리소스 전체 수정	PUT /users/1
PATCH	리소스 부분 수정	PATCH /users/1
DELETE	리소스 삭제	DELETE /users/1

HTTP 상태 코드

적절한 상태 코드를 사용하는 것이 중요합니다.

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()

# 2xx: 성공
# 200 OK - 일반적인 성공
@app.get("/items/{item_id}", status_code=status.HTTP_200_OK)
def read_item(item_id: int):
    return {"item_id": item_id}

# 201 Created - 리소스 생성 성공
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: dict):
    return item

# 204 No Content - 성공했지만 반환할 내용 없음
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
    return None

# 4xx: 클라이언트 오류
# 400 Bad Request
@app.post("/invalid/")
def invalid_request():
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={"detail": "Invalid request"}
    )

# 401 Unauthorized
@app.get("/protected/")
def protected_route():
    return JSONResponse(
        status_code=status.HTTP_401_UNAUTHORIZED,
        content={"detail": "Authentication required"}
    )

# 403 Forbidden
@app.get("/admin/")
def admin_only():
    return JSONResponse(
        status_code=status.HTTP_403_FORBIDDEN,
        content={"detail": "Access denied"}
    )

# 404 Not Found
@app.get("/items/{item_id}")
def get_item(item_id: int):
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"detail": "Item not found"}
    )

# 5xx: 서버 오류
# 500 Internal Server Error
@app.get("/error/")
def server_error():
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"}
    )

URL 설계 베스트 프랙티스

좋은 URL 설계

# ✅ 명사 사용, 복수형
GET    /users          # 모든 사용자 조회
GET    /users/123      # 특정 사용자 조회
POST   /users          # 사용자 생성
PUT    /users/123      # 사용자 전체 수정
PATCH  /users/123      # 사용자 부분 수정
DELETE /users/123      # 사용자 삭제

# ✅ 계층 구조
GET    /users/123/posts           # 사용자의 게시글 목록
GET    /users/123/posts/456       # 특정 게시글
POST   /users/123/posts           # 게시글 작성
DELETE /users/123/posts/456       # 게시글 삭제

# ✅ 필터링, 정렬, 페이징
GET /users?role=admin              # 역할별 필터
GET /users?sort=created_at&order=desc  # 정렬
GET /users?page=2&limit=10         # 페이징

# ✅ 검색
GET /users/search?q=john           # 검색

# ❌ 피해야 할 패턴
GET    /getUser/123               # 동사 사용 (X)
POST   /createUser                # 동사 사용 (X)
GET    /user                      # 단수형 (X)
GET    /users/delete/123          # 메서드를 URL에 포함 (X)

완전한 CRUD API 예제

블로그 API

from fastapi import FastAPI, HTTPException, Query, Path, status
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum

app = FastAPI(
    title="Blog API",
    description="완전한 RESTful 블로그 API",
    version="1.0.0"
)

# ===== 모델 정의 =====

class PostStatus(str, Enum):
    draft = "draft"
    published = "published"
    archived = "archived"

class PostBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1)
    status: PostStatus = PostStatus.draft
    tags: List[str] = []

class PostCreate(PostBase):
    pass

class PostUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    content: Optional[str] = Field(None, min_length=1)
    status: Optional[PostStatus] = None
    tags: Optional[List[str]] = None

class Post(PostBase):
    id: int
    author_id: int
    created_at: datetime
    updated_at: datetime
    view_count: int = 0

    class Config:
        orm_mode = True

# ===== 임시 데이터베이스 =====

posts_db = []
post_counter = 1

# ===== API 엔드포인트 =====

# 1. 게시글 목록 조회 (필터링, 정렬, 페이징)
@app.get(
    "/posts",
    response_model=List[Post],
    tags=["posts"],
    summary="게시글 목록 조회"
)
def list_posts(
    status: Optional[PostStatus] = Query(None, description="상태별 필터"),
    tag: Optional[str] = Query(None, description="태그별 필터"),
    search: Optional[str] = Query(None, description="제목 검색"),
    sort_by: str = Query("created_at", description="정렬 기준"),
    order: str = Query("desc", regex="^(asc|desc)$"),
    page: int = Query(1, ge=1, description="페이지 번호"),
    limit: int = Query(10, ge=1, le=100, description="페이지당 항목 수")
):
    """
    게시글 목록을 조회합니다.

    - **status**: 상태별 필터링
    - **tag**: 태그별 필터링
    - **search**: 제목 검색
    - **sort_by**: 정렬 기준 (created_at, updated_at, view_count)
    - **order**: 정렬 순서 (asc, desc)
    - **page**: 페이지 번호
    - **limit**: 페이지당 항목 수
    """

    # 필터링
    filtered_posts = posts_db

    if status:
        filtered_posts = [p for p in filtered_posts if p["status"] == status]

    if tag:
        filtered_posts = [p for p in filtered_posts if tag in p["tags"]]

    if search:
        filtered_posts = [
            p for p in filtered_posts
            if search.lower() in p["title"].lower()
        ]

    # 정렬
    reverse = (order == "desc")
    filtered_posts.sort(key=lambda x: x.get(sort_by, 0), reverse=reverse)

    # 페이징
    start = (page - 1) * limit
    end = start + limit

    return filtered_posts[start:end]

# 2. 특정 게시글 조회
@app.get(
    "/posts/{post_id}",
    response_model=Post,
    tags=["posts"],
    summary="게시글 조회"
)
def get_post(
    post_id: int = Path(..., description="게시글 ID", ge=1)
):
    """특정 게시글을 조회합니다."""

    for post in posts_db:
        if post["id"] == post_id:
            # 조회수 증가
            post["view_count"] += 1
            return post

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Post with id {post_id} not found"
    )

# 3. 게시글 생성
@app.post(
    "/posts",
    response_model=Post,
    status_code=status.HTTP_201_CREATED,
    tags=["posts"],
    summary="게시글 작성"
)
def create_post(post: PostCreate, author_id: int = 1):
    """새 게시글을 작성합니다."""

    global post_counter

    now = datetime.now()
    new_post = {
        "id": post_counter,
        "author_id": author_id,
        "title": post.title,
        "content": post.content,
        "status": post.status,
        "tags": post.tags,
        "created_at": now,
        "updated_at": now,
        "view_count": 0
    }

    posts_db.append(new_post)
    post_counter += 1

    return new_post

# 4. 게시글 전체 수정 (PUT)
@app.put(
    "/posts/{post_id}",
    response_model=Post,
    tags=["posts"],
    summary="게시글 전체 수정"
)
def update_post_full(post_id: int, post: PostCreate):
    """게시글을 전체 수정합니다 (모든 필드 필수)."""

    for existing_post in posts_db:
        if existing_post["id"] == post_id:
            # 전체 교체
            existing_post["title"] = post.title
            existing_post["content"] = post.content
            existing_post["status"] = post.status
            existing_post["tags"] = post.tags
            existing_post["updated_at"] = datetime.now()
            return existing_post

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Post with id {post_id} not found"
    )

# 5. 게시글 부분 수정 (PATCH)
@app.patch(
    "/posts/{post_id}",
    response_model=Post,
    tags=["posts"],
    summary="게시글 부분 수정"
)
def update_post_partial(post_id: int, post: PostUpdate):
    """게시글을 부분 수정합니다 (선택적 필드)."""

    for existing_post in posts_db:
        if existing_post["id"] == post_id:
            # 제공된 필드만 업데이트
            update_data = post.dict(exclude_unset=True)

            for field, value in update_data.items():
                existing_post[field] = value

            existing_post["updated_at"] = datetime.now()
            return existing_post

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Post with id {post_id} not found"
    )

# 6. 게시글 삭제
@app.delete(
    "/posts/{post_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["posts"],
    summary="게시글 삭제"
)
def delete_post(post_id: int):
    """게시글을 삭제합니다."""

    for i, post in enumerate(posts_db):
        if post["id"] == post_id:
            posts_db.pop(i)
            return

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Post with id {post_id} not found"
    )

# 7. 게시글 통계
@app.get(
    "/posts/stats/summary",
    tags=["posts", "stats"],
    summary="게시글 통계"
)
def get_post_stats():
    """게시글 통계를 조회합니다."""

    total = len(posts_db)
    by_status = {}
    total_views = 0

    for post in posts_db:
        status = post["status"]
        by_status[status] = by_status.get(status, 0) + 1
        total_views += post["view_count"]

    return {
        "total_posts": total,
        "by_status": by_status,
        "total_views": total_views,
        "average_views": total_views / total if total > 0 else 0
    }

# 8. 게시글 검색
@app.get(
    "/posts/search",
    response_model=List[Post],
    tags=["posts"],
    summary="게시글 검색"
)
def search_posts(
    q: str = Query(..., min_length=1, description="검색어"),
    fields: List[str] = Query(["title", "content"], description="검색 필드")
):
    """게시글을 검색합니다."""

    results = []
    q_lower = q.lower()

    for post in posts_db:
        if "title" in fields and q_lower in post["title"].lower():
            results.append(post)
        elif "content" in fields and q_lower in post["content"].lower():
            if post not in results:
                results.append(post)

    return results

데이터 검증

Pydantic으로 강력한 데이터 검증을 구현할 수 있습니다.

from pydantic import BaseModel, Field, validator, EmailStr
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(
        ...,
        min_length=3,
        max_length=50,
        regex="^[a-zA-Z0-9_]+$",
        description="영문, 숫자, 언더스코어만 가능"
    )
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)
    website: Optional[str] = None

    @validator('username')
    def username_alphanumeric(cls, v):
        """사용자명 추가 검증"""
        if not v.isalnum() and '_' not in v:
            raise ValueError('Username must be alphanumeric')
        return v

    @validator('password')
    def password_strength(cls, v):
        """비밀번호 강도 검증"""
        if not any(char.isdigit() for char in v):
            raise ValueError('Password must contain at least one digit')
        if not any(char.isupper() for char in v):
            raise ValueError('Password must contain at least one uppercase letter')
        return v

    @validator('website')
    def validate_website(cls, v):
        """웹사이트 URL 검증"""
        if v and not v.startswith(('http://', 'https://')):
            raise ValueError('Website must start with http:// or https://')
        return v

# 사용
@app.post("/users/")
def create_user(user: UserCreate):
    return user

에러 처리

일관된 에러 응답 형식을 제공합니다.

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# 커스텀 예외
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

class DuplicateItemError(Exception):
    def __init__(self, field: str, value: str):
        self.field = field
        self.value = value

# 예외 핸들러
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "NotFound",
            "message": f"Item with id {exc.item_id} not found",
            "path": str(request.url)
        }
    )

@app.exception_handler(DuplicateItemError)
async def duplicate_item_handler(request: Request, exc: DuplicateItemError):
    return JSONResponse(
        status_code=409,
        content={
            "error": "Conflict",
            "message": f"{exc.field} '{exc.value}' already exists",
            "field": exc.field
        }
    )

# 검증 에러 핸들러
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": "ValidationError",
            "message": "Request validation failed",
            "details": errors
        }
    )

# 일반 예외 핸들러
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "error": "InternalServerError",
            "message": "An unexpected error occurred",
            "path": str(request.url)
        }
    )

인증과 권한

JWT 기반 인증을 구현해봅시다.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from datetime import datetime, timedelta
import jwt

app = FastAPI()

# 설정
SECRET_KEY = "your-secret-key"  # 실제로는 환경 변수 사용
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

security = HTTPBearer()

# 모델
class User(BaseModel):
    username: str
    email: str
    role: str = "user"

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

# 임시 사용자 DB
users_db = {
    "john": {
        "username": "john",
        "email": "john@example.com",
        "password": "hashed_password",
        "role": "admin"
    }
}

# JWT 토큰 생성
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# JWT 토큰 검증
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")

        if username is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials"
            )

        return username

    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired"
        )
    except jwt.JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials"
        )

# 현재 사용자 가져오기
def get_current_user(username: str = Depends(verify_token)):
    user = users_db.get(username)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return User(**user)

# 관리자 권한 확인
def require_admin(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return current_user

# 로그인
@app.post("/auth/login", response_model=Token)
def login(username: str, password: str):
    user = users_db.get(username)
    if not user or user["password"] != password:  # 실제로는 해시 비교
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password"
        )

    access_token = create_access_token(data={"sub": username})
    return {"access_token": access_token}

# 보호된 엔드포인트
@app.get("/users/me", response_model=User)
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

# 관리자 전용 엔드포인트
@app.get("/admin/users")
def list_all_users(admin: User = Depends(require_admin)):
    return list(users_db.values())

페이징 구현

효율적인 페이징 시스템을 구현합니다.

from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import List, Generic, TypeVar
from math import ceil

app = FastAPI()

# 제네릭 페이징 응답
T = TypeVar('T')

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    size: int
    pages: int

    @classmethod
    def create(cls, items: List[T], total: int, page: int, size: int):
        return cls(
            items=items,
            total=total,
            page=page,
            size=size,
            pages=ceil(total / size) if size > 0 else 0
        )

class Item(BaseModel):
    id: int
    name: str

# 임시 데이터
items_db = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]

@app.get("/items", response_model=PaginatedResponse[Item])
def list_items(
    page: int = Query(1, ge=1, description="페이지 번호"),
    size: int = Query(10, ge=1, le=100, description="페이지 크기")
):
    total = len(items_db)
    start = (page - 1) * size
    end = start + size

    items = [Item(**item) for item in items_db[start:end]]

    return PaginatedResponse.create(
        items=items,
        total=total,
        page=page,
        size=size
    )

버전 관리

API 버전을 관리하는 방법입니다.

from fastapi import FastAPI, APIRouter

app = FastAPI()

# v1 라우터
router_v1 = APIRouter(prefix="/api/v1", tags=["v1"])

@router_v1.get("/users")
def get_users_v1():
    return {"version": "1.0", "users": []}

# v2 라우터
router_v2 = APIRouter(prefix="/api/v2", tags=["v2"])

@router_v2.get("/users")
def get_users_v2():
    return {
        "version": "2.0",
        "users": [],
        "metadata": {"total": 0}  # v2에서 추가된 기능
    }

# 라우터 등록
app.include_router(router_v1)
app.include_router(router_v2)

# 또는 헤더 기반 버전 관리
from fastapi import Header, HTTPException

@app.get("/users")
def get_users(api_version: str = Header("1.0")):
    if api_version == "1.0":
        return {"version": "1.0", "users": []}
    elif api_version == "2.0":
        return {"version": "2.0", "users": [], "metadata": {}}
    else:
        raise HTTPException(400, "Unsupported API version")

실전 예제: 전자상거래 API

from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum

app = FastAPI(title="E-Commerce API", version="1.0.0")

# ===== Enums =====

class OrderStatus(str, Enum):
    pending = "pending"
    paid = "paid"
    shipped = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"

# ===== Models =====

class Product(BaseModel):
    id: int
    name: str
    description: str
    price: float = Field(..., gt=0)
    stock: int = Field(..., ge=0)
    category: str

class CartItem(BaseModel):
    product_id: int
    quantity: int = Field(..., gt=0)

class Order(BaseModel):
    id: int
    user_id: int
    items: List[CartItem]
    total: float
    status: OrderStatus
    created_at: datetime
    updated_at: datetime

# ===== Database =====

products_db = [
    {
        "id": 1,
        "name": "노트북",
        "description": "고성능 노트북",
        "price": 1500000,
        "stock": 10,
        "category": "electronics"
    },
    {
        "id": 2,
        "name": "마우스",
        "description": "무선 마우스",
        "price": 30000,
        "stock": 50,
        "category": "electronics"
    }
]

orders_db = []
order_counter = 1

# ===== Products API =====

@app.get("/products", response_model=List[Product], tags=["products"])
def list_products(
    category: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    in_stock: bool = True
):
    """상품 목록 조회"""
    results = products_db

    if category:
        results = [p for p in results if p["category"] == category]

    if min_price:
        results = [p for p in results if p["price"] >= min_price]

    if max_price:
        results = [p for p in results if p["price"] <= max_price]

    if in_stock:
        results = [p for p in results if p["stock"] > 0]

    return results

@app.get("/products/{product_id}", response_model=Product, tags=["products"])
def get_product(product_id: int):
    """상품 상세 조회"""
    for product in products_db:
        if product["id"] == product_id:
            return product
    raise HTTPException(404, "Product not found")

# ===== Orders API =====

@app.post("/orders", response_model=Order, status_code=201, tags=["orders"])
def create_order(items: List[CartItem], user_id: int = 1):
    """주문 생성"""
    global order_counter

    # 상품 확인 및 재고 검증
    total = 0
    for item in items:
        product = next((p for p in products_db if p["id"] == item.product_id), None)

        if not product:
            raise HTTPException(404, f"Product {item.product_id} not found")

        if product["stock"] < item.quantity:
            raise HTTPException(
                400,
                f"Insufficient stock for {product['name']}"
            )

        total += product["price"] * item.quantity

    # 주문 생성
    now = datetime.now()
    order = {
        "id": order_counter,
        "user_id": user_id,
        "items": [item.dict() for item in items],
        "total": total,
        "status": OrderStatus.pending,
        "created_at": now,
        "updated_at": now
    }

    # 재고 감소
    for item in items:
        for product in products_db:
            if product["id"] == item.product_id:
                product["stock"] -= item.quantity

    orders_db.append(order)
    order_counter += 1

    return order

@app.get("/orders/{order_id}", response_model=Order, tags=["orders"])
def get_order(order_id: int):
    """주문 조회"""
    for order in orders_db:
        if order["id"] == order_id:
            return order
    raise HTTPException(404, "Order not found")

@app.patch("/orders/{order_id}/status", response_model=Order, tags=["orders"])
def update_order_status(order_id: int, status: OrderStatus):
    """주문 상태 변경"""
    for order in orders_db:
        if order["id"] == order_id:
            # 상태 전환 검증
            current = order["status"]

            # 취소된 주문은 변경 불가
            if current == OrderStatus.cancelled:
                raise HTTPException(400, "Cannot update cancelled order")

            # 배송 완료 후에는 취소 불가
            if current == OrderStatus.delivered and status == OrderStatus.cancelled:
                raise HTTPException(400, "Cannot cancel delivered order")

            order["status"] = status
            order["updated_at"] = datetime.now()
            return order

    raise HTTPException(404, "Order not found")

@app.get("/orders", response_model=List[Order], tags=["orders"])
def list_orders(
    user_id: Optional[int] = None,
    status: Optional[OrderStatus] = None
):
    """주문 목록 조회"""
    results = orders_db

    if user_id:
        results = [o for o in results if o["user_id"] == user_id]

    if status:
        results = [o for o in results if o["status"] == status]

    return results

자주 묻는 질문

Q1. PUT과 PATCH의 차이는?

A:

• PUT: 리소스 전체를 교체 (모든 필드 필수)
• PATCH: 리소스 일부만 수정 (선택적 필드)

# PUT - 모든 필드 필수
@app.put("/users/{id}")
def update_user(id: int, user: UserFull):
    # 전체 교체
    pass

# PATCH - 선택적 필드
@app.patch("/users/{id}")
def update_user_partial(id: int, user: UserPartial):
    # 제공된 필드만 업데이트
    pass

Q2. REST API에서 복잡한 쿼리는 어떻게 처리하나요?

A: POST 요청에 검색 조건을 body로 전달합니다.

class SearchQuery(BaseModel):
    filters: dict
    sort: Optional[str] = None
    page: int = 1
    size: int = 10

@app.post("/users/search")
def search_users(query: SearchQuery):
    # 복잡한 검색 로직
    pass

Q3. 대량 작업은 어떻게 처리하나요?

A: 배치 엔드포인트를 만들거나 비동기 작업을 사용합니다.

@app.post("/users/batch")
def create_users_batch(users: List[UserCreate]):
    results = []
    for user in users:
        # 각 사용자 생성
        results.append(create_user(user))
    return {"created": len(results), "items": results}

Q4. API 문서를 커스터마이즈하려면?

A: FastAPI의 메타데이터와 docstring을 활용합니다.

@app.get(
    "/items/{item_id}",
    summary="아이템 조회",
    description="ID로 특정 아이템을 조회합니다",
    response_description="조회된 아이템",
    tags=["items"],
    deprecated=False
)
def get_item(item_id: int):
    """
    아이템을 조회합니다:

    - **item_id**: 조회할 아이템의 ID
    """
    pass

Q5. 성능을 개선하려면?

A: 캐싱, 페이징, 필드 선택을 구현합니다.

# 캐싱
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache

@app.get("/items/{item_id}")
@cache(expire=60)  # 60초 캐싱
def get_item(item_id: int):
    pass

# 필드 선택
@app.get("/users")
def get_users(fields: List[str] = Query(None)):
    # 필요한 필드만 반환
    pass

다음 단계

완전한 REST API를 만들 수 있게 되었다면, 다음을 학습해보세요:

1. 데이터베이스 연동 - SQLAlchemy로 실제 DB 사용하기
2. 테스트 - pytest로 API 테스트 작성하기
3. 배포 - Docker, Kubernetes로 프로덕션 배포하기
4. 모니터링 - 로깅, 메트릭, 알림 설정하기

참고 자료

• REST API 디자인 가이드
• HTTP 상태 코드
• FastAPI 공식 문서
• OpenAPI 사양