FastAPI 시작하기

FastAPI로 빠르고 현대적인 웹 API를 만드는 방법을 배워봅시다.

FastAPI란?

FastAPI는 Python으로 API를 빠르게 개발할 수 있는 현대적인 웹 프레임워크입니다. 타입 힌트를 활용하여 자동으로 데이터 검증과 문서화를 제공합니다.

주요 특징

• 빠른 성능 - Node.js, Go와 비슷한 수준의 속도
• 자동 문서화 - Swagger UI, ReDoc 자동 생성
• 타입 안정성 - Python 타입 힌트 활용
• 비동기 지원 - async/await 완벽 지원
• 쉬운 학습 - 직관적이고 간단한 API

왜 FastAPI인가?

# Flask (전통적)
@app.route('/items/<int:item_id>')
def read_item(item_id):
    # 수동으로 검증 필요
    if item_id < 0:
        return {"error": "Invalid ID"}, 400
    return {"item_id": item_id}

# FastAPI (현대적)
@app.get("/items/{item_id}")
def read_item(item_id: int):
    # 자동 검증, 자동 문서화
    return {"item_id": item_id}

설치하기

# FastAPI 설치
pip install fastapi

# ASGI 서버 (Uvicorn)
pip install "uvicorn[standard]"

# 한 번에 설치
pip install "fastapi[all]"

첫 번째 API

가장 간단한 FastAPI 애플리케이션을 만들어봅시다.

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

서버 실행

# 기본 실행
uvicorn main:app --reload

# 포트 지정
uvicorn main:app --reload --port 8080

# 호스트 지정 (외부 접속 허용)
uvicorn main:app --reload --host 0.0.0.0

자동 문서 확인

서버를 실행한 후 다음 URL로 접속하세요:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

경로 파라미터

URL 경로에서 값을 추출할 수 있습니다.

from fastapi import FastAPI

app = FastAPI()

# 기본 경로 파라미터
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

# 여러 개의 경로 파라미터
@app.get("/users/{user_id}/items/{item_id}")
def get_user_item(user_id: int, item_id: str):
    return {"user_id": user_id, "item_id": item_id}

# Enum으로 제한
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}

    return {"model_name": model_name, "message": "Have some residuals"}

# 파일 경로 (특수 케이스)
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    return {"file_path": file_path}

경로 파라미터 검증

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(
        ...,  # 필수 파라미터
        title="Item ID",
        description="조회할 아이템의 ID",
        ge=1,  # greater than or equal (>=)
        le=1000  # less than or equal (<=)
    )
):
    return {"item_id": item_id}

쿼리 파라미터

URL의 쿼리 스트링에서 값을 받을 수 있습니다.

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 기본 쿼리 파라미터
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 선택적 파라미터
@app.get("/items/{item_id}")
def read_item(item_id: str, q: Optional[str] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

# 여러 값 받기
@app.get("/items/")
def read_items(q: Optional[list[str]] = None):
    return {"q": q}
# 요청: /items/?q=foo&q=bar
# 응답: {"q": ["foo", "bar"]}

# 불린 파라미터
@app.get("/items/{item_id}")
def read_item(item_id: str, short: bool = False):
    item = {"item_id": item_id}
    if not short:
        item.update(
            {"description": "This is an amazing item"}
        )
    return item

쿼리 파라미터 검증

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(
    q: Optional[str] = Query(
        None,
        min_length=3,
        max_length=50,
        regex="^fixedquery$",
        title="Query string",
        description="쿼리 파라미터 예시"
    ),
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100)
):
    results = {"page": page, "size": size}
    if q:
        results.update({"q": q})
    return results

Request Body

POST, PUT 요청에서 데이터를 받을 때 사용합니다.

Pydantic 모델

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# 데이터 모델 정의
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

# 검증 추가
class ItemWithValidation(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="가격은 0보다 커야 합니다")
    tax: Optional[float] = Field(None, ge=0, le=100)

@app.post("/validated-items/")
def create_validated_item(item: ItemWithValidation):
    return item

중첩된 모델

from pydantic import BaseModel, HttpUrl
from typing import Optional, List, Set

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()
    images: Optional[List[Image]] = None

@app.post("/items/")
def create_item(item: Item):
    return item

# 요청 예시:
# {
#   "name": "Foo",
#   "price": 45.2,
#   "tags": ["electronics", "computers"],
#   "images": [
#     {"url": "http://example.com/img1.jpg", "name": "Image 1"},
#     {"url": "http://example.com/img2.jpg", "name": "Image 2"}
#   ]
# }

Body + Path + Query 파라미터

@app.put("/items/{item_id}")
def update_item(
    item_id: int,  # Path 파라미터
    item: Item,  # Body
    q: Optional[str] = None  # Query 파라미터
):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

응답 모델

응답의 형식을 정의하고 불필요한 데이터를 숨길 수 있습니다.

from pydantic import BaseModel, EmailStr
from typing import Optional

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None

@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
    # 비밀번호는 응답에서 제외됨
    return user

# 응답 상태 코드 지정
from fastapi import status

@app.post("/users/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user: UserIn):
    return user

# 리스트 응답
@app.get("/users/", response_model=List[UserOut])
def read_users():
    return [
        {"username": "john", "email": "john@example.com"},
        {"username": "jane", "email": "jane@example.com"}
    ]

에러 처리

적절한 에러 응답을 반환할 수 있습니다.

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
            headers={"X-Error": "There goes my error"}
        )
    return {"item": items[item_id]}

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

@app.exception_handler(ItemNotFoundException)
def item_not_found_handler(request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"message": f"Item {exc.item_id} not found"}
    )

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise ItemNotFoundException(item_id=item_id)
    return {"item": items[item_id]}

의존성 주입

코드를 재사용하고 관심사를 분리할 수 있습니다.

from fastapi import Depends, FastAPI, HTTPException
from typing import Optional

app = FastAPI()

# 간단한 의존성
def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
    return commons

# 클래스 기반 의존성
class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
    return commons

인증 의존성

from fastapi import Depends, HTTPException, Header
from typing import Optional

def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

실전 예제

예제 1: 간단한 Todo API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI(title="Todo API", version="1.0.0")

# 데이터 모델
class TodoItem(BaseModel):
    id: Optional[int] = None
    title: str
    description: Optional[str] = None
    completed: bool = False

class TodoUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

# 임시 데이터베이스
todos: List[TodoItem] = []
todo_counter = 1

# 전체 목록 조회
@app.get("/todos", response_model=List[TodoItem])
def get_todos(skip: int = 0, limit: int = 10):
    return todos[skip : skip + limit]

# 특정 항목 조회
@app.get("/todos/{todo_id}", response_model=TodoItem)
def get_todo(todo_id: int):
    for todo in todos:
        if todo.id == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")

# 새 항목 생성
@app.post("/todos", response_model=TodoItem, status_code=201)
def create_todo(todo: TodoItem):
    global todo_counter
    todo.id = todo_counter
    todo_counter += 1
    todos.append(todo)
    return todo

# 항목 수정
@app.put("/todos/{todo_id}", response_model=TodoItem)
def update_todo(todo_id: int, todo_update: TodoUpdate):
    for todo in todos:
        if todo.id == todo_id:
            if todo_update.title is not None:
                todo.title = todo_update.title
            if todo_update.description is not None:
                todo.description = todo_update.description
            if todo_update.completed is not None:
                todo.completed = todo_update.completed
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")

# 항목 삭제
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int):
    for i, todo in enumerate(todos):
        if todo.id == todo_id:
            todos.pop(i)
            return
    raise HTTPException(status_code=404, detail="Todo not found")

# 통계
@app.get("/todos/stats/summary")
def get_stats():
    total = len(todos)
    completed = sum(1 for todo in todos if todo.completed)
    pending = total - completed
    return {
        "total": total,
        "completed": completed,
        "pending": pending
    }

예제 2: 사용자 관리 API

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
import hashlib

app = FastAPI()

# 모델
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    full_name: Optional[str] = None
    password: Optional[str] = None

class User(UserBase):
    id: int
    is_active: bool = True

# 임시 데이터베이스
users_db: List[dict] = []
user_counter = 1

# 비밀번호 해싱 (실제로는 bcrypt 사용 권장)
def hash_password(password: str) -> str:
    return hashlib.sha256(password.encode()).hexdigest()

# 사용자 생성
@app.post("/users/", response_model=User, status_code=201)
def create_user(user: UserCreate):
    global user_counter

    # 중복 확인
    if any(u["username"] == user.username for u in users_db):
        raise HTTPException(status_code=400, detail="Username already exists")

    if any(u["email"] == user.email for u in users_db):
        raise HTTPException(status_code=400, detail="Email already exists")

    # 사용자 생성
    user_dict = user.dict()
    user_dict["id"] = user_counter
    user_dict["is_active"] = True
    user_dict["password"] = hash_password(user.password)

    users_db.append(user_dict)
    user_counter += 1

    # 비밀번호 제외하고 반환
    return {k: v for k, v in user_dict.items() if k != "password"}

# 사용자 목록
@app.get("/users/", response_model=List[User])
def list_users(skip: int = 0, limit: int = 10):
    return [
        {k: v for k, v in user.items() if k != "password"}
        for user in users_db[skip : skip + limit]
    ]

# 사용자 조회
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    for user in users_db:
        if user["id"] == user_id:
            return {k: v for k, v in user.items() if k != "password"}
    raise HTTPException(status_code=404, detail="User not found")

# 사용자 수정
@app.patch("/users/{user_id}", response_model=User)
def update_user(user_id: int, user_update: UserUpdate):
    for user in users_db:
        if user["id"] == user_id:
            if user_update.email is not None:
                user["email"] = user_update.email
            if user_update.full_name is not None:
                user["full_name"] = user_update.full_name
            if user_update.password is not None:
                user["password"] = hash_password(user_update.password)
            return {k: v for k, v in user.items() if k != "password"}
    raise HTTPException(status_code=404, detail="User not found")

# 사용자 삭제
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
    for i, user in enumerate(users_db):
        if user["id"] == user_id:
            users_db.pop(i)
            return
    raise HTTPException(status_code=404, detail="User not found")

예제 3: 파일 업로드

from fastapi import FastAPI, File, UploadFile, HTTPException
from typing import List
import shutil
from pathlib import Path

app = FastAPI()

# 업로드 디렉토리
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

# 단일 파일 업로드
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    try:
        # 파일 저장
        file_path = UPLOAD_DIR / file.filename
        with file_path.open("wb") as buffer:
            shutil.copyfileobj(file.file, buffer)

        return {
            "filename": file.filename,
            "content_type": file.content_type,
            "size": file_path.stat().st_size
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        file.file.close()

# 여러 파일 업로드
@app.post("/upload-multiple/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
    uploaded_files = []

    for file in files:
        try:
            file_path = UPLOAD_DIR / file.filename
            with file_path.open("wb") as buffer:
                shutil.copyfileobj(file.file, buffer)

            uploaded_files.append({
                "filename": file.filename,
                "size": file_path.stat().st_size
            })
        finally:
            file.file.close()

    return {"uploaded_files": uploaded_files}

# 파일 목록
@app.get("/files/")
def list_files():
    files = []
    for file_path in UPLOAD_DIR.iterdir():
        if file_path.is_file():
            files.append({
                "filename": file_path.name,
                "size": file_path.stat().st_size
            })
    return {"files": files}

예제 4: 백그라운드 작업

from fastapi import BackgroundTasks, FastAPI
import time

app = FastAPI()

def write_log(message: str):
    """시간이 오래 걸리는 작업"""
    time.sleep(5)
    with open("log.txt", "a") as log:
        log.write(f"{message}\n")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 백그라운드에서 실행
    background_tasks.add_task(write_log, f"Notification sent to {email}")

    # 즉시 응답
    return {"message": "Notification sent in background"}

자주 묻는 질문

Q1. FastAPI vs Flask, 어떤 것을 선택해야 하나요?

A: 다음 기준으로 선택하세요:

FastAPI 선택:

• 새 프로젝트 시작
• API 위주의 개발
• 타입 안정성 필요
• 자동 문서화 필요
• 고성능 필요

Flask 선택:

• 기존 프로젝트 유지보수
• 간단한 웹 애플리케이션
• 풍부한 플러그인 생태계 필요

Q2. async/await를 꼭 사용해야 하나요?

A: 선택사항입니다. 일반 함수도 작동합니다.

# 동기 (CPU 집약적 작업에 적합)
@app.get("/sync")
def sync_endpoint():
    return {"message": "Sync"}

# 비동기 (I/O 집약적 작업에 적합)
@app.get("/async")
async def async_endpoint():
    return {"message": "Async"}

Q3. 데이터베이스는 어떻게 연결하나요?

A: SQLAlchemy나 Tortoise ORM을 사용할 수 있습니다.

# SQLAlchemy 예시
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

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

@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

Q4. CORS는 어떻게 설정하나요?

A: CORSMiddleware를 사용합니다.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 프로덕션에서는 특정 도메인만
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Q5. 환경 변수는 어떻게 관리하나요?

A: pydantic의 BaseSettings를 사용합니다.

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()

다음 단계

FastAPI 기초를 마스터했다면, 다음 주제를 학습해보세요:

1. REST API 만들기 - 완전한 RESTful API 구현하기
2. 데이터베이스 - SQLAlchemy로 데이터베이스 연결하기
3. 인증 - JWT, OAuth2 구현하기
4. 배포 - Docker, Kubernetes로 프로덕션 배포하기

참고 자료

• FastAPI 공식 문서
• Pydantic 문서
• Uvicorn 문서
• HTTP 상태 코드