FastAPI로 REST API 만들기 – 초보자도 쉽게 따라하는 완벽 가이드

1. 도입 – 학습 목표 및 필요성

FastAPI로 REST API 만들기는 현대 웹 개발에서 필수적인 기술입니다. FastAPI는 Python 기반의 최신 웹 프레임워크로, 빠른 성능과 직관적인 문법으로 많은 개발자들의 사랑을 받고 있습니다. 이 튜토리얼을 통해 여러분은 기본적인 CRUD 작업부터 데이터베이스 연동, 인증 시스템 구현까지 실전에서 바로 활용할 수 있는 REST API를 만드는 방법을 배우게 됩니다. FastAPI의 자동 문서화 기능과 타입 힌팅을 활용하면 개발 생산성을 크게 향상시킬 수 있으며, 비동기 처리를 통해 높은 성능의 API 서버를 구축할 수 있습니다. 본 가이드는 Python 기본 문법을 알고 있는 초보자부터 중급 개발자까지 모두에게 유용한 내용을 담고 있습니다.

2. 기본 개념 설명

FastAPI란 무엇인가?

FastAPI는 Sebastián Ramírez가 개발한 Python 웹 프레임워크로, Starlette와 Pydantic을 기반으로 합니다. 주요 특징은 다음과 같습니다:

• 빠른 성능: NodeJS나 Go와 비슷한 수준의 높은 성능을 자랑합니다
• 자동 문서화: Swagger UI와 ReDoc을 통한 자동 API 문서 생성
• 타입 검증: Pydantic을 활용한 자동 데이터 검증 및 직렬화
• 비동기 지원: async/await를 통한 비동기 처리 지원

REST API 기본 원칙

REST(Representational State Transfer)는 웹 서비스 설계를 위한 아키텍처 스타일입니다. 주요 원칙은:

• 무상태성(Stateless): 각 요청은 독립적이며 서버는 클라이언트 상태를 저장하지 않습니다
• 자원 기반: URL로 자원을 식별하고, HTTP 메서드(GET, POST, PUT, DELETE)로 작업을 정의합니다
• 표현: JSON, XML 등의 형식으로 자원을 표현합니다

3. 단계별 구현 가이드

Step 1: 개발 환경 설정

먼저 Python 가상환경을 생성하고 필요한 패키지를 설치합니다:

# 가상환경 생성
python -m venv venv

# 가상환경 활성화 (Windows)
venv\Scripts\activate

# 가상환경 활성화 (Mac/Linux)
source venv/bin/activate

# FastAPI 및 필요 패키지 설치
pip install fastapi uvicorn[standard] pydantic sqlalchemy

Step 2: 기본 프로젝트 구조 설정

프로젝트를 체계적으로 관리하기 위해 다음과 같은 구조를 만듭니다:

my_api/
├── main.py
├── models.py
├── schemas.py
├── database.py
└── routers/
    ├── __init__.py
    └── items.py

Step 3: 데이터 모델 정의

Pydantic을 사용하여 데이터 검증 모델을 정의합니다. schemas.py 파일에 다음 코드를 작성합니다:

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

class ItemBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: Optional[float] = Field(None, gt=0)

class ItemResponse(ItemBase):
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True

Step 4: 데이터베이스 설정

SQLAlchemy를 사용하여 데이터베이스를 설정합니다. database.py 파일을 작성합니다:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 5: ORM 모델 정의

models.py에 데이터베이스 테이블 모델을 정의합니다:

from sqlalchemy import Column, Integer, String, Float, DateTime
from datetime import datetime
from database import Base

class Item(Base):
    __tablename__ = "items"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String, nullable=True)
    price = Column(Float, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

Step 6: API 엔드포인트 구현

routers/items.py에 CRUD 작업을 위한 엔드포인트를 구현합니다:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from database import get_db
from models import Item
from schemas import ItemCreate, ItemUpdate, ItemResponse

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/", response_model=List[ItemResponse])
def get_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(Item).offset(skip).limit(limit).all()
    return items

@router.get("/{item_id}", response_model=ItemResponse)
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@router.put("/{item_id}", response_model=ItemResponse)
def update_item(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    update_data = item.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_item, key, value)
    
    db.commit()
    db.refresh(db_item)
    return db_item

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    db.delete(db_item)
    db.commit()
    return None

4. 실제 코드 예제와 설명

메인 애플리케이션 구성

main.py 파일에서 모든 컴포넌트를 통합합니다:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from database import engine, Base
from routers import items

# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)

# FastAPI 앱 인스턴스 생성
app = FastAPI(
    title="My REST API",
    description="FastAPI로 REST API 만들기 튜토리얼",
    version="1.0.0"
)

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 라우터 등록
app.include_router(items.router)

@app.get("/")
def read_root():
    return {"message": "Welcome to FastAPI REST API"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

애플리케이션 실행

개발 서버를 실행하고 API를 테스트합니다:

# 서버 실행
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 자동 생성된 API 문서 확인
# http://localhost:8000/docs (Swagger UI)
# http://localhost:8000/redoc (ReDoc)

API 테스트 예제

Python requests 라이브러리를 사용한 테스트 예제:

import requests

BASE_URL = "http://localhost:8000"

# 아이템 생성
response = requests.post(
    f"{BASE_URL}/items/",
    json={
        "title": "노트북",
        "description": "고성능 개발용 노트북",
        "price": 1500000
    }
)
print(response.json())

# 모든 아이템 조회
response = requests.get(f"{BASE_URL}/items/")
print(response.json())

# 특정 아이템 조회
response = requests.get(f"{BASE_URL}/items/1")
print(response.json())

# 아이템 수정
response = requests.put(
    f"{BASE_URL}/items/1",
    json={"price": 1400000}
)
print(response.json())

# 아이템 삭제
response = requests.delete(f"{BASE_URL}/items/1")
print(response.status_code)

5. 고급 활용 방법

인증 및 보안

JWT 토큰 기반 인증을 구현하여 API를 보호할 수 있습니다:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

비동기 처리

async/await를 활용하여 성능을 향상시킬 수 있습니다:

from fastapi import FastAPI
import asyncio

@app.get("/async-items/")
async def get_async_items():
    # 비동기 데이터베이스 쿼리
    await asyncio.sleep(1)  # 시뮬레이션
    return {"items": ["item1", "item2"]}

백그라운드 작업

BackgroundTasks를 사용하여 비동기 작업을 처리합니다:

from fastapi import BackgroundTasks

def send_notification(email: str, message: str):
    # 이메일 전송 로직
    print(f"Sending email to {email}: {message}")

@app.post("/send-notification/")
async def create_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(send_notification, email, "Welcome!")
    return {"message": "Notification sent in background"}

6. 마무리 및 추가 학습 자료

이 튜토리얼을 통해 FastAPI로 REST API 만들기의 핵심 개념과 실전 구현 방법을 배웠습니다. FastAPI는 빠른 개발 속도와 높은 성능을 동시에 제공하는 강력한 프레임워크입니다. 이제 여러분은 데이터베이스 연동, CRUD 작업, 인증, 비동기 처리 등 실무에서 필요한 기본기를 갖추게 되었습니다.

추가 학습 자료

• 공식 문서: FastAPI 공식 문서에서 더 자세한 내용을 확인하세요
• 심화 주제: WebSocket, GraphQL 통합, 테스트 작성, 배포 방법 등을 학습하세요
• 실전 프로젝트: 블로그 API, 전자상거래 API, SNS API 등을 직접 구현해보세요
• 성능 최적화: 캐싱, 데이터베이스 인덱싱, 쿼리 최적화 기법을 익히세요

계속해서 코드를 작성하고 실험하면서 FastAPI의 강력한 기능들을 마스터해보세요. 행운을 빕니다!