LM Studio를 활용하면 편하게 로컬에서 최신 언어 모델을 실행하고 자신의 애플리케이션에 연동할 수 있습니다. 이번 글에서는 Gemma 4 모델을 LM Studio로 로드한 뒤, FastAPI를 이용해 간단한 REST 엔드포인트로 서빙하는 방법을 소개해보겠습니다.

준비 사항

• 운영체제: Windows 10 또는 11 (64 비트).
• 하드웨어: AVX2를 지원하는 CPU와 16 GB RAM 이상을 권장합니다. GPU는 선택 사항이지만 있으면 속도가 빨라집니다.
• 소프트웨어: LM Studio (0.4.12 이상)와 Python 3.8 이상, FastAPI, Uvicorn.

1. LM Studio 설치와 모델 로드

1.1 LM Studio 설치

1. 다운로드: LM Studio 사이트에서 Windows용 설치 파일을 내려받습니다.
2. 설치: 실행 파일을 열어 설치를 완료합니다.

3. 실행: 설치 후 LM Studio를 실행하면 모델 탐색 화면이 나타납니다.

• 저는 자동으로 gemma-4-e4b 모델을 추천해주어 바로 설치했습니다. 추천해주지 않은 다른 모델을 사용하고자 할 때는 아래 1.2의 방법을 활용하면 됩니다.
• 저와 같이 사용하고자 하는 모델을 바로 다운로드 하신 경우에는, 1.3으로 바로 넘어갑니다.

1.2 모델 다운로드

1. 상단의 검색창에서 Gemma 4를 검색합니다. 여러 변형(E2B, E4B, 26B A4B, 31B)이 목록에 표시됩니다.
2. 시작은 가벼운 E2B 또는 E4B 모델로 해보는 것이 좋습니다.
3. 해당 모델의 Get 버튼을 클릭해 다운로드합니다. 다운로드가 완료되면 My Models 섹션에서 모델을 확인할 수 있습니다.

1.3 모델 로드와 테스트

1. My Models에서 원하는 Gemma 4 모델을 선택하고 Load를 클릭하면 모델이 메모리에 올라갑니다.
2. 로드가 완료되면 좌측 메뉴의 Chat 탭에서 모델과 대화를 나눌 수 있습니다. 간단히 “안녕하세요?”와 같은 메시지를 입력해 응답을 확인해보세요.
3. 그리고 이후 FastAPI 테스트를 위해서 Developer 탭에서 서버를 활성화니다. 기본 포트는 1234입니다. 이 서버는 OpenAI 방식과 유사한 REST API를 제공합니다.

2. FastAPI로 간단한 백엔드 구축

LM Studio의 Python SDK(lmstudio-python)를 이용하면 다른 서비스에서 로컬 모델을 쉽게 호출할 수 있습니다. 공식 개발 문서의 예시처럼 pip install lmstudio로 SDK를 설치한 뒤, 모델을 불러와 응답을 받는 코드를 작성할 수 있습니다. 이를 FastAPI와 결합해 간단한 REST 서비스로 만들겠습니다.

2.1 환경 준비

1. Python 가상환경을 생성하고 활성화합니다.

2. 필요한 패키지를 설치합니다.

    pip install lmstudio fastapi uvicorn

3. LM Studio 앱에서 모델을 로드한 상태인지 확인합니다. SDK는 이미 실행 중인 LM Studio 인스턴스에 연결합니다.

2.2 FastAPI 애플리케이션 코드

아래 코드는 FastAPI를 사용해 /chat 엔드포인트를 생성하고, 요청으로 받은 프롬프트를 LM Studio에 전달해 응답을 반환합니다. Medium 기사에서 소개한 방법을 참고했습니다.

import lmstudio as lms
from fastapi import FastAPI

app = FastAPI()

MODEL_ID = "google/gemma-4-e4b"
model = None

@app.on_event("startup")
def startup():
    global model

    # 첫 convenience API 호출 전에 먼저 설정
    lms.configure_default_client("localhost:1234")

    # 서버 확인
    if not lms.Client.is_valid_api_host("localhost:1234"):
        raise RuntimeError("LM Studio 서버가 localhost:1234 에서 실행 중이 아닙니다.")

    model = lms.llm(MODEL_ID)

@app.get("/health")
def health():
    return {"ok": True}

@app.get("/ask")
def ask(q: str):
    result = model.respond(q)
    return {"answer": str(result)}

위 코드에서 google/gemma-4-e4b는 LM Studio에서 다운로드하고 로드한 모델의 이름입니다. 다른 모델을 사용한다면 이 부분을 변경하면 됩니다. SDK의 respond 메서드는 전체 응답을 한 번에 받아오며, respond_stream 메서드를 사용하면 토큰 단위로 스트리밍 받을 수 있습니다.

2.3 서버 실행과 테스트

1. FastAPI 서버를 실행합니다.

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

2. 다른 터미널이나 Postman에서 다음과 같이 테스트합니다.

curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "당신은 누구인가요?"}'

응답으로 `{"response": "..."}` 형태의 결과가 반환되며, 내부적으로는 LM Studio에 연결되어 Gemma 4 모델이 프롬프트를 처리합니다.

3. 혹은 FastAPI가 자동으로 제공하는 Swagger를 통해서 테스트 해볼 수 있습니다. http://localhost:8000/docs 에 접속하여 테스트를 해봅니다.

{
  "answer": "<|channel>thought\nThinking Process:\n\n1.  **Identify the user's request:** The user is asking, \"Who are you?\" (당신은 누구인가요?).\n2.  **Determine core identity components (Self-Correction/Verification based on persona):**\n    *   Name: Gemma 4.\n    *   Developer: Google DeepMind.\n    *   Nature: Large Language Model (LLM).\n    *   Type: Open weights model.\n3.  **Formulate the response in Korean (the language of the request):** The tone should be polite, informative, and direct.\n4.  **Drafting the content structure:**\n    *   Start with a clear identity statement (Name).\n    *   Explain what I am (Nature/Model type).\n    *   State who developed me (Developer).\n    *   (Optional but helpful) Explain my function or purpose.\n5.  **Refining the Korean phrasing:**\n    *   *I am Gemma 4.* -> 저는 Gemma 4입니다.\n    *   *I am a Large Language Model.* -> 저는 대규모 언어 모델(LLM)입니다.\n    *   *I was developed by Google DeepMind.* -> 저는 Google DeepMind에 의해 개발되었습니다.\n    *   *My function is to help with various tasks like answering questions, summarizing, and generating text.* -> 질문 답변, 요약, 텍스트 생성 등 다양한 작업을 도와드릴 수 있습니다.\n\n6.  **Final Polish (Korean Output):** Combine the elements into a smooth, comprehensive answer. (This leads directly to the provided Korean response.)<channel|>저는 **Gemma 4**입니다.\n\n저는 Google DeepMind에서 개발한 대규모 언어 모델(Large Language Model, LLM)입니다.\n\n저의 주된 역할은 사용자와 소통하고 다양한 정보를 제공하며, 질문에 답변하거나 글을 작성하는 등 여러 가지 언어 기반 작업을 도와드리는 것입니다. 궁금한 점이 있거나 도움이 필요한 부분이 있으면 언제든지 말씀해 주세요!"
}

4. 응답의 내용을 살펴보면 정상적인 최종 답변만 나온 상태가 아니라, 모델의 중간 사고 과정(Thinking Process)과 최종 답변이 함께 노출된 형태입니다.

   • <|channel>thought 이 부분은 모델이 내부적으로 답을 만들기 위해 사용한 사고 채널처럼 보입니다.

   • Thinking Process: 아래의 1~6번 사용자의 질문을 어떻게 해석하고, 어떤 요소를 답변에 넣을지 정리한 중간 추론 과정입니다.

   • <channel|>저는 **Gemma 4**입니다. 이후 이 부분이 실제 사용자에게 보여주려던 최종 응답입니다.

   • 즉, 이 JSON의 answer 값에는 “생각 과정 + 최종 답변”이 한 문자열에 같이 들어가 있습니다. 해석하면 이런 구조입니다.

     1. 사용자의 질문 파악 사용자가 “당신은 누구인가요?”라고 물었다고 인식
     2. 답변에 넣을 핵심 정보 정리
        • 이름: Gemma 4
        • 개발사: Google DeepMind
        • 정체: 대규모 언어 모델
        • 성격: 오픈 웨이트 모델
     3. 답변 언어 결정 사용자가 한국어로 물었으니 한국어로 답변
     4. 답변 구성 초안 작성
        • 나는 누구인지
        • 어떤 모델인지
        • 누가 만들었는지
        • 무엇을 할 수 있는지
     5. 한국어 문장 다듬기 영어 개념을 자연스러운 한국어 문장으로 바꾸는 단계
     6. 최종 문장 생성 실제 사용자에게 보여줄 최종 답변 완성
        • 이 과정은 사용자에게 그대로 보이면 안 되는 경우가 많습니다. 따라서 현재의 응답 형태는 로직 구현에서는 사용하되, 답변 받은 그대로를 서빙하지는 않도록 해야겠습니다.

3. 마무리와 다음 단계

이번 글에서는 Windows 환경에서 LM Studio로 Gemma 4 모델을 다운로드·로드하고, Python의 FastAPI를 사용해 간단한 REST API를 만드는 방법을 살펴보았습니다.

LM Studio가 제공하는 로컬 API를 직접 호출할 수도 있지만, FastAPI를 덧씌우면 네트워크를 통한 활용이나 추가 로직을 포함한 애플리케이션 통합이 쉬워집니다. 필요에 따라 다음과 같은 확장을 시도해볼 수 있겠습니다.

• 다른 모델 사용: E4B보다 더 큰 26B A4B나 31B 모델을 로드해 성능과 응답 품질을 비교해 볼 수 있습니다.
• 스트리밍 응답 처리: respond_stream 메서드를 사용하여 웹소켓이나 SSE(Server-Sent Events) 형태로 토큰을 실시간으로 전달할 수 있습니다.
• 보안 강화: FastAPI에 인증과 권한 제어를 추가하고, LM Studio 서버 자체의 인증 기능을 활성화해 데이터 노출을 방지합니다.

LM Studio와 FastAPI를 통해 로컬 LLM을 간편하게 활용해 보시면, 여러분의 프로젝트나 서비스에 맞춰 자유롭게 커스터마이징할 수 있습니다.

제게는 FastAPI가 익숙하고 빠르기 때문에 FastAPI를 사용한 AI 서빙을 소개하였습니다. 하지만 프로젝트의 성격이 일반적인 '웹 서비스'가 아니라 '모델 추론' 그 자체에 집중되어 있다면, 범용 웹 프레임워크보다는 ML Serving 전용 프레임워크를 고민해 볼 수 있습니다.

이번 글에서는 기존의 FastAPI 기반 손글씨 숫자 인식 백엔드를 BentoML 구조로 전환하여, 더욱 AI 모델 서빙에 최적화된 구조로 개선하는 방법을 살펴봅니다.

1. 왜 FastAPI 대신 BentoML인가?

FastAPI는 매우 훌륭한 범용 웹 API 프레임워크입니다. 하지만 모델 서빙 관점에서 보면 다음과 같은 차이가 있습니다.

• 서빙 엔트리포인트의 단순화: 모델 로드, 전처리, API 정의를 하나의 'Service' 클래스 안에서 직관적으로 관리할 수 있습니다.
• ML 특화 기능: 추후 GPU 가속을 위한 Runner, 마이크로서비스 확장을 위한 배치(Batch) 처리, 모델 버전 관리 등을 별도 설정 없이 바로 사용할 수 있습니다.
• 표준화된 패키징: bentofile.yaml 하나로 의존성과 환경을 정의하고 Docker 이미지까지 쉽게 빌드할 수 있습니다.

결국, 지금 프로젝트는 웹 서비스라기보다 모델 추론 서비스에 가깝기 때문에 BentoML로 옮겨볼 명분이 충분하다고 생각합니다.

2. 현재 backend 구조 vs 목표 구조

기존 FastAPI 구조는 웹 애플리케이션의 관례를 따르느라 파일이 여러 갈래로 찢어져 있었습니다.

기존 FastAPI 구조

backend/
├─ app/
│  ├─ api/routes/predict.py  # 엔드포인트
│  ├─ core/config.py         # 설정
│  ├─ schemas/prediction.py  # Pydantic 모델
│  └─ services/
│     ├─ image_preprocessing.py # 전처리 로직
│     └─ inference.py           # 모델 로드 및 추론
├─ models/
│  └─ digit_model_28.joblib
└─ main.py

변경 후 BentoML 구조

digit-recognition/
├─ bentoml_backend/
│  ├─ service.py         # 핵심 서빙 로직 (통합)
│  ├─ preprocessing.py    # 전처리 (재사용)
│  ├─ bentofile.yaml     # 패키징 설정
│  ├─ requirements.txt    # 의존성
│  └─ models/
│     └─ digit_model_28.joblib
└─ frontend/

3. BentoML 서비스 코드 구현

핵심은 service.py입니다. 기존에 흩어져 있던 책임을 하나의 클래스로 모읍니다.

from __future__ import annotations

import base64
import io
from pathlib import Path

import bentoml
import joblib
import numpy as np
from PIL import Image
from pydantic import BaseModel

# 기존 전처리 로직 재사용
from preprocessing import preprocess_canvas_png_to_28x28

MODEL_PATH = Path(__file__).resolve().parent / "models" / "digit_model_28.joblib"

class ImageRequest(BaseModel):
    image: str  # Base64 data URL

class PredictionResponse(BaseModel):
    digit: int

@bentoml.service(
    name="digit-recognizer",
    traffic={"timeout": 10},
)
class DigitRecognizerService:
    def __init__(self) -> None:
        # 서비스 시작 시 모델을 한 번만 로드하여 메모리에 유지
        self.model = joblib.load(MODEL_PATH)

    @bentoml.api
    def predict(self, request: ImageRequest) -> PredictionResponse:
        # 1. Base64 이미지 디코딩
        _, encoded = request.image.split(",", 1)
        image_data = base64.b64decode(encoded)
        image = Image.open(io.BytesIO(image_data))

        # 2. 전처리 (28x28 grayscale 변환 등)
        data_28 = preprocess_canvas_png_to_28x28(image)

        # 3. 모델 입력 규격($1 \times 784$)에 맞게 변형
        values = data_28.flatten().reshape(1, -1)

        # 4. 추론
        prediction = self.model.predict(values)[0]

        return PredictionResponse(digit=int(prediction))

전처리 코드(preprocessing.py)

이 프로젝트의 품질을 결정하는 전처리 로직(Crop, Resize, Center of Mass 등)은 프레임워크와 무관한 순수 로직이므로 파일명만 바꿔서 그대로 재사용합니다.

4. 패키징 설정: bentofile.yaml

BentoML의 강력함은 이 설정 파일에서 나옵니다. 어떤 파일을 포함할지, 어떤 라이브러리가 필요한지 명시합니다.

service: "service:DigitRecognizerService"
labels:
  project: "digit-recognition"
  framework: "bentoml"
include:
  - "*.py"
  - "models/*.joblib"
python:
  packages:
    - bentoml
    - joblib
    - numpy
    - pillow
    - scikit-learn
    - pydantic

5. 실행 및 프론트엔드 연동

실행 방식의 변화

기존 uvicorn 대신 bentoml serve 명령어를 사용합니다.

# 개발 모드 실행
bentoml serve service:DigitRecognizerService --reload

이 경우, http://localhost:3000/ 로 접속 시 아래의 Swagger 페이지를 확인할 수 있습니다.

프론트엔드(Vue.js 등)의 대응

이번 BentoML의 변경에서 가장 좋은 점은 프론트엔드 코드를 거의 고칠 필요가 없다는 것입니다. BentoML도 동일하게 POST /predict 엔드포인트를 생성하며, 우리가 정의한 ImageRequest 스키마가 기존 FastAPI의 스키마와 동일하다면 HTTP 계약(Contract)이 유지되기 때문입니다.

포트 번호나 API 베이스 URL 정도만 새 서버 주소에 맞게 업데이트하면 됩니다.

6. 마이그레이션 순서 (Best Practice)

안전한 전환을 위해 다음 순서를 권장합니다.

1. 로직 분리: 기존 image_preprocessing.py를 독립된 파일로 추출합니다.
2. 서비스 작성: service.py를 작성하고 모델 로드 테스트를 진행합니다.
3. 계약 검증: bentoml serve를 띄운 후, Postman이나 curl로 기존과 동일한 JSON 응답이 오는지 확인합니다.
4. 프론트엔드 연결: VITE_API_BASE_URL 등을 수정하여 실제 캔버스와 연동합니다.
5. 정리: 검증이 완료되면 더 이상 필요 없는 app/, main.py 등 FastAPI 관련 파일을 제거합니다.

7. 마무리

FastAPI에서 BentoML로의 교체는 단순히 라이브러리를 바꾸는 작업이 아닙니다. 이것은 프로젝트의 중심을 "범용 웹 API"에서 "전문화된 모델 서비스"로 옮기는 작업입니다.

이렇게 구조를 정리해두면, 나중에 모델을 업데이트하거나, 다른 모델(예: PyTorch 기반)로 교체하거나, 대규모 트래픽을 처리하기 위해 스케일 아웃을 할 때 훨씬 유연하게 대응할 수 있습니다.

결론적으로는 아래의 흐름을 거쳐서 서빙을 할 수 있습니다.

1. FastAPI 서버에서 모델 로드
2. 예측 API 구성
3. 입력 이미지 전처리
4. Vue 캔버스 UI 구현
5. 프론트엔드와 백엔드 연결
6. 로컬에서 실행하고 테스트

1. 이번 글의 범위

이번 글은 아래와 같이 저번 글에서 학습해둔 모델을 가지고 있다는 전제로 출발합니다.

• 학습된 모델 파일이 이미 존재함
• 예: backend/models/digit_model_28.joblib

이제 목표는 학습된 모델을 웹 서비스로 연결하는 것 으로 생각하고, 아래와 같은 생각을 할 수 있을 것 같습니다.

• 모델 파일을 FastAPI에서 어떻게 불러오나?
• 브라우저에서 그린 숫자를 어떻게 서버로 보내나?
• 서버는 그 이미지를 어떻게 모델 입력 형태로 바꾸나?
• 예측 결과를 화면에 어떻게 보여주나?

위 궁금증을 해결하기 위해서 하나하나 설명을 시작해보겠습니다.

2. 프로젝트 구조

서빙 관점에서 핵심 구조만 보면 아래와 같습니다.

digit-recognition/
├─ backend/
│  ├─ app/
│  │  ├─ api/routes/predict.py
│  │  ├─ core/config.py
│  │  ├─ schemas/prediction.py
│  │  └─ services/
│  │     ├─ image_preprocessing.py
│  │     └─ inference.py
│  ├─ models/
│  ├─ requirements.txt
│  └─ main.py
└─ frontend/
   ├─ src/
   │  ├─ components/DigitRecognizer.vue
   │  ├─ views/HomeView.vue
   │  ├─ router/index.ts
   │  └─ main.ts
   ├─ package.json
   ├─ vite.config.ts
   └─ .env.local

구조를 역할 기준으로 나누면 아래처럼 이해할 수 있습니다.

• backend/app/services/inference.py: 모델 로드와 예측
• backend/app/services/image_preprocessing.py: 입력 이미지 전처리
• backend/app/api/routes/predict.py: 예측 API
• frontend/src/components/DigitRecognizer.vue: 사용자가 숫자를 그리는 UI

3. 전체 서빙 흐름

모델 서빙 흐름은 아래의 내용으로 설명할 수 있습니다.

(Gemini 생성)

[사용자]
  -> 브라우저 캔버스에 숫자 그리기
  -> 예측 버튼 클릭

[Vue.js]
  -> canvas 이미지를 Base64 PNG로 변환
  -> POST /predict 요청 전송

[FastAPI]
  -> Base64 디코딩
  -> 이미지 전처리
  -> 모델 추론
  -> JSON 응답 반환

[Vue.js]
  -> 예측 결과 표시

핵심은 모델 파일 자체보다 입력 형식을 모델이 기대하는 형태로 잘 맞춰주는 것입니다.
브라우저에서 그린 자유로운 이미지를 28x28 숫자 배열로 바꾸는 과정이 실제 서빙 품질을 좌우합니다.

4. 백엔드: FastAPI 서버 구성

4-1. 진입점

# backend/main.py

from app.main import app

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)

backend/main.py는 핵심 로직이 있는 파일이 아니라, 기존 실행 방식이나 편의성을 위해 남겨둔 파일입니다. 실제 앱은 backend/app/main.py에 있습니다.

서버 실행은 아래처럼 하면 됩니다.

cd backend
uvicorn app.main:app --reload

4-2. FastAPI 앱 초기화

# backend/app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.routes.predict import router as predict_router


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(predict_router)

backend/app/main.py에서는 FastAPI 앱을 생성하고, CORS와 라우터를 등록합니다.

여기서 중요한 포인트는 두 가지입니다.

1. 프론트엔드에서 API를 호출할 수 있도록 CORSMiddleware 추가
2. /predict 라우터 등록

개발 환경에서는 프론트엔드와 백엔드가 서로 다른 포트에서 실행되므로 CORS 설정이 필요합니다.
현재는 모든 Origin을 허용하고 있어 개발 중 연결이 쉽습니다.

5. 모델 파일 경로 관리

# backend/app/core/config.py

from pathlib import Path


BACKEND_DIR = Path(__file__).resolve().parents[2]
MODELS_DIR = BACKEND_DIR / "models"
MODEL_FILENAME = "digit_model_28.joblib"
MODEL_PATH = MODELS_DIR / MODEL_FILENAME
LEGACY_MODEL_PATH = BACKEND_DIR / MODEL_FILENAME

모델 파일 위치는 backend/app/core/config.py에서 관리합니다.

핵심 값은 다음과 같습니다.

• MODELS_DIR
• MODEL_FILENAME
• MODEL_PATH
• LEGACY_MODEL_PATH

이렇게 경로를 별도 설정 파일로 분리해 두면 좋은 점이 있습니다.

• 추론 코드에서 경로 문자열을 하드코딩하지 않아도 됩니다.
• 모델 저장 위치가 바뀌어도 수정 지점이 줄어듭니다.
• 레거시 경로 fallback도 깔끔하게 처리할 수 있습니다.

즉, 서빙 코드에서 중요한 "모델 파일을 어디서 읽을 것인가"라는 문제를 한 곳에서 관리하는 구조입니다.

6. 모델 로드 (서버 시작 시 한 번만)

# backend/app/services/inference.py

import base64
import io
from pathlib import Path

import joblib
import numpy as np
from PIL import Image

from app.core.config import LEGACY_MODEL_PATH, MODEL_PATH
from app.services.image_preprocessing import preprocess_canvas_png_to_28x28


def _resolve_model_path() -> Path:
    for candidate in (MODEL_PATH, LEGACY_MODEL_PATH):
        if candidate.exists():
            return candidate

    raise RuntimeError(
        "Model file not found. Run `python scripts/train_model_28.py` to create "
        f"`{MODEL_PATH.name}`."
    )


def load_model():
    model_path = _resolve_model_path()

    try:
        return joblib.load(model_path)
    except Exception as exc:
        raise RuntimeError(f"Error loading model from {model_path}: {exc}") from exc


model = load_model()


def debug_print_image_28(data_28: np.ndarray) -> None:
    """
    28x28 데이터를 터미널에 대략 시각화.
    """
    print("\n[AI가 인식한 28x28 이미지 데이터]\n")
    for row in data_28:
        line = ""
        for value in row:
            if value > 180:
                line += "##"
            elif value > 50:
                line += ".."
            else:
                line += "  "
        print(line)
    print("\n" + "-" * 40)


def predict_digit_from_data_url(data_url: str) -> int:
    _, encoded = data_url.split(",", 1)
    image_data = base64.b64decode(encoded)
    img = Image.open(io.BytesIO(image_data))

    data_28 = preprocess_canvas_png_to_28x28(img)
    debug_print_image_28(data_28)

    values = data_28.flatten().reshape(1, -1)
    print(
        "X range:",
        float(values.min()),
        float(values.max()),
        "mean:",
        float(values.mean()),
    )

    try:
        scores = model.decision_function(values)[0]
        top = np.argsort(scores)[::-1][:3]
        print("top3 scores:", [(int(i), float(scores[i])) for i in top])
    except Exception:
        pass

    prediction = model.predict(values)[0]
    return int(prediction)

실제 모델 로드 로직은 backend/app/services/inference.py에 있습니다.

이 파일은 크게 아래 역할을 맡습니다.

1. 모델 파일 찾기
2. joblib로 모델 로드
3. 입력 이미지 예측

6-1. 모델 파일 찾기

_resolve_model_path()는 먼저 표준 경로를 확인하고, 없으면 레거시 경로까지 확인합니다.

즉, 다음 두 위치를 순서대로 탐색합니다.

• backend/models/digit_model_28.joblib
• backend/digit_model_28.joblib

6-2. 서버 시작 시 미리 로드

중요한 부분은 이 코드입니다.

model = load_model()

이 구문이 모듈 레벨에 있기 때문에, FastAPI 서버가 시작될 때 모델이 메모리에 한 번 올라갑니다.

이 방식의 장점은 명확합니다.

• 요청마다 모델 파일을 다시 읽지 않아도 됩니다.
• 응답 속도가 빨라집니다.
• 추론 API 구조가 단순해집니다.

AI 서빙에서는 "요청이 올 때마다 모델을 다시 로드하지 않는 것"이 기본입니다.

7. 예측 API 만들기

# backend/app/api/routes/predict.py

from fastapi import APIRouter, HTTPException

from app.schemas.prediction import ImageRequest, PredictionResponse
from app.services.inference import predict_digit_from_data_url

router = APIRouter()

@router.post("/predict", response_model=PredictionResponse)
async def predict_digit(request: ImageRequest) -> PredictionResponse:
    try:
        digit = predict_digit_from_data_url(request.image)
        return PredictionResponse(digit=digit)
    except Exception as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

예측 엔드포인트는 backend/app/api/routes/predict.py에 있습니다.

API는 매우 단순합니다.

• 경로: POST /predict
• 요청: Base64 이미지 문자열
• 응답: 예측 숫자

요청 예시

{
  "image": "data:image/png;base64,...."
}

응답 예시

{
  "digit": 7
}

이 구조가 좋은 이유는 프론트엔드와 백엔드의 계약이 간단하기 때문입니다.
캔버스에서 만든 이미지를 그대로 보내고, 서버는 숫자 하나만 반환하면 됩니다.

또한 예측 중 에러가 나면 HTTPException으로 400 응답을 내려주기 때문에, 프론트엔드에서 실패 상황도 처리할 수 있습니다.

8. 요청/응답 스키마 설계

# backend/app/schemas/prediction.py

from pydantic import BaseModel

class ImageRequest(BaseModel):
    image: str

class PredictionResponse(BaseModel):
    digit: int

backend/app/schemas/prediction.py에는 Pydantic 모델이 정의되어 있습니다.

• ImageRequest
• PredictionResponse

역할은 단순하지만 매우 중요합니다.

• 요청 JSON 구조를 명확하게 고정합니다.
• 응답 타입을 일관되게 유지합니다.
• FastAPI 자동 문서화에 도움을 줍니다.

작은 프로젝트라도 이런 식으로 요청/응답을 명시적으로 나누면 API가 훨씬 읽기 쉬워집니다.

9. Base64 이미지 처리와 추론

프론트엔드에서 보내는 값은 파일 업로드가 아니라 Data URL 문자열입니다.

예시

data:image/png;base64,iVBORw0K...

inference.py에서는 이 값을 아래 순서로 처리합니다.

1. 콤마 기준으로 메타데이터와 Base64 본문 분리
2. base64.b64decode()로 디코딩
3. Pillow의 Image.open()으로 이미지 객체 생성
4. 전처리 함수 호출
5. 모델 입력으로 변환 후 예측

전처리 결과는 28x28 배열이고, 이 값을 flatten() 해서 (1, 784) 형태로 바꾼 뒤 model.predict()에 넣습니다.

즉, 서빙의 핵심 로직은 사실 아래 한 줄로 압축됩니다.

브라우저 이미지 -> 전처리 -> 784차원 벡터 -> 모델 예측

10. 이미지 전처리(서빙 품질의 핵심)

# backend/app/services/image_preprocessing.py

import numpy as np
from PIL import Image, ImageFilter, ImageOps

def center_by_mass_np(data: np.ndarray) -> np.ndarray:
    """
    질량 중심(centroid)을 이미지 중앙(13.5, 13.5)에 가깝게 이동.
    SciPy 없이 numpy만으로 shift 구현.
    """
    ys, xs = np.indices(data.shape)
    total = float(data.sum())
    if total <= 0:
        return data

    cy = float((ys * data).sum() / total)
    cx = float((xs * data).sum() / total)

    target = (data.shape[0] - 1) / 2.0
    shift_y = int(round(target - cy))
    shift_x = int(round(target - cx))

    out = np.zeros_like(data)
    height, width = data.shape

    y0_src = max(0, -shift_y)
    y1_src = min(height, height - shift_y)
    x0_src = max(0, -shift_x)
    x1_src = min(width, width - shift_x)

    y0_dst = max(0, shift_y)
    y1_dst = min(height, height + shift_y)
    x0_dst = max(0, shift_x)
    x1_dst = min(width, width + shift_x)

    out[y0_dst:y1_dst, x0_dst:x1_dst] = data[y0_src:y1_src, x0_src:x1_src]
    return out

def preprocess_canvas_png_to_28x28(img: Image.Image) -> np.ndarray:
    """
    캔버스 PNG를 MNIST 스타일 28x28 grayscale 배열로 전처리.
    """
    img = img.convert("L")
    img = ImageOps.invert(img)

    arr = np.asarray(img)
    mask = arr > 30

    if mask.any():
        ys, xs = np.where(mask)
        x0, x1 = xs.min(), xs.max()
        y0, y1 = ys.min(), ys.max()

        box_width = int(x1 - x0 + 1)
        box_height = int(y1 - y0 + 1)
        pad = int(0.35 * max(box_width, box_height))

        x0 = max(0, x0 - pad)
        y0 = max(0, y0 - pad)
        x1 = min(arr.shape[1] - 1, x1 + pad)
        y1 = min(arr.shape[0] - 1, y1 + pad)

        img = img.crop((x0, y0, x1 + 1, y1 + 1))

    width, height = img.size
    side = max(width, height)
    square = Image.new("L", (side, side), 0)
    square.paste(img, ((side - width) // 2, (side - height) // 2))
    img = square

    img = img.resize((56, 56), Image.Resampling.LANCZOS)
    img = img.filter(ImageFilter.MaxFilter(size=3))
    img = img.filter(ImageFilter.MinFilter(size=3))
    img = img.filter(ImageFilter.GaussianBlur(radius=0.4))
    img = img.resize((28, 28), Image.Resampling.LANCZOS)

    data = np.asarray(img).astype(np.float32)
    return center_by_mass_np(data)

backend/app/services/image_preprocessing.py는 서빙 정확도를 결정하는 핵심 파일입니다.

모델은 28x28 흑백 숫자 입력을 기대하지만, 실제 브라우저 입력은 아래처럼 제각각입니다.

• 숫자 크기가 다름
• 위치가 치우쳐 있음
• 여백이 많음
• 획이 거칠거나 두꺼움

그래서 바로 모델에 넣지 않고 전처리를 수행합니다.

10-1. grayscale 변환

이미지를 L 모드로 변환해 흑백 이미지로 만듭니다.

10-2. 색 반전

캔버스는 흰 배경에 검은 숫자이므로, 이를 MNIST 스타일에 가깝게 맞추기 위해 반전합니다.

(MNIST 데이터 예시)

10-3. 숫자 영역 탐지

임계값보다 큰 픽셀만 남겨 숫자가 있는 실제 영역을 찾습니다.

10-4. bounding box 크롭

숫자 영역을 잘라내되, 너무 타이트하지 않게 패딩을 추가합니다.

10-5. 정사각형 보정

가로세로 비율이 달라도 정사각형 배경 위에 중앙 정렬합니다.

10-6. 리사이즈와 필터

코드는 아래 흐름으로 이미지를 다듬습니다.

1. 56x56 확대
2. MaxFilter
3. MinFilter
4. GaussianBlur
5. 28x28 축소

10-7. 질량 중심 정렬

center_by_mass_np()는 숫자의 질량 중심을 계산해 중앙으로 이동시킵니다.

이 단계가 중요한 이유는 사용자가 숫자를 약간 치우치게 그려도 모델이 더 안정적으로 받아들일 수 있기 때문입니다.

결국 전처리의 목적은 하나입니다.

사용자 입력을 학습 데이터와 최대한 비슷한 분포로 맞추는 것

서빙에서는 모델 성능만큼 전처리 품질도 중요합니다.

11. 프론트엔드: Vue.js 입력 화면 구성

<script setup lang="ts">
import { computed, onMounted, ref } from 'vue'

type PredictResponse = {
  digit?: number
}

const CANVAS_SIZE = 280
const BRUSH_SIZE = 25
const DEFAULT_API_BASE_URL = 'http://127.0.0.1:8000'

const canvas = ref<HTMLCanvasElement | null>(null)
const context = ref<CanvasRenderingContext2D | null>(null)
const isDrawing = ref(false)
const prediction = ref<number | null>(null)
const loading = ref(false)

const apiBaseUrl = computed(
  () => import.meta.env.VITE_API_BASE_URL?.trim() || DEFAULT_API_BASE_URL,
)

const getContext = () => {
  if (!context.value) {
    throw new Error('Canvas context is not initialized.')
  }

  return context.value
}

const getCanvas = () => {
  if (!canvas.value) {
    throw new Error('Canvas element is not available.')
  }

  return canvas.value
}

const clearCanvas = () => {
  const canvasElement = getCanvas()
  const ctx = getContext()

  ctx.fillStyle = 'white'
  ctx.fillRect(0, 0, canvasElement.width, canvasElement.height)
  prediction.value = null
}

const beginStroke = (x: number, y: number) => {
  const ctx = getContext()

  isDrawing.value = true
  ctx.beginPath()
  ctx.moveTo(x, y)
}

const continueStroke = (x: number, y: number) => {
  if (!isDrawing.value) {
    return
  }

  const ctx = getContext()

  ctx.lineTo(x, y)
  ctx.stroke()
}

const stopDrawing = () => {
  isDrawing.value = false
}

const startDrawing = (event: MouseEvent) => {
  beginStroke(event.offsetX, event.offsetY)
}

const draw = (event: MouseEvent) => {
  continueStroke(event.offsetX, event.offsetY)
}

const getTouchPosition = (event: TouchEvent) => {
  const canvasElement = getCanvas()
  const rect = canvasElement.getBoundingClientRect()
  const touch = event.touches[0]

  if (!touch) {
    return null
  }

  return {
    x: touch.clientX - rect.left,
    y: touch.clientY - rect.top,
  }
}

const handleTouchStart = (event: TouchEvent) => {
  const position = getTouchPosition(event)

  if (!position) {
    return
  }

  beginStroke(position.x, position.y)
}

const handleTouchMove = (event: TouchEvent) => {
  const position = getTouchPosition(event)

  if (!position) {
    return
  }

  event.preventDefault()
  continueStroke(position.x, position.y)
}

const predict = async () => {
  const canvasElement = getCanvas()

  loading.value = true

  try {
    const response = await fetch(`${apiBaseUrl.value}/predict`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        image: canvasElement.toDataURL('image/png'),
      }),
    })

    if (!response.ok) {
      throw new Error(`Prediction request failed with ${response.status}.`)
    }

    const result: PredictResponse = await response.json()
    prediction.value = typeof result.digit === 'number' ? result.digit : null
  } catch (error) {
    console.error('Prediction error:', error)
    window.alert('서버에 연결할 수 없습니다. FastAPI 서버가 실행 중인지 확인하세요.')
  } finally {
    loading.value = false
  }
}

onMounted(() => {
  const canvasElement = getCanvas()
  const ctx = canvasElement.getContext('2d')

  if (!ctx) {
    throw new Error('2D canvas context is not supported in this browser.')
  }

  context.value = ctx
  ctx.lineWidth = BRUSH_SIZE
  ctx.lineCap = 'round'
  ctx.lineJoin = 'round'
  ctx.strokeStyle = 'black'

  clearCanvas()
})
</script>

<template>
  <section class="flex min-h-screen items-center justify-center bg-slate-100 px-4 py-8">
    <div class="w-full max-w-md rounded-2xl bg-white p-8 text-center shadow-2xl">
      <h1 class="mb-3 text-2xl font-bold text-slate-800">숫자를 그려보세요 (0-9)</h1>
      <p class="mb-6 text-sm text-slate-500">
        캔버스에 숫자를 그린 뒤 예측하기 버튼을 눌러보세요.
      </p>

      <div class="mb-6 flex justify-center">
        <canvas
          ref="canvas"
          :width="CANVAS_SIZE"
          :height="CANVAS_SIZE"
          class="touch-none cursor-crosshair rounded-lg border-2 border-slate-700 bg-white shadow-inner"
          @mousedown="startDrawing"
          @mousemove="draw"
          @mouseup="stopDrawing"
          @mouseleave="stopDrawing"
          @touchstart="handleTouchStart"
          @touchmove="handleTouchMove"
          @touchend="stopDrawing"
        />
      </div>

      <div class="mb-6 flex justify-center gap-4 border border-black pt-1">
        <button
          type="button"
          class="rounded-lg bg-slate-500 px-6 py-2 font-medium text-white transition hover:bg-slate-600"
          @click="clearCanvas"
        >
          지우기
        </button>
        <button
          type="button"
          class="rounded-lg bg-blue-600 px-6 py-2 font-bold text-white transition hover:bg-blue-700 disabled:cursor-not-allowed disabled:bg-blue-300"
          :disabled="loading"
          @click="predict"
        >
          {{ loading ? '분석 중...' : '예측하기' }}
        </button>
      </div>

      <div v-if="prediction !== null" class="rounded-lg border border-blue-200 bg-blue-50 p-4">
        <p class="text-sm text-slate-600">예측 결과:</p>
        <p class="text-5xl font-black text-blue-600">{{ prediction }}</p>
      </div>
    </div>
  </section>
</template>

프론트엔드의 핵심은 frontend/src/components/DigitRecognizer.vue입니다.

이 컴포넌트는 아래 기능을 모두 담당합니다.

• 캔버스 생성
• 마우스 드로잉
• 터치 드로잉
• 캔버스 초기화
• 예측 API 호출
• 결과 표시

앱 구조는 간단합니다.

• main.ts: Vue 앱 생성
• router/index.ts: / 라우트 등록
• HomeView.vue: DigitRecognizer.vue 렌더링

즉, 실제 사용자 경험은 거의 전부 DigitRecognizer.vue에 모여 있습니다.

12. 캔버스 입력 구현

이 컴포넌트는 280x280 크기의 HTML canvas를 사용합니다.

const CANVAS_SIZE = 280
const BRUSH_SIZE = 25

이렇게 큰 캔버스에서 사용자가 편하게 숫자를 그리고, 서버에서는 이를 28x28로 줄입니다.

캔버스 관련 구현 포인트는 아래와 같습니다.

12-1. 초기화

clearCanvas()에서 캔버스를 흰색으로 채우고 예측 결과를 초기화합니다.

12-2. 마우스 드로잉

• mousedown
• mousemove
• mouseup
• mouseleave

이 이벤트를 이용해 선을 이어 그립니다.

12-3. 터치 드로잉

• touchstart
• touchmove
• touchend

모바일에서도 그릴 수 있도록 터치 이벤트를 별도로 처리합니다.

즉, 단순 데스크톱 데모가 아니라 모바일 입력까지 고려한 구현입니다.

13. 프론트엔드에서 예측 요청 보내기

예측 버튼을 누르면 프론트엔드는 캔버스 이미지를 Data URL로 직렬화합니다.

canvasElement.toDataURL('image/png')

그 다음 fetch()로 아래 요청을 보냅니다.

{
  "image": "data:image/png;base64,..."
}

서버 응답을 받은 뒤, digit 값이 숫자이면 화면에 그대로 렌더링합니다.

이 방식의 장점은 다음과 같습니다.

• 파일 업로드 UI가 필요 없습니다.
• 구현이 단순합니다.
• 브라우저 캔버스와 자연스럽게 연결됩니다.

14. API 주소를 환경 변수로 분리하기

DigitRecognizer.vue는 API 주소를 하드코딩하지 않고 환경 변수에서 읽습니다.

우선순위는 아래와 같습니다.

1. VITE_API_BASE_URL
2. 없으면 기본값 http://127.0.0.1:8000

frontend/.env.local 예시는 다음과 같습니다.

VITE_API_BASE_URL=http://127.0.0.1:8000

이렇게 하면 개발 환경, 스테이징 환경, 배포 환경에서 같은 프론트엔드 코드를 그대로 재사용할 수 있습니다.

15. 로컬에서 실행하기

이번 글은 학습 이후 단계만 다루므로, 모델 파일이 이미 있다고 가정합니다.

15-1. 백엔드 실행

cd backend
python -m venv venv
venv\Scripts\activate
pip install -r requirements.txt
uvicorn app.main:app --reload

백엔드는 기본적으로 http://127.0.0.1:8000에서 실행됩니다.

15-2. 프론트엔드 실행

cd frontend
npm install
npm run dev

15-3. 환경 변수 확인

frontend/.env.local

VITE_API_BASE_URL=http://127.0.0.1:8000

이제 브라우저에서 프론트엔드에 접속한 뒤 숫자를 그리고 예측하기 버튼을 누르면 됩니다.

16. 실제 요청/응답 예시

요청

POST /predict
Content-Type: application/json

{
  "image": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAA..."
}

응답

{
  "digit": 3
}

이 응답 구조는 매우 단순하지만, 프론트엔드 입장에서는 가장 다루기 쉬운 형태입니다.

17. 서빙 단계에서 자주 만나는 문제

17-1. 모델 파일을 찾지 못하는 경우

원인 후보

• 모델 파일 경로가 다름
• 기대한 위치에 모델 파일이 없음

확인 포인트

• backend/models/digit_model_28.joblib 존재 여부
• config.py의 경로 설정

17-2. 프론트엔드에서 서버 연결 실패

원인 후보

• FastAPI 서버가 꺼져 있음
• API 주소가 잘못 설정됨
• CORS 설정 문제

확인 포인트

• http://127.0.0.1:8000 접속 가능 여부
• VITE_API_BASE_URL 값 확인
• 백엔드 CORS 설정 확인

17-3. 예측이 불안정한 경우

원인 후보

• 전처리 결과가 학습 데이터 형식과 다름
• 숫자가 너무 작거나 한쪽에 치우쳐 있음
• 획 굵기와 입력 스타일 차이

해결 방향

• 전처리 파라미터 조정
• 캔버스 브러시 크기 조정
• 더 강한 모델로 교체

18. 마무리

모델 학습이 끝났다고 해서 서비스가 완성되는 것은 아닙니다.
실제로 사용자가 써볼 수 있으려면 API, 전처리, 입력 UI, 프론트엔드-백엔드 연결이 모두 맞물려야 합니다.

이번 프로젝트의 서빙 파트는 바로 그 최소 구성을 잘 보여줍니다.

• FastAPI가 모델을 메모리에 로드하고
• Vue.js가 사용자의 입력을 만들고
• 전처리 로직이 입력을 모델 형식에 맞추고
• 예측 결과를 다시 사용자에게 반환합니다

한 줄로 정리하면 이렇습니다.

AI 모델 서빙은 모델 파일을 올리는 작업이 아니라, 사용자의 입력을 모델과 연결되는 서비스 흐름으로 바꾸는 작업이다.

머신러닝 입문에서 빠지지 않는 MNIST 손글씨 숫자 데이터셋으로, 숫자(0~9)를 분류하는 모델을 학습하고 파일로 저장까지 해보겠습니다. 코드 길이는 10줄 남짓이지만, 데이터 로드 → 전처리 → 모델 학습 → 저장이라는 기본 흐름은 다 들어 있습니다.

1. 간단한 학습 코드

먼저 전체 코드를 한 번 보고, 아래에서 줄마다 상세히 살펴보겠습니다. 파일명은 train_model_28.py 정도로 저장해 두면 좋겠습니다.

# train_model_28.py
import joblib
import numpy as np
from sklearn.datasets import fetch_openml
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import LinearSVC

# 1. 데이터 불러오기
X, y = fetch_openml("mnist_784", version=1, return_X_y=True, as_frame=False)
y = y.astype(int)

# 2. 파이프라인 및 모델 설계
model = make_pipeline(
    StandardScaler(with_mean=False),  # 0이 많은 데이터에서 메모리/속도 측면에 유리
    LinearSVC(verbose=1)              # 학습 로그 출력
)

# 3. 모델 학습
model.fit(X, y)

# 4. 학습된 모델 저장
joblib.dump(model, "digit_model_28.joblib")
print("saved")

물리적인 코드 줄 수는 짧긴 하지만 학습 스크립트 한 번 돌려서 모델 파일 뽑기까지는 이 정도로 충분합니다.

2. 개발 환경 세팅

코드를 세세하게 살펴보기에 앞서서 패키지 설치를 해야합니다. 필요한 패키지는 세 개입니다. 아래의 명령어를 활용하여 패키지를 설치하여 줍니다. 가상환경을 먼저 세팅하고 진행하면 좋겠습니다.

pip install numpy scikit-learn joblib

• numpy: 배열/수치 연산
• scikit-learn: 데이터셋 로드, 전처리, 모델 학습
• joblib: 학습된 모델 저장/로드

3. 라이브러리 임포트

본격적으로 코드를 살펴보겠습니다. 라이브러리 임포트 부분은 아래와 같습니다.

import joblib
import numpy as np
from sklearn.datasets import fetch_openml
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import LinearSVC

여기서는 크게 세 가지가 중요합니다.

• fetch_openml: MNIST를 내려받아 X, y로 받기
• make_pipeline: 전처리 + 모델을 한 덩어리로 묶기
• joblib: 학습 결과를 파일로 저장하기

4. MNIST 다운로드

그 다음은 데이터를 불러오는 부분입니다.

X, y = fetch_openml("mnist_784", version=1, return_X_y=True, as_frame=False)
y = y.astype(int)

MNIST는 0부터 9까지의 손글씨 숫자를 모아 만든, 머신러닝 입문에서 가장 많이 쓰이는 데이터셋 중 하나입니다. 이미지 분류를 아주 작은 규모로 경험해보기 좋아서, 빠르게 기초 예제을 만들 때도 자주 등장합니다.

4.1 MNIST 상세 정보

• 구성: 총 70,000장(학습 60,000장 + 테스트 10,000장)
• 이미지 크기: 28×28 픽셀, 흑백 1채널
• 픽셀 값: 대체로 0~255 범위의 명암값. 0에 가까울수록 검은색, 255에 가까울수록 흰색에 가까움
• 라벨: 0~9의 숫자 클래스 10개

4.2 X, y는 어떤 형태인가

fetch_openml("mnist_784")는 각 이미지를 28×28로 주는 대신, 학습이 편하도록 길이 784짜리 벡터로 펴서 제공합니다.

• X: (70000, 784) 형태의 2차원 배열. 한 행이 이미지 한 장
• y: 길이 70,000짜리 레이블 배열

즉, mnist_784라는 이름은 28×28을 펼친 784차원 입력이라는 의미입니다.

4.3 왜 y를 int로 바꾸는가

OpenML에서 내려받은 y는 종종 문자열로 들어옵니다. 분류 자체는 문자열이어도 되지만, 뒤에서 평가 지표를 쓰거나 혼동행렬을 그릴 때는 정수형이 더 다루기 편해서 y.astype(int)로 고정해 둡니다.

4.4 fetch_openml을 쓸 때 알아두면 좋은 점

1. 첫 실행은 다운로드가 필요: 인터넷 속도나 OpenML 상태에 따라 시간이 걸릴 수 있습니다.
2. 캐시 사용: 한 번 받아두면 이후에는 로컬 캐시를 재사용하는 경우가 많습니다.
3. 메모리 사용량: MNIST는 크기가 아주 큰 편은 아니지만, 배열 형태로 로드하면 작업 환경에 따라 꽤 메모리를 씁니다. 노트북에서 돌릴 때는 다른 무거운 프로그램을 좀 닫아두는 게 편합니다.
4. 데이터 타입: 내려받은 X의 dtype은 상황에 따라 float로 들어올 수 있습니다. 이 글에서는 스케일링을 적용하므로 그대로 진행해도 문제 없습니다.

이제 데이터가 준비됐으니, 전처리와 모델을 파이프라인으로 묶어 학습을 진행해 보겠습니다.

5. 파이프라인과 전처리

model = make_pipeline(
    StandardScaler(with_mean=False),
    LinearSVC(verbose=1)
)

여기서 핵심은 전처리와 모델을 묶어서 한 번에 관리한다는 점입니다. 나중에 예측할 때도 같은 전처리를 자동으로 적용할 수 있어서 실수가 줄어듭니다.

5.1 StandardScaler 사용 이유

MNIST 픽셀 값은 0~255 범위입니다. 이런 입력에서 스케일 조정은 학습 안정성/속도에 도움 되는 경우가 많습니다.

5.2 with_mean=False의 의미

MNIST는 배경이 대부분 0이라서 0이 많은 편입니다. StandardScaler가 평균을 빼기 시작하면(기본값) 0이었던 값들이 0이 아닌 값으로 바뀌면서, 내부 표현이 촘촘해져 메모리/연산 비용이 늘어날 수 있습니다. 그래서 평균을 빼지 않도록 with_mean=False를 둡니다.

6. LinearSVC와 verbose

파이프라인의 두 번째 단계는 LinearSVC입니다. 이름 그대로 선형(Linear) SVM 분류기이고, scikit-learn에서는 빠르고 단단한 기준 모델로 많이 씁니다.

6.1 SVM이란

분류 문제에서 클래스 사이를 가르는 경계를 찾는 모델입니다. SVM은 그중에서도 경계를 가능한 넓은 여백(margin)으로 잡으려는 성질이 있고, 그래서 데이터가 어느 정도 잘 정리되어 있으면 생각보다 성능이 잘 나옵니다.

MNIST는 입력이 784차원(28×28)이라 차원이 꽤 높습니다. 이런 경우 선형 모델이 의외로 잘 먹히는 편이고, 학습/예측 속도도 빠릅니다.

6.2 LinearSVC를 고른 이유

• 학습이 비교적 빠름: 딥러닝처럼 GPU가 꼭 필요하지 않고, CPU로도 학습 가능
• 기준 모델로 좋음: 일단 되는 모델을 먼저 만들어두면, 이후 전처리/모델 교체의 효과를 비교하기가 쉬워짐
• 설명이 가능함: 완벽히 해석 가능한 모델은 아니지만, 적어도 선형 결합으로 분류한다는 직관이 있음

6.3 추가 사항

지금 글에서는 최대한 단순하게 두었지만, 실제로 돌리다 보면 아래를 고려해야 할 상황이 올 수 있습니다.

• 수렴 경고(ConvergenceWarning): 데이터/환경에 따라 반복 횟수가 부족하다는 경고가 뜰 수 있음. 이 때는 아래의 방법을 검토해보아야 함

  • max_iter를 늘리기
  • C(규제 강도)를 조정
  • 전처리를 조금 더 안정적으로 만들기

• 생각보다 긴 학습 시간: MNIST 전체 7만 장을 그대로 학습하면 꽤 걸릴 수 있음. 빠르게 감만 볼 거면 일부 샘플만 뽑아서 시작하는 것도 방법.

6.4 verbose=1 이란

LinearSVC(verbose=1)

verbose는 학습 로그를 출력하는 옵션입니다. MNIST처럼 데이터가 많으면 학습이 한동안 조용히 돌아가서 멈춘 건가 싶을 때가 있는데, 로그가 나오면 진행 중이라는 걸 확인할 수 있습니다.

로그가 너무 많으면 verbose=0으로 끄면 되고, 반대로 학습 상태를 더 보고 싶으면 값을 올려볼 수도 있습니다.

7. 학습

이제 실제 학습을 돌려보겠습니다.

model.fit(X, y)

학습은 이 한 줄로 끝납니다. 다만 실제로 돌려보면 어디서 얼마나 걸리는지, 돌아가고 있는 게 맞는지, 너무 오래 걸리면 어떻게 할지 같은 현실적인 포인트들이 생깁니다. 이 부분을 조금 보강해두면 시행착오가 줄어들 수 있습니다.

7.1 fit에서 실제로 일어나는 일

1. 파이프라인 1단계 StandardScaler가 입력 X로 스케일링 기준을 계산
2. 계산한 기준으로 X를 변환
3. 파이프라인 2단계 LinearSVC가 분류 경계(가중치)를 학습

즉, fit 한 번으로 전처리 기준 + 모델 파라미터가 같이 만들어집니다. 그래서 저장할 때 파이프라인 통째로 저장하는 게 편합니다.

7.2 예상보다 긴 학습 시간

MNIST 전체(70,000장)를 한 번에 학습하면 PC 사양에 따라 꽤 걸릴 수 있습니다. 처음에는 아래처럼 샘플을 조금만 뽑아서 감을 보는 것도 방법입니다.

# 빠른 테스트용(선택): 처음 10,000개만으로 먼저 학습
X_small = X[:10000]
y_small = y[:10000]
model.fit(X_small, y_small)

일단 돌아가는 걸 확인한 뒤 전체 학습으로 늘리면, 디버깅이 훨씬 수월합니다.

7.3 수렴 경고가 뜨는 경우

LinearSVC는 데이터/환경에 따라 수렴(최적화 완료) 전에 반복이 끝나서 경고가 뜰 수 있습니다. 이때는 보통 아래 중 하나로 해결합니다.

• max_iter 늘리기
• C 조정하기(규제 강도)

예시

model = make_pipeline(
    StandardScaler(with_mean=False),
    LinearSVC(verbose=1, max_iter=5000, C=1.0)
)

여기서 C는 클수록 규제가 약해지고(모델이 더 자유로워지고), 작을수록 규제가 강해집니다. 처음에는 기본값으로 두고, 경고가 반복되면 max_iter부터 올려보는 편이 안전합니다.

7.4 학습이 제대로 됐는지 빠르게 확인하는 방법

학습 직후에 아주 간단히라도 확인하고 넘어가면 좋습니다.

print("train score:", model.score(X[:2000], y[:2000]))

이 점수는 대충은 맞추는지 보는 용도입니다. 제대로 된 평가는 다음 단계에서 학습/테스트를 나눠서 하게 됩니다.

8. 모델 저장

이제 학습이 끝났으니 모델을 파일로 저장해 보겠습니다.

joblib.dump(model, "digit_model_28.joblib")
print("saved")

학습이 끝나면 모델을 파일로 저장해 둬야 다음에 다시 학습하지 않고 바로 사용할 수 있습니다.

scikit-learn 모델은 pickle로도 저장할 수 있지만, numpy 배열이 포함된 경우 joblib이 더 편하고 빠른 편이라 자주 씁니다.

9. 저장한 모델로 예측하기

오늘 글의 핵심 코드는 모델을 학습시키고 저장하는 것(Training)까지였습니다. 그렇다면 이렇게 저장된 .joblib 파일은 나중에 어떻게 써먹을 수 있을까요?

새로운 파이썬 파일(예: predict.py)을 만들어서 다음과 같이 활용할 수 있습니다. 이것이 바로 우리가 만든 AI를 서비스에 적용하는 '추론(Inference)' 단계입니다.

이번 파트에서는 이미지 파일(사진/스캔/그림판 캡처 등) 을 읽어서 테스트해 보겠습니다. 핵심은 입력 이미지를 MNIST 형태(28×28, 흑백, 0~255)로 맞춘 뒤 784로 펼친다 입니다.

아래 사진과 같이 그림판을 활용하여 숫자를 하나 그려서 digit.png 라는 파일로 저장하였습니다.

9.1 준비

이미지 처리를 위해 Pillow가 필요합니다.

pip install pillow

9.2 예측 코드 예제

아래 코드는 digit.png 파일을 읽어 예측합니다.

• 숫자는 화면에서 가운데에 크게 오도록 하는 게 유리합니다
• MNIST는 대체로 검은 배경 + 밝은(흰색) 숫자 형태라서, 반대로 되어 있으면 자동으로 뒤집도록 처리했습니다.

# predict_image.py
import joblib
import numpy as np
from PIL import Image, ImageOps

MODEL_PATH = "digit_model_28.joblib"
IMAGE_PATH = "digit.png"  # 테스트할 이미지 파일 경로

model = joblib.load(MODEL_PATH)

# 1) 이미지 로드 → 흑백 변환
img = Image.open(IMAGE_PATH).convert("L")

# 2) 여백/배경 정리(선택)
# 너무 어두운 배경이 섞이면 성능이 떨어질 수 있어서, 자동 대비 보정을 살짝 넣어둡니다
img = ImageOps.autocontrast(img)

# 3) 28x28로 리사이즈
# MNIST는 28x28 기준이라 일단 맞춥니다
img = img.resize((28, 28))

# 4) numpy 배열(0~255)로 변환
arr = np.array(img, dtype=np.float32)

# 5) MNIST와 배경/글자 밝기 방향 맞추기
# 평균이 높으면(전체가 밝으면) 흰 배경일 가능성이 커서 반전합니다
if arr.mean() > 127:
    arr = 255 - arr

# 6) (1, 784)로 펼치기
x = arr.reshape(1, -1)

pred = model.predict(x)[0]
print(f"예측 결과: {pred}")

9.3 잘 안 맞을 때 체크리스트

• 숫자가 너무 작거나 한쪽에 치우쳤는지
• 배경에 노이즈가 많은지(종이 질감, 그림자, 테이블 무늬)
• 숫자 색이 너무 옅은지(연필로 살짝 쓴 사진은 대비가 약해서 불리합니다)
• 여러 숫자가 같이 찍혔는지(이 모델은 한 장에 숫자 하나를 가정합니다)

이 예제는 가볍게 돌려보는 버전이라, 실전 수준으로 가려면 중앙 정렬(무게중심 이동), 이진화(threshold), 크롭/패딩 같은 전처리를 더해주는 게 효과가 큽니다.

10. 마무리

오늘 한 건 단순합니다. MNIST를 받아서, 전처리 붙이고, 선형 분류기로 학습하고, 모델 파일로 저장했습니다. 비록 최신 유행하는 딥러닝 코드는 아닐지라도, LinearSVC와 Pipeline의 조합은 실제 현업에서도 정형 데이터나 베이스라인 모델을 만들 때 사용되는 강력하고 검증된 방법론이라고 합니다. 코드가 짧아도 이 흐름을 직접 한 번 돌려보면 다음 단계가 훨씬 쉬워집니다.

AI 모델 학습의 다음 단계로는 아래 정도를 해보면 좋겠습니다.

1. 모델 평가하기: train_test_split 모듈을 사용하여 데이터를 학습용/테스트용으로 나누고, accuracy_score로 내 모델이 몇 %의 정확도를 가지는지 직접 측정해보기
2. 알고리즘 교체하기: LinearSVC 자리에 RandomForestClassifier나 다른 알고리즘을 넣어서 파이프라인을 실행해 보고 성능을 비교해보기
3. 딥러닝으로 넘어가기: 머신러닝의 기초를 다졌으니, 이제 텐서플로우(TensorFlow)나 파이토치(PyTorch)를 이용해 이 손글씨 데이터를 인공신경망(CNN)으로 더 정확하게 분류하는 방법을 학습해보기

다음 글에서는 방금 제시한 모델 학습의 다음 단계가 아닌, 오늘 학습한 모델을 FastAPI로 서빙하고 Vue.js로 테스트하는 예제를 만들어보도록 하겠습니다.

하지만 시스템이 커지고, 비동기 처리(@Async)나 외부 API 연동이 늘어나는 순간 이 편리함은 종종 걸림돌으로 변하게 될 수 있습니다. 예를 들어, "롤백이 됐다고 믿었는데” 실제로는 이미 커밋된 변경이 남아 데이터 정합성을 깨뜨리는 사고로 이어지기도 합니다.

이 글에서는 다음의 내용을 정리합니다.

• 비동기 환경에서 왜 메인 트랜잭션이 자동 롤백되지 않는지
• @Transactional의 범위를 어떻게 설계해야 커넥션과 성능이 무너지지 않는지
• 자동 롤백이 불가능한 영역을 공학적으로 푸는 방법(보상 트랜잭션, Outbox Pattern)
• 구현할 때 자주 밟는 함정(프록시, 내부 호출, 예외 관찰)

1. 왜 비동기에서 발생한 예외는 메인 트랜잭션을 롤백시키지 못할까

핵심은 Spring 트랜잭션 컨텍스트가 ThreadLocal 기반으로 관리된다는 점입니다.

1.1 ThreadLocal은 “스레드 전용 사물함”

Spring은 트랜잭션 상태(커넥션, 동기화 정보 등)를 ThreadLocal에 저장합니다. ThreadLocal은 특정 스레드 내부에서만 접근 가능한 저장소이기 때문에, 스레드가 바뀌면 동일한 트랜잭션 컨텍스트를 공유할 수 없습니다.

여기서 @Async가 등장합니다.

• @Async는 보통 새로운 스레드를 직접 만드는 것이 아니라 TaskExecutor(스레드 풀)에 위임해 다른 스레드에서 실행합니다.
• 결과적으로 실행 스레드가 달라지고, 메인 스레드의 ThreadLocal에 들어 있던 트랜잭션 컨텍스트는 전파되지 않습니다.

즉 흐름은 이렇게 됩니다.

1. 메인 스레드에서 트랜잭션을 시작하고 DB 작업을 수행합니다
2. @Async 메서드는 다른 스레드에서 실행되며 메인 트랜잭션을 모릅니다
3. 메인 스레드는 로직이 끝나면 커밋할 수 있고, 그 이후 비동기 쪽에서 예외가 나도 이미 커밋된 메인 트랜잭션을 자동으로 되돌릴 방법이 없습니다

시퀀스 다이어그램으로 확인해보면 아래와 같습니다.

여기서 기억할 문장 하나면 충분합니다.

비동기에서 예외가 나도 메인 트랜잭션을 “자동으로” 롤백시키지 못한다

비동기 메서드에 @Transactional을 따로 걸어도, 그건 새로운 별도 트랜잭션일 뿐 메인 트랜잭션과 동일해지지 않습니다.

2. @Transactional을 무심코 넓히면 생기는 또 다른 문제

비동기의 문제를 이해했다면, 이제 @Transactional을 “너무 쉽게” 붙였을 때 생기는 실전 문제를 봐야 합니다.

2.1 DB 커넥션 점유 시간이 길어진다

@Transactional이 메서드 전체를 감싸면, 해당 메서드가 끝날 때까지 DB 커넥션이 점유됩니다. 여기서 메서드 중간에 외부 API 호출, 파일 업로드, 긴 연산이 들어가면 어떤 일이 생길까요.

• 애플리케이션은 DB 작업을 하지 않으면서도 커넥션을 붙잡고 있게 됩니다
• 트래픽이 몰리면 커넥션 풀이 고갈되고, 시스템 전반이 지연 또는 장애로 이어집니다

비동기 이슈가 “정합성”의 문제라면, 트랜잭션 범위 남용은 “성능과 안정성”의 문제를 일으킵니다.

3. 해결 전략 1: 트랜잭션 범위를 정밀하게 줄이기 (TransactionTemplate)

모든 로직을 트랜잭션으로 감싸는 대신, 실제로 DB 쓰기가 필요한 구간만 트랜잭션을 적용하는 것이 출발점입니다. 이때 유용한 도구가 TransactionTemplate입니다.

public void placeOrder(OrderRequest request) {
    // 1) 트랜잭션 밖: 무거운 검증/외부 조회 (커넥션 미점유)
    validateOrder(request);

    // 2) 트랜잭션 안: 실제 DB 저장만 정밀하게 감싸기
    transactionTemplate.execute(status -> {
        return orderRepository.save(new Order(request));
    });

    // 3) 트랜잭션 밖: 사후 처리(알림 등)
    notificationService.send(request);
}

이 방식의 장점은 명확합니다.

• DB 커넥션을 붙잡는 시간이 줄어든다
• 예외 발생 시 영향 범위가 좁아진다
• 트랜잭션이 필요한 곳과 아닌 곳이 코드 레벨에서 분리된다

하지만 트랜잭션 범위를 줄이면, 다음 질문이 따라옵니다.

DB 저장은 성공했는데, 그 직후 서버가 죽어서 외부 작업(알림/이벤트 발행)을 못 하면 어떡하지

이 질문에 답하는 것이 다음 전략입니다.

4. 해결 전략 2: 자동 롤백이 불가능한 영역을 설계로 메운다

비동기와 외부 연동이 있는 순간, 데이터 정합성은 “자동 롤백”이 아니라 “설계”의 문제입니다. 대표적인 해결책은 두 가지입니다.

• 보상 트랜잭션(Compensating Transaction)
• Transactional Outbox Pattern

각각이 어떤 문제를 풀어주는지, 언제 쓰면 좋은지 이어서 보겠습니다.

5. 보상 트랜잭션: 이미 커밋된 것을 논리적으로 되돌리기

비동기 작업이 실패했는데 메인 트랜잭션이 이미 커밋됐다면, 시스템이 자동으로 되돌려주지 않습니다. 그러면 우리가 반대 방향의 상태 전이를 설계해야 합니다.

예를 들어

• 주문을 COMPLETED로 변경한 뒤
• 외부 쿠폰 발급이 실패했다면
• 주문을 CANCELLED로 되돌리는 로직이 보상 트랜잭션이 될 수 있습니다

5.1 보상 로직은 멱등해야 한다

보상 트랜잭션은 반드시 멱등성(idempotency) 을 가져야 합니다.

• 이미 CANCELLED라면 아무 동작을 하지 않고 유지
• 중복 호출되어도 결과가 안정적으로 유지

외부 시스템은 “부분 성공”이 가능하다는 점도 반드시 고려해야 합니다.

• 외부 API는 성공했는데 응답을 받기 전에 타임아웃이 발생
• 쿠폰은 발급됐지만 로컬 저장이 실패

그래서 보상 로직은 단순히 “실패 시 되돌리기”가 아니라 부분 성공 가능성까지 감안한 정합성 규칙이 필요합니다.

6. 보상 트랜잭션의 실패 격리: Propagation.REQUIRES_NEW

보상 로직은 보통 “실패한 흐름”과 분리되어야 합니다. 그래야 보상 작업 자체가 메인 흐름에 휘말려 같이 롤백되는 상황을 피할 수 있습니다.

@Transactional(propagation = Propagation.REQUIRES_NEW)

• REQUIRES_NEW는 기존 트랜잭션 유무와 관계없이 항상 새로운 트랜잭션을 시작합니다
• 보상 로직을 메인 흐름과 실패 격리(failure isolation) 시킵니다

다만 REQUIRES_NEW를 사용한다고 바로 해결이 되는 것은 아닙니다.

• 보상 로직 자체도 실패할 수 있습니다
• 실패 시 재시도 정책, 실패 기록(로그/DB/큐), 운영 관제가 함께 필요합니다

7. Transactional Outbox Pattern: “DB 저장”과 “외부 작업”의 간극 메우기

트랜잭션 범위를 줄였을 때 가장 무서운 시나리오는 이것입니다.

• DB는 커밋됐는데
• 이벤트 발행/외부 API 호출을 하기 전에 서버가 죽었다

이 문제를 푸는 대표 패턴이 Transactional Outbox입니다.

7.1 작동 원리

1. 메인 비즈니스 변경(주문 저장 등)과 함께 “해야 할 작업(이벤트)”을 같은 트랜잭션으로 Outbox 테이블에 기록
2. 트랜잭션 커밋 이후, 별도 프로세스(스케줄러/CDC/워커)가 Outbox를 읽어 실제 외부 작업을 수행
3. 실패하더라도 Outbox 기록이 남아 재시도 가능

Outbox는 “외부 작업을 트랜잭션에 포함시키지 않으면서도 유실을 막는 방법”입니다.

보상 트랜잭션이 “되돌리는 설계”라면, Outbox는 “확실히 실행되게 만드는 설계”에 가깝습니다.

8. 실패를 우아하게 다루는 법: 회복 탄력성(Resilience)

비동기, 워커, Outbox를 도입하는 순간 우리는 재시도를 전제로 설계해야 합니다.

8.1 멱등성(Idempotency)

동일한 요청이 두 번 전달되어도 결과가 한 번만 반영되도록 해야 합니다.

• 예: 주문 번호를 유니크 키로 사용해 중복 결제 방지
• 예: 이벤트 ID를 기준으로 중복 처리 방지

8.2 지수 백오프와 DLQ

• 지수 백오프(Exponential Backoff): 실패 시 1초, 2초, 4초… 간격을 늘리며 재시도해 대상 시스템 부하를 조절
• DLQ(Dead Letter Queue): 일정 횟수 이상 실패한 작업은 별도로 보관하고 알림을 통해 수동 처리 가능하게 함

9. 마지막 함정: 프록시와 내부 호출(Self-Invocation)

“다 적용한 것 같은데 REQUIRES_NEW가 먹지 않는다”는 경우, 의외로 원인은 단순합니다. 대부분 프록시를 우회한 내부 호출입니다.

Spring의 @Transactional은 보통 AOP 프록시 기반으로 동작합니다.

• 외부에서 빈을 호출할 때 프록시가 가로채며 트랜잭션을 시작합니다
• 같은 클래스 내부에서 this.someMethod()로 호출하면 프록시를 거치지 않아 트랜잭션이 적용되지 않습니다

그래서 보상 로직은 별도 서비스로 분리하는 것이 가장 안전합니다.

10. 예시 코드: 비동기 실패 시 보상 트랜잭션 실행

아래 예시는 “비동기 작업(외부 쿠폰 발급)이 실패하면 주문 상태를 취소로 되돌린다”는 단순화된 시나리오입니다.

// 1) 보상 로직을 별도 서비스로 분리
@Service
public class OrderCompensator {

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void cancelOrder(Long orderId) {
        // 멱등성 고려: 이미 CANCELLED면 no-op
        // 주문 상태를 'CANCELLED'로 변경
    }
}

// 2) 메인 서비스에서 비동기 작업 수행
@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderCompensator compensator;
    private final CouponService couponService;

    @Async
    public void issueCouponAsync(Long orderId) {
        try {
            couponService.issue(orderId);
        } catch (Exception e) {
            // 비동기 실패 감지 -> 프록시를 거쳐 보상 트랜잭션 실행
            compensator.cancelOrder(orderId);

            // 실무: 실패 기록, 재시도 큐 적재, 알림 등을 함께 고려
        }
    }
}

10.1 예외 관찰 포인트

반환 타입에 따라 예외를 다루는 방식이 달라집니다.

• void 반환이면 AsyncUncaughtExceptionHandler로 예외를 수집할 수 있습니다
• Future/CompletableFuture면 호출 측에서 예외를 관찰하고 후속 처리를 설계할 수 있습니다

여기서는 “실패를 즉시 감지하고 보상을 실행하는 패턴”에 초점을 맞췄습니다.

11. 실전 체크리스트

비동기 + 트랜잭션 조합에서 정합성을 높이려면 아래를 점검하세요.

• 실패를 어디에서 관찰할 것인가

  • try-catch, handler, future 등

• 보상 로직은 멱등한가

  • 중복 호출 방어, 상태 전이 규칙 명확화

• 보상 로직은 실패 격리되어 있는가

  • REQUIRES_NEW 또는 별도 처리 흐름

• 보상 실패 시 복구 전략이 있는가

  • 재시도 정책, 실패 기록, 운영 관제

• 외부 시스템의 부분 성공 가능성을 고려했는가

  • 타임아웃, 네트워크 오류, 중복 요청

• 트랜잭션 범위가 불필요하게 넓지 않은가

  • 외부 호출, 긴 연산이 트랜잭션 안에 있는지 점검

• 유실을 막아야 하는가

  • 필요하다면 Outbox Pattern 도입 검토

결론

Spring의 @Transactional은 강력한 도구이지만, 분산 환경과 비동기 처리가 포함된 구조에서는 한계가 명확합니다.

• 단순한 내부 DB 로직은 @Transactional로 생산성을 높이기
• 외부 연동과 성능 민감 구간은 TransactionTemplate로 범위를 좁히기
• 분산 환경에서 유실이 치명적이면 Outbox Pattern으로 실행을 보장하기
• 이미 커밋된 변경을 되돌려야 하면 보상 트랜잭션을 설계하기

비동기 환경에서 데이터 정합성을 유지하는 핵심은 결국 “자동 롤백”이 아니라 “정확한 이해 위에 세운 설계”입니다. ThreadLocal, 프록시, 커넥션 풀 같은 Spring의 실제 동작을 이해하고, 그 위에 보상/Outbox/회복탄력성을 얹을 때 시스템은 견고해집니다.

결론부터 말씀드리면, 아직 공식적으로 지원하지 않는 모델이라 약간의 라이브러리 코드 수정이 필요했습니다. 그 전체 과정을 차근차근 따라가 보시죠!

1. 테스트 환경 준비

1.1. FastAPI 예제 코드 다운로드

먼저 AutoGen 공식 GitHub 레포지토리에서 FastAPI 예제 코드를 준비합니다. 아래 경로에서 코드를 확인하고 다운로드할 수 있습니다. 저희는 이 중에서 app_agent.py를 사용할 겁니다.

• AutoGen FastAPI 샘플: https://github.com/microsoft/autogen/tree/main/python/samples/agentchat_fastapi

1.2. 필요 라이브러리 설치

다음으로, 예제 실행에 필요한 파이썬 패키지들을 pip으로 설치합니다.

pip install -U "autogen-agentchat" "autogen-ext[openai]" "autogen-ext[ollama]" "fastapi" "uvicorn[standard]" "PyYAML"

1.3. 모델 설정 파일 추가

프로젝트 루트 디렉터리에 model_config.yaml 파일을 생성하고, 로컬에서 실행 중인 Ollama의 Gemma3 모델을 사용하도록 설정합니다. {my-local-gemma3-model} 부분은 실제 Ollama 호스트 주소(예: http://localhost:11434)로 변경해주세요.

# Use Ollama with Gemma3
provider: autogen_ext.models.ollama.OllamaChatCompletionClient
config:
  model: gemma3:27b
  host: '{my-local-gemma3-model}'

2. 실행 및 에러 발생

이제 모든 준비가 끝났으니 서버를 실행해 봅니다. 터미널에서 uvicorn app_agent:app --port 8001 --reload 명령어로 서버를 실행하고 http://localhost:8001로 접속하면 간단한 채팅 UI가 나타납니다.

하지만 메시지를 보내는 순간... 500 Internal Server Error가 발생합니다!

서버 로그를 확인해 보면 더 자세한 에러 내용을 볼 수 있습니다.

핵심 에러 메시지는 다음과 같습니다.

[Error] Traceback (most recent call last):
  ...
KeyError: 'gemma3'

...
ValueError: model_info is required when model name is not a valid OpenAI model

에러 로그를 보니 _MODEL_INFO 딕셔너리에서 'gemma3'라는 키를 찾지 못해 KeyError가 발생했고, 이로 인해 모델 정보를 제대로 불러오지 못해 ValueError로 이어진 것을 알 수 있습니다.

3. 원인 분석 및 해결

3.1. 원인: 공식 미지원 모델

원인은 간단했습니다. 현재 제가 설치한 버전의 AutoGen 라이브러리가 gemma3 모델을 공식적으로 지원하지 않기 때문입니다. AutoGen 공식 문서를 확인해 봐도 지원하는 family 리스트에 gemma3는 포함되어 있지 않았습니다.

3.2. 해결책: 라이브러리 코드 직접 수정

이런 경우, 라이브러리가 업데이트될 때까지 기다리거나 직접 코드를 수정해서 모델을 추가하는 방법이 있습니다. 저희는 후자를 택했습니다. 가상환경에 설치된 AutoGen 라이브러리 파일을 직접 수정하여 gemma3 모델 정보를 추가해 보겠습니다.

1. autogen_ext\models\ollama\_model_info.py 수정

Ollama 클라이언트가 모델 정보를 참조하는 이 파일에 gemma3 정보를 추가합니다.

• _MODEL_INFO 딕셔너리에 모델의 특징(비전, 함수 호출 지원 여부 등)을 추가합니다.
• _MODEL_TOKEN_LIMITS 딕셔너리에 모델의 컨텍스트 길이를 추가합니다.

# autogen_ext\models\ollama\_model_info.py

_MODEL_INFO: Dict[str, ModelInfo] = {
    # ... 다른 모델 정보들
    "gemma3": {
        "vision": False,
        "function_calling": False,
        "json_output": True,
        "family": "gemma3", # family를 직접 지정해줄 수 있습니다.
        "structured_output": True,
    },
    # ...
}

_MODEL_TOKEN_LIMITS: Dict[str, int] = {
    # ... 다른 모델 토큰 정보들
    "gemma3": 8192,
    # ...
}

참고: 위 코드에서 family를 추가하고 ModelFamily.UNKNOWN 대신 문자열 "gemma3"를 사용했습니다. 더 명확하게 모델을 구분하기 위함이며, autogen_core의 ModelFamily 클래스에 GEMMA_3 = "gemma3"와 같이 상수를 추가하여 관리하는 것도 좋은 방법입니다.

2. app_agent.py 코드 수정

기존 예제 코드는 OpenAI 모델을 기본으로 가정하는 부분이 있어, Ollama 클라이언트를 명시적으로 사용하도록 코드를 수정해야 합니다.

• autogen_core.models.ChatCompletionClient 대신 autogen_ext.models.ollama.OllamaChatCompletionClient를 임포트합니다.
• get_agent 함수 내에서 OllamaChatCompletionClient.load_component(model_config)를 호출하여 모델 클라이언트를 로드합니다.

아래는 수정한 전체 app_agent.py 코드입니다.

import json
import os
import traceback
from datetime import datetime
from typing import Any

import aiofiles
import yaml
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
# Ollama 클라이언트를 직접 임포트합니다.
from autogen_ext.models.ollama import OllamaChatCompletionClient 

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

app = FastAPI()

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

app.mount("/static", StaticFiles(directory="."), name="static")

@app.get("/")
async def root():
    return FileResponse("app_agent.html")

model_config_path = "model_config.yaml"
state_path = "agent_state.json"
history_path = "agent_history.json"

async def get_agent() -> AssistantAgent:
    """Get the assistant agent, load state from file."""
    async with aiofiles.open(model_config_path, "r") as file:
        model_config = yaml.safe_load(await file.read())
    # Ollama 클라이언트를 명시적으로 로드합니다.
    model_client = OllamaChatCompletionClient.load_component(model_config)
    agent = AssistantAgent(
        name="assistant",
        model_client=model_client,
        system_message="You are a helpful assistant.",
    )
    if not os.path.exists(state_path):
        return agent
    async with aiofiles.open(state_path, "r") as file:
        state = json.loads(await file.read())
    await agent.load_state(state)
    return agent

# (이하 코드는 원본과 거의 동일)
# get_history, history, chat 함수 ...
# ...

@app.post("/chat", response_model=TextMessage)
async def chat(request: TextMessage) -> TextMessage:
    try:
        agent = await get_agent()
        response = await agent.on_messages(messages=[request], cancellation_token=CancellationToken())

        state = await agent.save_state()
        async with aiofiles.open(state_path, "w") as file:
            await file.write(json.dumps(state))

        history = await get_history()

        class DateTimeEncoder(json.JSONEncoder):
            def default(self, obj):
                if isinstance(obj, datetime):
                    return obj.isoformat()
                return super().default(obj)

        request_dict = request.model_dump()
        response_dict = response.chat_message.model_dump()

        history.append(request_dict)
        history.append(response_dict)
        async with aiofiles.open(history_path, "w") as file:
            await file.write(json.dumps(history, cls=DateTimeEncoder))

        assert isinstance(response.chat_message, TextMessage)
        return response.chat_message
    except Exception as e:
        error_message = {
            "type": "error",
            "content": f"Error: {str(e)}",
            "source": "system"
        }
        print("[Error]", traceback.format_exc())
        raise HTTPException(status_code=500, detail=error_message) from e

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

4. 테스트 성공!

위와 같이 라이브러리와 예제 코드를 수정한 뒤 서버를 다시 실행하고 메시지를 보내면... 드디어 정상적으로 로컬 Gemma3 모델이 응답을 생성합니다!

마무리

AutoGen의 FastAPI 예제를 로컬 Gemma3 모델과 연동하면서 발생한 문제를 해결하는 과정을 공유해 보았습니다. 최신 모델을 사용하다 보면 이처럼 라이브러리가 아직 지원하지 않는 경우가 종종 발생하는데, 오픈소스 프로젝트는 직접 코드를 수정해서 빠르게 대응해 볼 수 있다는 점이 큰 매력인 것 같습니다.

이 글이 저처럼 새로운 모델을 AutoGen에 적용해 보려는 분들께 작은 도움이 되었으면 좋겠습니다. 감사합니다!

문제 상황: 데이터가 누락되다

특정 로그 데이터가 Kibana에서 보이지 않는 문제가 발생했습니다. 문제가 된 데이터 구조는 아래와 같이 values 필드에 여러 객체를 담고 있는 형태였습니다.

// 문제가 되는 데이터 구조
{
  "properties": {
    "values": [
      {"bpm": 80, "time": "2025-07-21T18:12:50.542801"},
      {"bpm": 120, "time": "2025-07-21T18:12:50.542801"}
    ]
  }
}

데이터 형식을 변경하는 중에, 어느 순간부터 인덱싱이 되지 않고 있었습니다.

원인 분석: 동적 매핑의 함정

원인은 Elasticsearch의 동적 매핑(Dynamic Mapping) 기능에 있었습니다. Elasticsearch는 필드에 처음 들어온 데이터의 타입을 보고 자동으로 매핑을 결정합니다.

1. 성공했던 인덱스: values 필드에 객체 리스트가 먼저 들어오면서, 해당 필드가 object 타입으로 정상 매핑되었습니다.
2. 실패한 인덱스: values 필드에 숫자 리스트([80, 120]과 같은)가 먼저 들어오면서, 필드가 long 타입으로 매핑되어 버렸습니다.

실제 인덱스 매핑을 GET panic-digital-phenotype-logs-*/_mapping 명령어로 확인해보니 두 인덱스의 values 필드 타입이 다른 것을 발견할 수 있었습니다.

// 성공했던 인덱스 (2025.07.21)
"values": {
  "properties": {
    "bpm": { "type": "long" },
    "time": { "type": "text" }
  }
}

// 실패하는 인덱스 (2025.07.22)
"values": {
  "type": "long"
}

long 타입으로 매핑된 필드에는 객체 리스트가 들어올 수 없으니, 데이터가 누락될 수밖에 없었던 것입니다.

해결 방법: 템플릿으로 매핑 고정하기

이 문제를 해결하기 위해 Composable Index Template을 사용하여 values 필드의 매핑을 명시적으로 지정했습니다. 이렇게 하면 데이터가 어떤 순서로 들어오든 항상 의도한 대로 매핑이 적용됩니다.

1. Composable Index Template 설정

Elasticsearch 7.x 이상에서는 아래와 같이 템플릿을 설정하여 values 필드를 항상 object 타입으로 지정할 수 있습니다.

"template": {
  "mappings": {
    "properties": {
      "values": {
        "type": "nested", // 또는 "object"
        "properties": {
          "bpm": { "type": "long" },
          "time": { "type": "date" } // date 타입으로 변경하는 것을 권장
        }
      }
    }
  }
}

팁! 객체 배열을 다룰 때는 object보다 nested 타입을 사용하는 것이 각 객체를 독립적으로 쿼리할 수 있어 더 유리합니다. 또한 time 필드는 text보다 date 타입으로 매핑하는 것이 시간 기반 검색 및 분석에 효율적입니다.

2. 기존 인덱스 삭제 또는 Rollover

인덱스 템플릿은 새로 생성되는 인덱스에만 적용됩니다. 이미 잘못된 매핑으로 생성된 인덱스에는 소용이 없죠. 따라서 기존의 잘못된 인덱스를 삭제하거나 Rollover를 통해 새 인덱스를 만들어야 합니다.

DELETE panic-digital-phenotype-logs-2025.07.22

이후 데이터가 들어와 새 인덱스가 생성되면, 우리가 템플릿에 정의한 매핑이 올바르게 적용됩니다.

3. 데이터 구조 일관성 유지

가장 중요한 것은 애플리케이션(Logger) 단에서 전송하는 데이터의 구조를 일관되게 유지하는 것입니다. values 필드에는 항상 객체 리스트 형태만 보내도록 통일해야 매핑 충돌을 근본적으로 방지할 수 있습니다.

실전 점검 체크리스트

템플릿 적용 후에도 문제가 해결되지 않는다면 아래 사항들을 순서대로 점검해 보세요.

• 템플릿 적용 후 새 인덱스에서 테스트: 템플릿이 잘 적용되었는지 새 인덱스의 매핑을 직접 확인하세요.
• 기존 인덱스 삭제 또는 Rollover: 문제가 된 인덱스가 삭제되었는지 확인하세요.
• 실제 전송 데이터 구조 확인: 애플리케이션에서 보내는 데이터가 의도한 구조와 일치하는지 다시 한번 검증하세요.
• Logstash 파이프라인 확인: Logstash 필터(e.g., drop, mutate)에서 데이터를 변경하거나 누락시키고 있지는 않은지 확인하세요.
• Elasticsearch 인덱스 매핑 재확인: 최종적으로 데이터가 저장된 인덱스의 매핑이 올바른지 확인하세요.

결론

Elasticsearch의 동적 매핑은 편리하지만, 다양한 형태의 데이터가 하나의 필드로 들어올 경우 예기치 않은 문제를 일으킬 수 있습니다. Composable Index Template을 사용해 필드 매핑을 명시적으로 고정하고, 데이터 구조를 일관되게 유지하는 것이 안정적인 ELK 스택 운영의 핵심입니다.

이번에 구동해볼 모델은 Google에서 출시한 경량 언어모델인 Gemma 3입니다.

Gemma 3 모델의 요구 메모리

아래는 Gemma 3 모델의 크기별, 정밀도별 메모리 요구사항입니다.

자세한 모델 정보는 공식 문서에서 확인할 수 있습니다.

데이터 타입 FP32, BF16 등의 개념

FP32 (Single Precision Floating Point)

FP32는 32비트 부동 소수점 데이터 타입입니다. 높은 정확성을 제공하지만 메모리와 계산 자원을 상대적으로 많이 소모합니다.

BF16 (Brain Floating Point)

BF16은 FP32의 부동 소수점 표현 방식을 절반으로 줄인 16비트 데이터 타입입니다. FP32와 유사한 동적 범위를 유지하면서 메모리 사용량과 연산 속도를 크게 개선하여 주로 딥러닝 모델 훈련 시 효율성을 높이는 데 사용됩니다.

FP16 (Half Precision Floating Point)

FP16은 16비트의 부동 소수점 데이터 타입으로 FP32 대비 메모리 사용량과 연산 속도가 개선됩니다. 하지만 표현 가능한 범위가 좁아 수치 표현에서 정확성 저하 문제가 발생할 수 있습니다.

M1 맥북에서 가능한 모델

제가 가진 8GB 메모리의 M1 맥북 프로에서는 Gemma 3 4B 모델의 4-bit(Q4_0) 또는 QAT(INT4) 버전을 테스트할 수 있을 것 같습니다.

홈브루(Homebrew)로 Ollama 설치하기

제 맥북에는 이미 홈브루가 설치되어 있지만, 설치되지 않았다면 홈브루 공식 사이트에서 먼저 설치하세요.

터미널에서 다음 명령어로 Ollama를 설치하고 서비스를 활성화합니다.

brew install ollama
brew services start ollama

Gemma 3 모델 다운로드하기

이제 Ollama를 통해 Gemma 3 모델을 다운로드합니다.

• 기본 4-bit 모델

ollama pull gemma3:4b

• 더 가벼운 QAT 양자화 버전(메모리 사용량 약 1/3 감소)

ollama pull gemma3:4b-it-qat

• 모델 정보:

  • gemma3:4b: 4B 파라미터, 128K 컨텍스트, 3.3 GB
  • gemma3:4b-it-qat: QAT 적용 모델로 품질을 유지하며 메모리를 현저히 절감

간단한 프롬프트 실행 예시

터미널에서 간단한 예시 명령어를 실행해봅니다.

ollama run gemma3:4b-it-qat "BF16, FP16, FP32의 차이점을 설명해주세요."

이 명령어를 통해 로컬에서 간편하게 Gemma 3 LLM을 활용할 수 있습니다.

다만, 한 번 동작을 하면 램 점유율이 내려가지 않고 os가 전체적으로 느려지는 것을 확인했습니다. 이를 해결하기 위해서 pkill ollama 명령어를 활용하여 ollama 실행을 종료하였습니다.

ollama 명령어 예시

Ollama 대화 모드 내 유용한 명령어:

ollama run을 통해 모델과 상호 작용하는 동안 /로 시작하는 특수 명령을 사용할 수 있습니다.

• /?: 사용 가능한 모든 슬래시 명령어 목록이 포함된 유용한 메뉴를 표시합니다.
• /set parameter <매개변수_이름> <값>: 현재 채팅 세션에 대한 모델의 런타임 매개변수를 일시적으로 변경합니다. 예를 들어 /set parameter temperature 0.9는 창의성을 높이고 /set parameter num_ctx 8192는 이 세션의 컨텍스트 창을 늘립니다.
• /show info: 현재 로드된 모델에 대한 자세한 정보(매개변수, 템플릿 구조, 라이선스 등)를 표시합니다.
• /show modelfile: 현재 실행 중인 모델을 만드는 데 사용된 Modelfile의 내용을 표시합니다. 기본 모델, 매개변수 및 프롬프트 템플릿을 이해하는 데 유용합니다.
• /save <세션_이름>: 현재 채팅 기록을 명명된 세션 파일에 저장합니다.
• /load <세션_이름>: 이전에 저장한 채팅 세션을 로드하여 대화 기록을 복원합니다.
• /bye 또는 /exit: 대화형 채팅 세션을 정상적으로 종료하고 모델을 메모리에서 언로드합니다(다른 세션에서 사용하지 않는 경우). 일반적으로 Ctrl+D를 사용하여 종료할 수도 있습니다.

프롬프트 환경을 넘어서 좀 더 사용자 친화적인 UI를 원한다면, OpenWebUI와 같은 플랫폼을 함께 사용해보는 것도 좋을 것 같습니다.

클로저란 무엇인가?

클로저는 간단히 말해, 자신이 정의된 스코프(scope)의 외부에서 호출되더라도 자신이 속한 환경의 변수들을 기억하고 참조할 수 있는 함수입니다. 이는 보통 함수가 다른 함수를 반환하거나 내부 함수가 외부의 변수를 사용할 수 있게 할 때 사용됩니다.

파이썬으로 클로저를 명확히 이해하려면 다음과 같은 예시를 살펴봅시다.

def outer_function(outer_var):
    def inner_function(inner_var):
        return outer_var + inner_var
    return inner_function

closure_instance = outer_function(10)
print(closure_instance(5))  # 출력: 15

outer_function이 반환한 inner_function은 outer_var 값인 10을 기억하고 있으며, inner_var로 받은 5를 더해 15를 출력합니다.

다시 말해 inner_function은 외부 함수가 끝난 뒤에도 outer_var을 기억하고 활용합니다. 이러한 특성을 갖춘 함수를 클로저라고 합니다.

클로저의 주요 특징

클로저가 성립하기 위한 조건과 특성은 다음과 같습니다:

1. 함수의 중첩(Nested Functions)

클로저는 반드시 내부에 다른 함수를 포함하는 구조입니다.

2. 자유 변수(Free Variables)

내부 함수가 자신의 로컬 스코프에 정의되지 않은 변수, 즉 외부 스코프에 존재하는 변수를 참조할 수 있습니다.

3. 클로저 속성 (__closure__)

클로저가 생성된 함수는 __closure__ 속성을 통해 저장된 환경 변수들의 정보를 확인할 수 있습니다.

def outer_function(outer_var):
    def inner_function():
        return outer_var
    return inner_function

closure_instance = outer_function(42)
print(closure_instance.__closure__[0].cell_contents)  # 출력: 42

클로저의 다양한 활용 사례

1. 상태 유지 및 관리

클로저는 객체지향 프로그래밍 없이도 간단하게 상태를 저장할 수 있게 해줍니다.

def counter():
    count = 0

    def increment():
        nonlocal count
        count += 1
        return count

    return increment

count_up = counter()
print(count_up())  # 출력: 1
print(count_up())  # 출력: 2

2. 데코레이터(Decorator) 구현

데코레이터는 기존의 함수를 확장하거나 수정할 때 자주 사용됩니다.

def logger(func):
    def wrapper(*args, **kwargs):
        print(f"함수 {func.__name__}이 호출되었습니다.")
        result = func(*args, **kwargs)
        print(f"함수 {func.__name__}이 종료되었습니다.")
        return result
    return wrapper

@logger
def greet():
    print("안녕하세요!")

greet()
# 출력:
# 함수 greet이 호출되었습니다.
# 안녕하세요!
# 함수 greet이 종료되었습니다.

3. 함수 팩토리(Function Factory)

반복적 로직을 가진 여러 유사 함수를 생성할 때도 클로저를 활용할 수 있습니다.

def multiplier_of(n):
    def multiplier(x):
        return x * n
    return multiplier

double = multiplier_of(2)
triple = multiplier_of(3)

print(double(5))  # 출력: 10
print(triple(5))  # 출력: 15

클로저 사용 시 주의할 점

1. 자유 변수의 공유 문제

클로저 내에서 변수를 참조할 때 예상하지 못한 공유가 발생할 수 있습니다.

def make_multipliers():
    multipliers = []
    for i in range(5):
        multipliers.append(lambda x: x * i)
    return multipliers

multiplier_funcs = make_multipliers()
print(multiplier_funcs[0](2))  # 출력: 8 (기대값: 0)

위 코드에서 lambda x: x * i에서 i는 루프가 끝난 뒤의 최종 값(4) 를 참조하게 되기 때문에, x * 4가 되어 8이 출력됩니다.

이는 변수를 기본값 인수로 전달하여 해결할 수 있습니다:

def make_multipliers():
    return [lambda x, i=i: x * i for i in range(5)]

multiplier_funcs = make_multipliers()
print(multiplier_funcs[0](2))  # 출력: 0

2. 메모리 누수 문제

클로저가 참조하는 변수들이 오랫동안 메모리에 남아 있어 메모리 누수가 발생할 수 있습니다. 이를 방지하려면 클로저의 사용 범위를 명확히 관리해야 합니다.

메모리 누수를 방지하는 방법은 주로 다음과 같이 구체화할 수 있습니다.

방법 1. 명확한 참조 관리

클로저가 외부 변수를 필요 이상으로 참조하지 않도록 최소한의 변수만 유지해야 합니다.

[잘못된 예]

def create_large_data_holder(data):
    def inner():
        return data
    return inner

large_data = [x for x in range(10**7)]
holder = create_large_data_holder(large_data)

# large_data가 계속 메모리에 유지됨

위의 경우에서 가비지 컬렉터는 “더 이상 참조가 없다” 고 판단할 때만 메모리를 회수하므로, 이 상황에선 회수가 불가능합니다.

[개선된 예]

def create_large_data_holder(data):
    important_summary = sum(data)  # 필요한 정보만 추출

    def inner():
        return important_summary
    return inner

large_data = [x for x in range(10**7)]
holder = create_large_data_holder(large_data)

del large_data  # 명시적으로 메모리 해제 가능

방법 2. weakref(약한 참조) 사용하기

약한 참조를 사용하면 참조되는 객체가 필요 없을 때 자동으로 메모리에서 삭제됩니다.

import weakref

class LargeObject:
    def __init__(self, data):
        self.data = data

def closure_with_weakref(obj):
    weak_obj = weakref.ref(obj)

    def inner():
        obj_ref = weak_obj()
        if obj_ref is None:
            return "Object has been garbage collected"
        return obj_ref.data
    return inner

large_obj = LargeObject([1, 2, 3])
closure = closure_with_weakref(large_obj)

print(closure())  # 출력: [1, 2, 3]

del large_obj  # 명시적 객체 삭제
print(closure())  # 출력: Object has been garbage collected

방법 3. 필요하지 않은 클로저 명시적 삭제

사용하지 않는 클로저 인스턴스는 명시적으로 제거하는 습관을 가집니다.

def data_keeper():
    data = [x for x in range(10**6)]

    def inner():
        return len(data)

    return inner

keeper = data_keeper()
print(keeper())  # 사용 후

del keeper  # 명시적 삭제

방법 4. 클로저보다 클래스나 제너레이터 활용

상태 유지가 과도하거나 복잡한 경우, 간단한 클래스나 제너레이터로 바꾸면 메모리 관리가 쉬워집니다.

클래스 사용 예시:

class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1
        return self.count

counter = Counter()
print(counter.increment())

클래스는 내부 상태를 더 명확히 관리할 수 있도록 해줍니다.

결론

클로저는 파이썬 개발자에게 코드의 간결성, 확장성, 재사용성을 높여주는 매우 효과적인 도구입니다. 하지만 클로저의 개념과 주의사항을 명확히 이해하고 사용해야 잠재적인 문제들을 예방하고 더 나은 코드를 작성할 수 있습니다.

지난 3편의 FastAPI 구조 살펴보기 글을 통해서 ASGI 서버가 어떻게 동작하고, FastAPI의 기반이 되는 Starlette가 요청을 어떻게 처리하는지 알아보았습니다. 이번 글에서는 FastAPI의 OpenAPI 기반 API 문서가 언제 생성되고, 어떤 구조로 동작하는지 코드를 기반으로 구체적으로 살펴보겠습니다.

시작하기에 앞서

• GitDiagram - FastAPI

GitDiagram이라는 github repo를 빠르게 시각화하여 이해하는데 도움을 주는 도구가 있습니다.

위 그림은 GitDiagram로 확인한 FastAPI의 Diagram 입니다.

구조 파악을 위해서 diagram 형식으로 보면 파악하기 좋은 것 같아서 가져와 보았습니다.

FastAPI 구조 살펴보기 1에서 설명드렸던 것 처럼 FastAPI는 ASGI 표준 위에서 동작합니다. Diagram을 확인하여 보면 대표적인 ASGI 서버인 Uvicorn이 요청을 받고 FastAPI 애플리케이션에 전달함을 확인할 수 있습니다.

그리고 diagram에 표시되어 있지는 않지만, FastAPI 구조 살펴보기 2, 3에서 설명드렸던 것 처럼 Uvicorn이 전달하는 FastAPI Application은 Starlette 클래스를 상속받아 생성되어 있습니다.

API Documentation은 Documentation 단락에서 OpenAPI Schema와 Swagger UI, ReDoc을 사용함을 파악할 수 있습니다.

이제 Swagger UI, ReDoc 같은 기능이 어디서 어떻게 생성되는지 하나씩 짚어보겠습니다.

1. FastAPI는 언제 문서를 생성하는가?

FastAPI 인스턴스를 생성할 때, 내부적으로 자동으로 문서 생성 경로를 설정합니다. 다음은 FastAPI의 생성자 내부를 살펴본 코드입니다.

fastapi/application.py : _init_

class FastAPI(Starlette):
    def __init__(
      self, 
      ..., 
      openapi_url="/openapi.json",  #L205 ~ 227
      docs_url="/docs",  #L399 ~ 422
      redoc_url="/redoc",  #L423 ~ 446
      ...) -> None:
          ...
          self.openapi_url = openapi_url  #L831
          self.docs_url = docs_url  #L834
          self.redoc_url = redoc_url  #L835
          ...
          self.setup()

• openapi_url: OpenAPI JSON 정의를 제공하는 경로 (/openapi.json)
• docs_url: Swagger UI 경로 (/docs)
• redoc_url: ReDoc 경로 (/redoc)

즉, 앱 인스턴스를 만드는 시점에 문서 경로들이 이미 등록되고 있습니다.

2. FastAPI 문서 관련 라우트는 어디서 추가될까?

문서 관련 경로들을 실제로 추가하는 코드는 setup 함수 안에 있습니다.

fastapi/application.py : setup()

def setup(self) -> None:
    ...
    if self.openapi_url:
        self.add_route(self.openapi_url, openapi, include_in_schema=False)
    if self.openapi_url and self.docs_url:
        self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)
    if self.openapi_url and self.redoc_url:
        self.add_route(self.redoc_url, redoc_html, include_in_schema=False)

• /openapi.json → self.openapi_json 핸들러
• /docs → Swagger UI HTML
• /redoc → ReDoc HTML

이처럼 FastAPI는 문서 관련 핸들러를 기본 라우터에 등록하여, 별도 설정 없이도 자동으로 문서가 제공됩니다.

3. 문서의 핵심: get_openapi

FastAPI의 OpenAPI 스펙(JSON)은 다음 함수에서 동적으로 생성됩니다.

fastapi/openapi/utils.py : get_openapi()

def get_openapi(
    title: str,
    version: str,
    routes: Sequence[BaseRoute],
    ... # 기타 설정들
) -> Dict[str, Any]:
    ...
    info: Dict[str, Any] = {"title": title, "version": version}
    output: Dict[str, Any] = {"openapi": openapi_version, "info": info}
    ...
    for route in routes or []:
        if isinstance(route, routing.APIRoute):
            result = get_openapi_path(
                route=route,
                operation_ids=operation_ids,
                schema_generator=schema_generator,
                model_name_map=model_name_map,
                field_mapping=field_mapping,
                separate_input_output_schemas=separate_input_output_schemas,
            )
            if result:
                path, security_schemes, path_definitions = result
                if path:
                    paths.setdefault(route.path_format, {}).update(path)
                if security_schemes:
                    components.setdefault("securitySchemes", {}).update(
                        security_schemes
                    )
                if path_definitions:
                    definitions.update(path_definitions)

• FastAPI는 모든 APIRoute를 탐색하여 OpenAPI의 paths에 자동으로 추가합니다.
• response_model, status_code, dependencies, summary 등도 이 과정에서 함께 문서화됩니다.

결과적으로 get_openapi()는 FastAPI 라우터에서 정의된 모든 정보를 수집해 OpenAPI 표준 JSON으로 변환합니다.

4. Swagger / ReDoc UI는 어떻게 동작하나?

Swagger와 ReDoc은 각각 swagger_ui_html, redoc_html 메서드를 통해 HTML이 제공됩니다.

fastapi/application.py : swagger_ui_html()

async def swagger_ui_html(req: Request) -> HTMLResponse:
    return get_swagger_ui_html(
        openapi_url=openapi_url,
        title=f"{self.title} - Swagger UI",
        ...
    )

fastapi/openapi/utils.py : get_swagger_ui_html()

def get_swagger_ui_html(...) -> HTMLResponse:
    html = f"""
    <!DOCTYPE html>
    <html>
    <head>
        ...
        <<script src="{swagger_js_url}"></script>
        ...
    </head>
    <body>
        <div id="swagger-ui"></div>
        <script>
        const ui = SwaggerUIBundle({{
            url: "{openapi_url}",
            ...
        }})
        </script>
    </body>
    </html>
    """
    return HTMLResponse(html)

이 코드를 통해 FastAPI는 Swagger UI와 ReDoc의 CDN을 포함한 HTML을 동적으로 렌더링하여 /docs, /redoc에서 제공하게 됩니다.

5. 커스터마이징 가능한가?

물론 가능합니다. FastAPI의 생성자에서 다음과 같은 인자들을 조정할 수 있습니다:

app = FastAPI(
    docs_url="/swagger",
    redoc_url=None,
    openapi_url="/custom-openapi.json"
)

이렇게 하면:

• Swagger UI: /swagger에서 접근 가능
• ReDoc: 비활성화
• OpenAPI JSON: /custom-openapi.json에서 제공

또한, app.openapi_schema에 접근하거나 app.openapi = custom_openapi로 오버라이딩하면 문서 스펙 자체도 직접 커스터마이징할 수 있습니다.

마치며

이번 글에서는 FastAPI가 API 문서를 어떻게 처리하고 자동으로 생성하는지를 GitHub 코드를 바탕으로 살펴보았습니다. 정리하면 다음과 같습니다.

• 문서는 FastAPI 인스턴스 생성 시 자동으로 등록
• /docs, /redoc, /openapi.json은 기본 경로로 설정됨
• get_openapi() 함수가 핵심 로직
• Swagger UI, ReDoc은 각각 HTML을 렌더링하여 UI 제공

오늘은 아래 글을 참고하여 작성했습니다:

• FastAPI GitHub 코드
• GitDiagram

1. GitLab Access Token 발급하기

Jenkins에서 GitLab 저장소에 접근하려면 먼저 GitLab에서 Access Token을 생성해야 합니다.

1. GitLab 우측 상단 프로필 > User Settings 클릭
2. 좌측 메뉴에서 Access Tokens 선택
3. 이름, 만료일, 권한 범위(예: read_repository, write_repository)를 지정한 뒤 토큰 발급

2. Jenkins에서 GitLab 플러그인 설치

Jenkins에서 GitLab 저장소와의 통신을 위해 GitLab Plugin을 설치해야 합니다.

• Jenkins 메인 화면 > Jenkins 관리 > Plugins
• 사진의 GitLab 플러그인을 찾아 설치

3. GitLab Access Token 등록

Jenkins가 GitLab 저장소에 접근할 수 있도록 Access Token을 등록합니다.

• Jenkins 메인 화면 > Jenkins 관리 > Credentials > Add Credentials
• Kind 에서 GitLab API token 선택
• API Token에 앞서 발급한 토큰 입력
• ID에는 jenkins에서 사용할 id 입력
• Description에는 추가 설명 입력

4. GitLab 연결 테스트

등록한 Gitlab 정보로 정상적으로 연결이 되는지 확인해봅니다.

• Jenkins 메인 화면 > Jenkins 관리 > System
• connection 이름과 gitlab 도메인을 입력
• Test Connection > Success 시 > Save

5. Jenkins Credentials에 GitLab 인증 정보를 등록

파이프라인에서 인증 정보를 사용할 수 있도록 Jenkins Credentials에 등록합니다.

• Jenkins 메인 화면 > Jenkins 관리 > Credentials > (global scope 선택)
• Add Credentials 클릭
• Kind는 Username with password 선택
  • Username: GitLab 사용자명
  • Password: 발급한 Personal Access Token
  • ID: jenkins에서 사용할 id 입력
• 입력 후 Create 선택

6. 테스트 파이프라인 생성

정상적으로 연동되었는지 확인하기 위해 새로운 파이프라인 Job을 생성합니다.

1. Jenkins 메인 화면 > New Item

2. 이름 설정 후 Pipeline 선택

3. Configuration > Pipeline > Pipeline script 입력

    pipeline {
        agent any
   
        stages {
            stage('Clone') {
                steps {
                    git branch: '${branch}', credentialsId: '${generated credentialId}', url: '${repository address}.git'
                }
            }
        }
    }

4. save 선택

7. 테스트 파이프라인 실행 결과 확인

Job을 실행한 후 아래와 같이 상태와 단계별 진행 현황을 확인할 수 있습니다.

• 실행 후 Status 및 각 Stage 확인
• clone 단계의 체크표시 클릭 → 상세 로그 확인 가능

8. BlueOcean UI로 시각적으로 확인

BlueOcean 플러그인을 설치했다면 파이프라인 실행 내역을 좀 더 직관적으로 확인할 수 있습니다.

마무리

Jenkins에서 GitLab과 연동하고 테스트 파이프라인을 구성하는 과정을 살펴봤습니다. 이후에는 이것을 활용하여 실제 CI/CD 파이프라인을 구성할 수 있습니다.

참고 : Installing Jenkins Docker On Windows

1. Docker 전용 네트워크 만들기

먼저 Jenkins와 Docker가 같은 네트워크에서 통신할 수 있도록 전용 bridge 네트워크를 생성합니다.

docker network create jenkins

이 네트워크는 Jenkins 컨테이너와 Docker-in-Docker 컨테이너 간 통신을 위해 사용됩니다.

2. Docker in Docker(DinD) 방식으로 Jenkins 실행

Jenkins 내에서 Docker 컨테이너를 실행할 수 있게 하기 위해서 Docker-in-Docker(DinD)방식을 사용합니다.

• Docker-in-Docker(DinD)의 필요성
  1. 컨테이너화된 빌드 환경: Jenkins 자체가 컨테이너로 실행될 때, 이 Jenkins 컨테이너 내에서 다른 Docker 컨테이너를 실행하기 위해서는 Docker 엔진이 필요합니다.
  2. 격리된 빌드 환경: 각 빌드 작업이 독립적인 컨테이너에서 실행되어 클린한 환경을 보장합니다.
  3. 파이프라인 자동화: CI/CD 파이프라인에서 애플리케이션을 Docker 이미지로 빌드하고 테스트할 수 있습니다.

docker run --name jenkins-blueocean --restart=on-failure --detach ^
  --network jenkins --env DOCKER_HOST=tcp://docker:2376 ^
  --env DOCKER_CERT_PATH=/certs/client --env DOCKER_TLS_VERIFY=1 ^
  --volume jenkins-data:/var/jenkins_home ^
  --volume jenkins-docker-certs:/certs/client:ro ^
  --publish 8080:8080 --publish 50000:50000 myjenkins-blueocean:2.492.3-1

실행 후, Jenkins와 Docker 간 인증을 위해 필요한 환경 변수도 함께 설정해줍니다.

3. Jenkins Dockerfile 작성

Jenkins 공식 이미지 위에 BlueOcean 및 Docker CLI 플러그인을 설치한 커스텀 이미지를 만듭니다.

FROM jenkins/jenkins:2.492.3-jdk17
USER root
RUN apt-get update && apt-get install -y lsb-release
RUN curl -fsSLo /usr/share/keyrings/docker-archive-keyring.asc \
  https://download.docker.com/linux/debian/gpg
RUN echo "deb [arch=$(dpkg --print-architecture) \
  signed-by=/usr/share/keyrings/docker-archive-keyring.asc] \
  https://download.docker.com/linux/debian \
  $(lsb_release -cs) stable" > /etc/apt/sources.list.d/docker.list
RUN apt-get update && apt-get install -y docker-ce-cli
USER jenkins
RUN jenkins-plugin-cli --plugins "blueocean docker-workflow"

4. 커스텀 이미지 빌드하기

위 Dockerfile을 기반으로 Jenkins BlueOcean 이미지를 생성합니다.

docker build -t myjenkins-blueocean:2.492.3-1 .

이 이미지에는 docker-workflow 플러그인이 포함되어 있어, Jenkins 내에서 Docker를 제어할 수 있습니다.

5. Jenkins BlueOcean 컨테이너 실행

이제 커스텀 이미지로 Jenkins 컨테이너를 실행합니다.

docker run --name jenkins-blueocean --restart=on-failure --detach ^
  --network jenkins --env DOCKER_HOST=tcp://docker:2376 ^
  --env DOCKER_CERT_PATH=/certs/client --env DOCKER_TLS_VERIFY=1 ^
  --volume jenkins-data:/var/jenkins_home ^
  --volume jenkins-docker-certs:/certs/client:ro ^
  --publish 8080:8080 --publish 50000:50000 myjenkins-blueocean:2.492.3-1

브라우저에서 http://localhost:8080 접속하면 Jenkins 설치 마법사가 시작됩니다.

6. 설치 마법사 진행

Jenkins 첫 실행 시 아래 경로의 비밀번호가 필요합니다.

docker exec -it jenkins-blueocean /bin/sh
vi /var/jenkins_home/secrets/initialAdminPassword

해당 파일을 열어 비밀번호를 복사하고 Jenkins 설치를 계속 진행하세요.

1. 플러그인 설치: ‘Install suggested plugins’ 선택
2. 관리자 계정 생성
3. Jenkins URL 설정: 우선 Not now 를 선택하고 진행하였습니다.
4. Jenkins 접속

이제 Jenkins에서 파이프라인을 시각적으로 구성하고 실행할 준비가 완료되었습니다!

마무리

이제 Jenkins와 BlueOcean을 로컬 Docker 환경에서 사용할 수 있습니다. 설정 이후에는 GitHub 또는 GitLab과 연동해서 실제 CI/CD 파이프라인을 구성할 수 있습니다.

단순히 Frontend 코드에서 url 끝에 / 제거하여 해결하였지만, 이 현상을 더 자세히 살펴보고 블로그 글로 작성해보았습니다.

위 상황을 RESTfulAPI 관점에서 Trailing Slash 라고 한다고 합니다.

FastAPI가 Trailing Slash를 마주한 상황에서 내부적으로 어떻게 동작하는지 자세히 살펴보도록 하겠습니다.

1. 요청 URL 차이에 따른 FastAPI의 처리 방식

FastAPI에서 아래 두 가지 URL 요청 방식은 언뜻 보기에 비슷해 보이지만, 실제 처리 과정에서 서로 다른 엔드포인트로 인식합니다.

GET /user/check?email={user_email}

GET /user/check/?email={user_email}

두 요청 모두 정상적인 요청이지만, FastAPI에서는 URL의 끝에 슬래시(/)를 붙이는 경우와 붙이지 않는 경우를 내부적으로 다른 route로 간주합니다.

• /check와 /check/는 다른 엔드포인트로 인식됩니다.
• FastAPI는 기본적으로 이 두 URL을 구분하여 처리합니다.
  • /check로 라우트가 정의되어 있으면 /check/ 요청 시 FastAPI는 307 Temporary Redirect를 반환해 브라우저를 /check로 리다이렉트합니다.
  • 마찬가지로 /check/로 정의되었다면 /check 요청 시 자동으로 /check/로 리다이렉트(307)합니다.

이러한 동작으로 인해 의도하지 않은 307 Redirect가 발생할 가능성이 있습니다.

2. Starlette의 리다이렉트 처리 메커니즘 살펴보기

이전에 작성했던 FastAPI 구조 살펴보기 에서 말씀드렸다시피, FastAPI는 Starlette 프레임워크를 기반으로 구축되었습니다. 실제로 URL 끝의 슬래시 여부에 따른 리다이렉트 처리는 Starlette의 라우팅 시스템에서 이루어집니다. 아래는 Starlette의 리다이렉트 처리 코드입니다.

# Starlette의 routing.py 파일에서 발췌
from starlette.responses import RedirectResponse

class Router:
    # ...
    async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
        # ...
        route_path = get_route_path(scope)
        if scope["type"] == "http" and self.redirect_slashes and route_path != "/":
            redirect_scope = dict(scope)
            if route_path.endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match != Match.NONE:
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return
         # ...
    # ...

2-1. 코드의 전반적인 목적

이 코드는 Starlette라는 Python 웹 프레임워크의 라우터(Router)에서 제공하는 기능 중 하나로, 사용자가 URL 경로 끝에 슬래시(/)를 붙이거나 빼고 접속했을 때 적절한 주소로 리디렉션해주는 기능입니다.

예를 들어

• 사용자가 /user 주소를 접속했는데, 실제 등록된 경로가 /user/라면, 이 코드는 자동으로 /user/로 리디렉션을 해줍니다.
• 반대로 사용자가 /user/ 주소를 접속했는데, 실제 등록된 경로가 /user라면 /user로 리디렉션을 해줍니다.

이 기능은 URL의 일관성을 유지하고, 동일한 페이지가 여러 URL로 접근되는 것을 방지하는 데 도움을 줍니다.

2-2. 단계별 상세 분석

1. 요청의 타입과 조건 확인

if scope["type"] == "http" and self.redirect_slashes and route_path != "/":

• 들어온 요청의 타입이 http인지 확인합니다. (웹 브라우저를 통해 들어온 일반 웹 요청)
• 설정에서 슬래시를 자동으로 리디렉션하도록(redirect_slashes=True) 되어 있는지 확인합니다.(기본값이 true)
• 요청받은 경로(route_path)가 루트 경로(/)가 아닌 경우에만 진행합니다. (루트 경로는 / 하나만 있는 것이기 때문에 리디렉션이 불필요함)

2. 리디렉션할 주소 만들기

redirect_scope = dict(scope)
if route_path.endswith("/"):
    redirect_scope["path"] = redirect_scope["path"].rstrip("/")
else:
    redirect_scope["path"] = redirect_scope["path"] + "/"

• 기존 요청(scope) 정보를 복사하여 새로 리디렉션할 주소를 만듭니다.
• 만약 현재 요청한 경로가 슬래시(/)로 끝난다면, 슬래시를 제거한 주소로 바꿉니다. 예: /user/ → /user
• 만약 현재 요청한 경로가 슬래시로 끝나지 않는다면, 슬래시를 추가한 주소로 바꿉니다. 예: /user → /user/

3. 수정한 주소가 실제 라우터에 등록된 경로인지 검사

for route in self.routes:
    match, child_scope = route.matches(redirect_scope)
    if match != Match.NONE:

• 새롭게 수정된 주소(redirect_scope)가 현재 라우터에 등록된 경로인지 확인합니다.
• route.matches()는 해당 경로와 현재 요청 주소가 일치하는지 체크하는 메서드입니다.
• 만약 일치하는 경로가 하나라도 발견되면, 다음 단계(리디렉션 수행)로 넘어갑니다.

4. 리디렉션 URL을 만들고 리디렉션 수행하기

redirect_url = URL(scope=redirect_scope)
response = RedirectResponse(url=str(redirect_url))
await response(scope, receive, send)
return

• 실제로 리디렉션할 전체 URL을 URL(scope=redirect_scope)를 통해 구성합니다.
• RedirectResponse 클래스를 통해 브라우저에게 리디렉션 명령을 보낼 준비를 합니다. (기본적으로 HTTP 307 또는 308 리디렉션 코드 사용)
• 만들어진 응답(response)을 브라우저에 보내서 사용자가 브라우저에서 올바른 주소로 다시 접속하게 만듭니다.
• 리디렉션을 수행하고 바로 종료하여, 원래의 잘못된 URL 처리가 더 이상 진행되지 않도록 합니다.

2-3. 동작 시나리오로 정리

예시 상황 1: /user로 접근했지만, 실제 URL은 /user/인 경우

• 사용자가 /user로 요청 → 리디렉션 URL을 /user/로 생성 → 이 경로가 실제 등록된 URL인지 확인 → 일치한다면 리디렉션 응답 발송 → 사용자는 /user/로 다시 접속됨.

예시 상황 2: /user/로 접근했지만, 실제 URL은 /user인 경우

• 사용자가 /user/로 요청 → 리디렉션 URL을 /user로 생성 → 이 경로가 실제 등록된 URL인지 확인 → 일치한다면 리디렉션 응답 발송 → 사용자는 /user로 다시 접속됨.

즉, URL의 슬래시(/) 유무에 따라 사용자를 올바른 주소로 안내하며, 이 과정에서 리다이렉트가 발생합니다.

3. 문제 상황

아래처럼 Route를 작성한 상황을 가정해봅시다.

# routes/user.py
@app.get("/check")
def get_check_email(email: str):
    # 함수 내용

이 경우 /check/와 같이 뒤에 슬래시가 추가된 URL로 요청을 보내면 자동으로 307 Redirect가 발생합니다. 실제 FastAPI에서는 아래와 같은 과정이 일어납니다.

1. 브라우저가 /user/check/?email={user_email} 요청
2. Starlette의 라우팅 시스템이 이 경로를 처리하며 등록된 라우트를 확인
3. /check/가 아닌 /check만 등록되어 있음을 확인
4. 자동으로 /user/check?email={user_email}으로 307 Temporary Redirect 응답 생성
5. 이 과정에서 HTTPS를 사용하지 않고 HTTP로 리다이렉트가 이루어지면 Mixed Content(혼합 콘텐츠) 문제가 발생할 수 있음

로컬 HTTP 환경과 배포 HTTPS 환경의 차이

로컬 개발 환경 (HTTP)

로컬에서 HTTP로 테스트할 때는 리다이렉트가 발생해도 문제가 없었습니다. 그 이유는 아래와 같습니다.

1. 로컬 환경에서는 모든 요청이 HTTP로 이루어지기 때문에 프로토콜 불일치가 발생하지 않음
2. 개발 서버는 보통 같은 도메인에서 실행되므로 CORS나 혼합 콘텐츠 문제가 발생하지 않음
3. 브라우저의 로컬 개발 환경에서는 보안 제한이 덜 엄격하게 적용되는 경우가 있음

예를 들어, 로컬에서 다음과 같은 요청이 발생했을 때

GET http://localhost:8000/user/check/?email={user_email}

리다이렉트 후

GET http://localhost:8000/user/check?email={user_email}

두 URL 모두 HTTP 프로토콜을 사용하므로 문제가 없습니다.

배포 환경 (HTTPS)

하지만 실제 서버에 배포하고 HTTPS를 사용하게 되면 상황이 달라집니다.

1. 기본적으로 보안 연결을 사용하므로 브라우저가 적용하는 보안 정책이 엄격해짐
2. 리다이렉트 과정에서 프로토콜이 변경될 수 있음
3. 설정이 잘못되면 HTTPS → HTTP로 리다이렉트될 가능성이 있음

예를 들어, 배포 환경에서 다음과 같은 요청이 발생했을 때:

GET https://example.com/user/check/?email={user_email}

리다이렉트 후 프로토콜이 변경

GET http://example.com/user/check?email={user_email}

이 경우 Mixed content 문제가 발생합니다. 현대 브라우저는 HTTPS 페이지에서 HTTP 리소스를 로드하는 것을 차단하므로, 리다이렉트 후 요청이 실패하게 됩니다.

4. 해결 방법

요청 URL 끝에 슬래시(/)가 추가된 경우, FastAPI는 기본적으로 307 Temporary Redirect를 발생시켜 정의된 라우트로 자동으로 리다이렉트합니다. 이를 방지하기 위한 방법은 다음과 같습니다.

4-1. URL을 정확히 일치시켜 요청하도록 클라이언트 수정

• 라우트 URL과 정확히 일치시켜 호출하는 것이 가장 좋습니다.

  GET /user/check?email={user_email}

4-2. FastAPI의 라우트 설정에 양쪽 경우 모두 처리하도록 설정

• URL 끝 슬래시를 허용하도록 두 가지 경로를 모두 처리

@router.get("/check", include_in_schema=True)
@router.get("/check/", include_in_schema=False)
def get_check_email(email: str):
    # 함수 내용

4-3. FastAPI의 리다이렉트 기능 비활성화

• FastAPI 인스턴스 생성 시 redirect_slashes 매개변수를 False로 설정

from fastapi import FastAPI

# 슬래시 리다이렉트 비활성화
app = FastAPI(redirect_slashes=False)

이 방법을 사용하면 슬래시 차이에 따른 리다이렉트가 발생하지 않지만, 정의되지 않은 경로에 대해서는 404 에러가 발생합니다.

문제 상황

아래와 같이, 동일한 내용의 document를 여러 번 insert 하려는 코드가 있을 때, 어떠한 문제가 발생할 수 있을까요?

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]
collection = db["mycollection"]

document = {
    "data": "example"
}
documents_to_insert = [document] * 5

result = collection.insert_many(documents_to_insert)

위 코드를 실행하면 다음과 같은 오류가 발생할 수 있습니다.

# pymongo.errors.BulkWriteError: batch op errors occurred, full error:
{
    'writeErrors': [
        {
            'index': 1,  # 두 번째 문서(index=1)에서 오류 발생
            'code': 11000,  # MongoDB의 Duplicate Key Error 코드
            'keyPattern': {'_id': 1},  # 중복된 키가 '_id' 필드임을 의미
            'keyValue': {'_id': ObjectId('67c7f9e56674ed3dfa230c9f')},  # 충돌된 ObjectId 값
            'errmsg': "E11000 duplicate key error collection: OIT.slot index: _id_ dup key: { _id: ObjectId('67c7f9e56674ed3dfa230c9f') }",
            'op': {  # 실패한 문서의 내용
                "data": "example",
                '_id': ObjectId('67c7f9e56674ed3dfa230c9f')
            }
        }
    ],
    'writeConcernErrors': [],
    'nInserted': 1,  # 1개의 문서만 삽입됨
    'nUpserted': 0,
    'nMatched': 0,
    'nModified': 0,
    'nRemoved': 0,
    'upserted': []
}

원인 분석

• 정확한 원인 분석을 위하여, 우선 문제 상황 코드를 살펴본 후에 insert_many 코드(pymongo → collection → insert_many)를 살펴보겠습니다.

  • 문제 상황 코드 확인

    • mongodb collection을 선언하고, document를 삽입하는 부분을 제외하고 볼 부분은 다음 부분 밖에 없습니다.

        document = {
            "data": "example"
        }
        documents_to_insert = [document] * 5 

    • documents_to_insert라는 리스트를 생성하는데, document를 5번 반복해서 리스트에 넣습니다.

    • 이때 리스트 내부의 요소들은 모두 동일한 객체에 대한 참조(레퍼런스)입니다.

    • 즉, documents_to_insert리스트의 모든 요소는 동일한 document 객체를 가리키게 됩니다.

    • 이 상황을 파악하고, insert_many가 어떻게 동작하는지 확인해봅시다.

  • insert_many 코드 확인

    • PyMongo github을 확인해보니 collection을 처리하는 동기와 비동기, 두 가지 버전을 확인할 수 있었습니다. 확인해보니 코드 구현은 동일하게 되어 있으므로 동기 버전의 코드만 가지고 확인을 해보겠습니다.

      • sync : https://github.com/mongodb/mongo-python-driver/blob/baf0344446ecfd4cbab118038165fffb61f020d4/pymongo/synchronous/collection.py#L904
      • async : https://github.com/mongodb/mongo-python-driver/blob/baf0344446ecfd4cbab118038165fffb61f020d4/pymongo/asynchronous/collection.py#L905

    • _id를 처리하는 부분의 코드입니다.

        # L960
        inserted_ids: list[ObjectId] = []
      
        def gen() -> Iterator[tuple[int, Mapping[str, Any]]]:
            """A generator that validates documents and handles _ids."""
            for document in documents:
                common.validate_is_document_type("document", document)
                if not isinstance(document, RawBSONDocument):
                    if "_id" not in document:
                        document["_id"] = ObjectId()  # type: ignore[index]
                    inserted_ids.append(document["_id"])
                yield (message._INSERT, document)

      • inserted_ids 리스트를 만들어, 삽입된 문서의 _id를 저장.
      • 문서 유효성을 검사 (common.validate_is_document_type(...)).
      • 문서에 _id가 없으면 자동으로 ObjectId()를 생성해서 추가.
      • yield를 통해 삽입할 데이터를 제너레이터(generator)로 반환.

  • 종합 해석

    1. 리스트 내부 요소가 동일한 객체를 참조

       document = {"data": "example"}
       documents_to_insert = [document] * 5

       위 코드에서 documents_to_insert 리스트의 모든 요소는 동일한 객체를 참조합니다. 즉, 리스트에 5개의 개별 객체가 있는 것이 아니라 하나의 document 객체가 5번 반복된 상태입니다.

    2. insert_many()의 동작 방식

       PyMongo는 _id 필드가 없는 문서에 대해 자동으로 ObjectId()를 생성합니다. 하지만 모든 요소가 동일한 객체를 가리키므로 _id가 하나만 생성되고, 이 _id가 리스트의 모든 요소에 적용됩니다.

       [{"data": "example", "_id": ObjectId('604a8f032d23d2e9c3a2c1e1')},
       {"data": "example", "_id": ObjectId('604a8f032d23d2e9c3a2c1e1')},
       {"data": "example", "_id": ObjectId('604a8f032d23d2e9c3a2c1e1')},
       {"data": "example", "_id": ObjectId('604a8f032d23d2e9c3a2c1e1')},
       {"data": "example", "_id": ObjectId('604a8f032d23d2e9c3a2c1e1')}]

       결과적으로 _id가 중복되어 Duplicate Key Error(11000)가 발생합니다.

해결 방법

얕은 복사가 문제가 된다는 것을 알았으니 해결 방법은 간단합니다.

1. dict()를 사용한 깊은 복사(Deep Copy)

dict(document)를 사용하면 개별 객체를 새로 생성할 수 있습니다.

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]
collection = db["mycollection"]

document = {"data": "example"}

documents_to_insert = [dict(document) for _ in range(5)]

result = collection.insert_many(documents_to_insert)
print("Inserted IDs:", result.inserted_ids)

이 방식은 _id가 각 문서에 대해 개별적으로 생성되도록 보장하여 중복 오류를 방지할 수 있습니다.

2. _id 필드를 명시적으로 추가

각 문서에 수동으로 _id를 할당하면 충돌을 방지할 수 있습니다.

from bson import ObjectId

documents_to_insert = [{"_id": ObjectId(), "data": "example"} for _ in range(5)]
result = collection.insert_many(documents_to_insert)

이 방법을 사용하면 PyMongo의 자동 _id 생성이 아닌, 우리가 직접 지정한 _id가 사용됩니다.

결론

PyMongo에서 insert_many()를 사용할 때, 같은 객체를 복제하는 방식은 _id 중복 오류를 유발할 수 있습니다. 이를 방지하기 위해서는 아래와 같은 방법을 사용하면 됩니다.

1. dict()를 사용하여 개별 객체를 생성
2. _id 필드를 직접 추가

PyMongo의 내부 동작을 이해하면, 예상치 못한 오류를 줄이고 안전하게 데이터를 삽입할 수 있습니다. 얕은 복사(Shallow Copy)를 조심합시다!

Python의 dataclass는 데이터 중심의 클래스를 간단하게 정의할 수 있도록 도와주는 기능입니다. 일반적으로 데이터를 저장하는 용도로 클래스를 사용할 때, __init__, __repr__, __eq__ 등의 메서드를 일일이 구현해야 하지만, dataclass를 사용하면 이러한 작업을 자동화할 수 있습니다.

Python 3.7에서 도입되었으며, dataclasses 모듈에서 제공됩니다.

2. Dataclass 사용법

기본 사용법

from dataclasses import dataclass

@dataclass
class Person:
    name: str
    age: int

p1 = Person(name="Alice", age=30)
print(p1)  # Person(name='Alice', age=30)

위 코드에서 dataclass 데코레이터를 사용하여 Person 클래스를 정의하면, 자동으로 __init__, __repr__, __eq__ 등의 메서드가 생성됩니다.

기본값 설정

필드에 기본값을 지정할 수도 있습니다.

@dataclass
class Person:
    name: str
    age: int = 25  # 기본값 설정

p1 = Person(name="Bob")
print(p1)  # Person(name='Bob', age=25)

필드 속성 설정

dataclass에서는 field()를 사용하여 필드의 동작을 보다 정교하게 제어할 수 있습니다.

from dataclasses import dataclass, field

@dataclass
class Person:
    name: str
    age: int = field(default=25, metadata={"info": "나이 정보"})
    friends: list = field(default_factory=list)  # 기본값으로 빈 리스트 지정

p1 = Person(name="Charlie")
p1.friends.append("David")
print(p1)  # Person(name='Charlie', age=25, friends=['David'])

비교 및 해싱 가능 여부 설정

dataclass는 기본적으로 객체를 비교할 수 있도록 __eq__ 메서드를 자동으로 생성합니다. 하지만 특정 설정을 변경할 수도 있습니다.

@dataclass(order=True)
class User:
    id: int
    username: str

u1 = User(1, "Alice")
u2 = User(2, "Bob")
print(u1 < u2)  # True (id를 기준으로 비교)

3. Dataclass의 장단점

장점

1. 코드 간결성: __init__, __repr__, __eq__ 등의 메서드를 자동 생성하여 코드량이 줄어듭니다.
2. 유지보수 용이: 데이터 중심 클래스의 정의가 간단해지므로 유지보수가 쉬워집니다.
3. 기본값 및 필드 속성 제어 가능: field()를 이용하여 다양한 속성을 설정할 수 있습니다.
4. 비교 및 정렬 기능 지원: order=True를 설정하면 객체 비교가 가능합니다.

단점

1. 동적 속성 추가 제한: dataclass로 생성된 객체는 기본적으로 slots을 사용하지 않지만, frozen=True를 설정하면 속성을 동적으로 추가할 수 없습니다.
2. 가변 타입의 기본값 문제: 리스트나 딕셔너리 같은 가변 객체를 기본값으로 설정하면 예상치 못한 동작이 발생할 수 있습니다. default_factory를 사용하는 것이 안전합니다.
3. 상속 시 제약: 여러 개의 부모 클래스를 상속할 경우 dataclass의 동작이 복잡해질 수 있습니다.
4. 성능 오버헤드: 작은 규모의 클래스에서는 dataclass가 필요 이상으로 많은 기능을 제공하여 성능 저하를 초래할 수 있습니다.
5. 객체 직렬화 제한: dataclass 객체는 pickle과 같은 일부 직렬화 라이브러리와 완전히 호환되지 않을 수 있습니다.

4. Dataclass 활용 예제

JSON 변환

dataclass를 사용하면 객체를 JSON으로 쉽게 변환할 수 있습니다.

import json
from dataclasses import dataclass, asdict

@dataclass
class Product:
    id: int
    name: str
    price: float

p1 = Product(1, "Laptop", 999.99)
p1_json = json.dumps(asdict(p1))  # JSON 변환
print(p1_json)  # {"id": 1, "name": "Laptop", "price": 999.99}

데이터 검증과 활용

dataclass를 사용하여 입력 데이터를 검증하는 것도 가능합니다.

@dataclass
class Temperature:
    celsius: float

    @property
    def fahrenheit(self) -> float:
        return self.celsius * 9 / 5 + 32

t = Temperature(25)
print(t.fahrenheit)  # 77.0

상속을 활용한 확장

Dataclass는 상속을 통해 확장할 수 있습니다.

@dataclass
class Employee:
    name: str
    department: str

@dataclass
class Manager(Employee):
    team_size: int

m1 = Manager(name="Eve", department="IT", team_size=10)
print(m1)  # Manager(name='Eve', department='IT', team_size=10)

데이터 클래스의 불변성 유지

frozen=True 옵션을 사용하면 객체가 불변(immutable)하게 설정됩니다.

@dataclass(frozen=True)
class Config:
    setting: str

c = Config(setting="dark mode")
# c.setting = "light mode"  # AttributeError 발생

자동 ID 생성

객체마다 자동으로 고유한 ID를 부여할 수도 있습니다.

import uuid
from dataclasses import dataclass, field

@dataclass
class Task:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    description: str

new_task = Task(description="Complete project")
print(new_task)  # Task(id='...', description='Complete project')

그러나 CRA가 deprecated 된지도 꽤 되었습니다.

React 공식문서를 확인해보면 풀스택 프레임워크로 사용할 때는 Next.js, Remix, Gatsby, Expo를 사용하는 방법을 알려줍니다.

그러나 React를 프레임워크 없이 사용하기 위해서 Vite(비트)를 사용하여 리액트를 시작하는 방법을 안내해보고자 합니다.

Vite 공식문서를 확인해보면 Vue.js를 시작하는 방법을 알려줍니다.

이는 Vite가 Vue 3와 함께 개발된 툴로써 Vue.js를 시작하는 방법을 안내하는 것으로 생각합니다. 그러나 Vite는 Vue.js를 비롯하여 React, Svelte, SolidJS 등 다양한 템플릿을 지원합니다.

그러면 Vite로 리액트 프로젝트 시작하는 방법을 안내하겠습니다.

Vite란?

Vite는 빠르고 간편한 프론트엔드 빌드 도구로, 개발 환경에서 즉각적인 HMR(Hot Module Replacement)을 제공하여 개발 속도를 향상시킵니다. Webpack보다 훨씬 가볍고 빠르게 리액트 프로젝트를 시작할 수 있습니다.

Vite로 리액트 프로젝트 생성하기

1. Node.js 설치 확인

Vite를 사용하려면 Node.js가 필요합니다. 다음 명령어를 입력하여 현재 설치된 Node.js 버전을 확인해보겠습니다.

node -v

만약 설치되어 있지 않다면 Node.js 공식 사이트에서 LTS 버전을 다운로드하여 설치하세요.

저는 기존에 설치했었던 18.12.1 버전을 사용해보겠습니다.

2. Vite로 프로젝트 생성

터미널을 열고 다음 명령어를 실행하여 새로운 리액트 프로젝트를 생성합니다.

npm create vite@latest "프로젝트 이름" -- --template react

위 명령어에서 "프로젝트 이름" 부분에 원하는 이름으로 변경하시면 프로젝트 폴더 이름으로 설정됩니다.

그리고 npm v7 이상에서는 -- 를 반드시 붙여야 템플릿이 적용됩니다.

3. 프로젝트 폴더로 이동

cd "프로젝트 이름"

4. 패키지 설치

npm install

패키지 설치까지 완료하고, 프로젝트를 확인하기 쉽게 code . 을 사용하여 vscode를 실행해주었습니다. vscode 단축키 control + shift + ₩ 을 사용하면 터미널을 실행할 수 있습니다.

5. 개발 서버 실행

다음 명령어를 실행하면 로컬 개발 서버가 시작됩니다.

npm run dev

이제 브라우저에서 http://localhost:5173 주소로 접속하여 리액트 애플리케이션을 확인할 수 있습니다.

프로젝트 구조

제가 Vite로 생성된 프로젝트의 기본 폴더 구조는 다음과 같습니다.

vite-react-test/
├── node_modules/
├── public/
│   ├── vite.svg
├── src/
│   ├── assets/
│   │   ├── react.svg
│   ├── App.css
│   ├── App.jsx
│   ├── index.css
│   ├── main.jsx
├── .gitignore
├── eslint.config.js
├── index.html
├── package-lock.json
├── package.json
├── README.md
├── vite.config.js

각 파일의 역할은 다음과 같습니다.

• src/ 폴더: 리액트 컴포넌트가 위치하는 곳
• public/ 폴더: 정적 파일이 위치하는 곳
• index.html: 애플리케이션의 진입점
• vite.config.js: Vite 관련 설정 파일
• eslint.config.js: ESLint 관련 설정 파일

결론

deprecated된 CRA를 대신해 Vite를 사용하여 리액트 프로젝트를 시작하는 방법을 알아보았습니다. 이 다음 단계로는 상태 관리 라이브러리, 스타일링 도구 등을 추가하여 프로젝트를 확장할 수 있습니다.

• 공식문서 : https://peps.python.org/pep-0572/

바다코끼리 연산자란 무엇인가요?

간단히 말하면, 바다코끼리 연산자는 "값을 변수에 할당하면서 표현식 내부에서 그 값을 사용할 수 있도록" 해줍니다. 일반적으로 파이썬에서는 값을 변수에 할당할 때와 그 값을 사용하는 코드를 따로 작성해야 했습니다. 하지만 바다코끼리 연산자는 이 과정을 한 번에 처리할 수 있게 해줍니다.

문법은 다음과 같습니다:

variable := expression

이제 예시를 통해 어떻게 사용되는지 알아보겠습니다.

예제 1: while 루프 간결화

보통 파일에서 데이터를 읽을 때, 아래와 같은 코드를 자주 보셨을 것 같습니다.

# 기존 방식
line = file.readline()
while line:
    print(line.strip())
    line = file.readline()

여기서 같은 file.readline() 호출이 두 번 반복되는 게 조금 거슬리지 않나요? 바다코끼리 연산자를 사용하면 이렇게 간결하게 바꿀 수 있습니다.

# 바다코끼리 연산자 사용
while (line := file.readline()):
    print(line.strip())

line 변수에 값을 할당하면서 동시에 while 조건식에서 그 값을 평가하는 방식입니다. 코드가 짧아졌을 뿐만 아니라 반복 호출을 줄여서 성능상으로도 조금 더 유리할 수 있습니다.

예제 2: 리스트 컴프리헨션에서의 활용

리스트 컴프리헨션에서도 바다코끼리 연산자를 유용하게 사용할 수 있습니다. 예를 들어, 특정 조건에 따라 필터링된 값을 처리하면서 원본 값을 저장하고 싶다면 다음과 같이 작성할 수 있습니다.

# 기존 방식
results = []
for item in data:
    processed = complex_function(item)
    if processed > threshold:
        results.append(processed)

바다코끼리 연산자를 활용하면 이렇게 줄일 수 있습니다.

# 바다코끼리 연산자 사용
results = [processed for item in data if (processed := complex_function(item)) > threshold]

코드가 더 깔끔해지고, 한눈에 "조건과 처리"를 볼 수 있어 가독성이 향상됩니다.

예제 3: 조건문에서의 활용

조건문에서 값 계산과 비교를 동시에 처리할 수도 있습니다. 예를 들어, 사용자의 입력값을 처리한다고 가정해 보겠습니다.

# 기존 방식
user_input = input("숫자를 입력하세요: ")
if user_input.isdigit():
    number = int(user_input)
    if number > 10:
        print("10보다 큰 숫자입니다.")

이 코드 역시 바다코끼리 연산자를 사용하면 이렇게 바뀝니다.

# 바다코끼리 연산자 사용
if (user_input := input("숫자를 입력하세요: ")).isdigit() and (number := int(user_input)) > 10:
    print("10보다 큰 숫자입니다.")

값을 변수에 할당하는 과정이 조건식 내부에서 자연스럽게 처리되니, 코드가 더 직관적이지 않나요?

주의할 점

1. 가독성: 바다코끼리 연산자가 코드의 간결성을 높여주는 건 사실이지만, 남용하면 오히려 가독성을 해칠 수 있습니다. 따라서 "이 코드가 다른 개발자가 쉽게 이해할 수 있을까?"를 생각해보고 적용하는 것이 좋겠습니다.
2. 파이썬 3.8 이상: 이 연산자는 파이썬 3.8에서 추가되었으니, 프로젝트 환경이 이를 지원하는지 확인하세요.

결론

바다코끼리 연산자는 간단해 보이지만 코드의 효율성과 가독성을 동시에 높일 수 있는 강력한 도구입니다. 하지만 모든 도구가 그렇듯, 적재적소에 적절히 사용하는 것이 중요합니다.

1. MongoDB Injection이란?

MongoDB 인젝션(MongoDB Injection)은 공격자가 악의적인 입력을 통해 MongoDB 쿼리를 조작함으로써, 데이터베이스의 기밀 정보에 접근하거나 데이터를 변조하는 공격 기법입니다. 이는 SQL 인젝션과 유사한 원리를 가지고 있으며, 일반적으로 사용자 입력값이 제대로 검증되지 않을 때 발생합니다.

2. MongoDB Injection의 원리

MongoDB는 JSON과 유사한 BSON(Binary JSON)을 기반으로 데이터를 처리합니다. 클라이언트에서 전달된 JSON 형식의 데이터를 데이터베이스 쿼리에 직접 사용하면, 악성 입력이 쿼리를 변경하거나 조작할 수 있습니다.

1. 취약점이 발생하는 상황

• 사용자 입력값을 검증하지 않고 쿼리에 직접 포함시키는 경우.
• 동적 쿼리 작성 시, 입력값을 JSON 객체로 변환할 때 공격 가능성이 존재.

# 취약한 코드 예시
user_input = {"username": "admin", "password": {"$ne": ""}}
db.users.find(user_input)

위 코드에서 {"$ne": ""}는 MongoDB의 $ne 연산자를 활용하여 특정 필드를 "비어 있지 않다"고 평가합니다. 이는 인증 없이도 로그인을 우회할 수 있도록 쿼리를 조작하는 MongoDB 인젝션입니다.

2. 공격 기법의 종류

1. 논리 연산자를 이용한 공격

   • $or 연산자를 사용하여 인증 우회:

     { "username": "admin", "$or": [ { "password": "" }, { "password": { "$ne": "" } } ] }

     • 결과적으로, password 필드 값이 무엇이든 쿼리가 참이 되어 인증이 우회됩니다.

2. 비교 연산자를 이용한 공격

   • $gt, $lt, $ne 등의 연산자를 사용해 조건 조작:

     { "username": "admin", "password": { "$ne": "" } }

     • 빈 패스워드를 입력해도 쿼리가 성공적으로 실행됩니다.

3. 코드 실행을 유발하는 공격

   • $where 연산자를 활용하여 JavaScript 코드를 실행:

     { "$where": "this.username === 'admin' && this.password.length > 0" }

     • JavaScript를 통해 복잡한 조건을 삽입하여 인증을 우회하거나 시스템을 공격할 수 있습니다.

3. 데이터 탈취 예제

• MongoDB, test_database, user collection에 다음과 같이 데이터를 추가하였습니다.

  {
  "_id": {
    "$oid": "67865e57d79949e7cc6c05ec"
  },
  "user": "test_user",
  "password": "test_password"
  }

• 그리고 아래와 같이 받은 데이터를 바로 활용하여 조회하는 FastAPI 코드를 작성했습니다.

  from fastapi import FastAPI, Request
  from pymongo import MongoClient
  

app = FastAPI() client = MongoClient("mongodb://localhost:27017/") db = client.test_database

@app.post("/login") async def login(request: Request): body = await request.json() print("Request body:", body) # 추가된 로그 user = body.get("user") password = body.get("password")

# 취약한 코드 예시
user_record = db.user.find_one({"user": user, "password": password})

if user_record:
    return {"message": "Login successful"}
return {"message": "Invalid credentials"}

만약 사용자가 `user` 필드에 `{ "$ne": null }`을 입력하면, 조건이 항상 참이 되어 인증이 우회될 수 있습니다.

![](https://velog.velcdn.com/images/dev-smile/post/116f4871-31b3-40f0-a0c7-57c385b0a6ef/image.png)

---

### 4. MongoDB Injection 예방 방법

#### (1) 입력 검증
사용자 입력을 항상 검증하고, 허용된 값만 사용하도록 제한합니다.
```python
from fastapi import FastAPI, Request
from pydantic import BaseModel, constr
from pymongo import MongoClient

app = FastAPI()
client = MongoClient("mongodb://localhost:27017/")
db = client.test_database


class LoginRequest(BaseModel):
    user: constr(strip_whitespace=True, min_length=1)
    password: constr(strip_whitespace=True, min_length=1)


@app.post("/login")
async def login(request: LoginRequest):
    sanitized_user = request.user
    sanitized_password = request.password

    # 안전한 코드 예시
    user_record = db.users.find_one({"user": sanitized_user, "password": sanitized_password})

    if user_record:
        return {"message": "Login successful"}
    return {"message": "Invalid credentials"}

(2) 파라미터화된 쿼리 사용

MongoDB에서는 일반적으로 사용자의 입력값을 직접 삽입하지 말고, 파라미터를 통해 안전하게 처리해야 합니다.

(3) ODM(Object Document Mapper) 사용

ODM 라이브러리(Motor, Beanie 등)를 사용하면 쿼리 작업 중 자동으로 Injection 방지가 가능합니다.

(4) 권한 관리 강화

데이터베이스에서 권한 관리를 철저히 하여, 애플리케이션의 취약점이 데이터베이스에 직접 영향을 미치지 않도록 해야 합니다.

(5) 최신 보안 패치 적용

MongoDB의 최신 버전을 사용하고, 보안 패치를 항상 유지합니다.

5. 정리 및 결론

MongoDB Injection은 잘못된 입력 검증에서 발생하는 심각한 보안 문제입니다. FastAPI와 같은 프레임워크를 사용할 때에도 철저한 입력 검증, 파라미터화된 쿼리, 그리고 ODM 사용과 같은 최선의 보안 관행을 준수하여 보안 문제를 겪지 않도록 합시다.

RESTful API에서 버저닝이 필요한 이유는 주로 호환성 유지와 유연성을 확보하기 위해서입니다. API는 시간이 지남에 따라 업데이트가 필요하며, 이를 사용자에게 적절히 제공하면서 기존 사용자가 중단 없이 API를 사용할 수 있도록 하기 위해 버저닝이 필수적입니다. 주요 이유를 다음과 같이 정리할 수 있습니다:

1-1. 호환성 유지

• 기존 API 사용자(클라이언트)는 새로운 버전의 API가 출시되어도 영향을 받지 않아야 합니다.
• 새로운 기능이나 수정사항이 추가되더라도 기존 API를 사용하는 시스템은 그대로 작동할 수 있도록 보장합니다.

1-2. 점진적인 업데이트

• API를 사용하는 다양한 클라이언트가 동시에 새 버전으로 업데이트될 수 없는 경우가 많습니다.
• 버저닝을 통해 클라이언트가 새로운 버전으로 전환할 시간을 제공합니다.

1-3. 변경사항 관리

• API의 구조, 요청/응답 포맷, 인증 방식, 엔드포인트 등이 변경될 때, 버전을 통해 변경사항을 명확히 구분할 수 있습니다.
• 예를 들어, 데이터 모델 변경, 새로운 필드 추가, 혹은 기존 필드 삭제 등의 작업을 수행할 수 있습니다.

1-4. 기능 테스트 및 롤백

• 새로운 기능을 포함한 API를 테스트하거나 문제가 발생했을 때 롤백이 쉽습니다.
• 서로 다른 버전의 API를 별도로 유지함으로써 안정성을 확보할 수 있습니다.

1-5. 다양한 클라이언트 요구 지원

• 클라이언트가 요구하는 기능이나 데이터가 서로 다를 수 있습니다. 버전별로 이를 맞춤 지원할 수 있습니다.
• 예를 들어, 모바일 클라이언트와 데스크톱 클라이언트의 요구사항이 다른 경우에 유용합니다.

1-6. API 생명주기 관리

• 오래된 버전을 지원 중단(EOL)할 때, 명확하게 알릴 수 있습니다.
• 버저닝을 통해 API의 수명과 사용자 전환 계획을 체계적으로 관리할 수 있습니다.

1-7. 버저닝 방식

• URI 버저닝

   /api/v1/resource
   /api/v2/resource

  • 직관적이고 널리 사용됨.
  • URL에 명시적으로 포함되므로 변경 관리가 쉬움.

• 쿼리 파라미터

   /api/resource?version=1
   /api/resource?version=2

  • RESTful하지 않다는 비판을 받을 수 있음.

• 헤더 버저닝

   GET /api/resource
   Header: Accept: application/vnd.api+json; version=1

  • URI가 깨끗하지만 사용하기 복잡할 수 있음.

2. FastAPI에서의 Versioning 구현

FastAPI는 경로(Path) 기반의 버전 관리를 간단히 구현할 수 있습니다.

2-1. URI 버저닝

• 경로를 활용한 Versioning
  • 경로에 버전 정보를 포함하는 방식은 가장 간단한 방법입니다.
  • 아래 코드는 /v1/items와 /v2/items 경로에서 각각 다른 버전의 API를 제공합니다.

from fastapi import FastAPI

app = FastAPI()

@app.get("/v1/items")
def read_items_v1():
    return {"version": "v1", "items": ["item1", "item2"]}

@app.get("/v2/items")
def read_items_v2():
    return {"version": "v2", "items": ["item1", "item2"], "additional_info": "v2 specific data"}

• APIRouter를 사용한 모듈화
  • 규모가 큰 애플리케이션에서는 APIRouter를 사용해 버전별 엔드포인트를 모듈화하는 것이 효율적입니다.

from fastapi import APIRouter, FastAPI

app = FastAPI()

v1_router = APIRouter()

@v1_router.get("/items")
def read_items():
    return {"version": "v1", "items": ["item1", "item2"]}

v2_router = APIRouter()

@v2_router.get("/items")
def read_items():
    return {"version": "v2", "items": ["item1", "item2"], "additional_info": "v2 specific data"}

app.include_router(v1_router, prefix="/v1")
app.include_router(v2_router, prefix="/v2")

위와 같이 APIRouter와 prefix를 사용하면 코드 구조를 더 명확하고 관리하기 쉽게 만들 수 있습니다.

2-2. Query Parameter 기반 Versioning

버전 정보를 쿼리 매개변수로 전달받는 방식도 있습니다. 이는 직관적이지만 경로나 헤더 기반 방식만큼 일반적이진 않습니다.

from fastapi import FastAPI, HTTPException, Query

app = FastAPI()

@app.get("/items")
def read_items(version: str = Query(...)):
    if version == "1.0":
        return {"version": "v1", "items": ["item1", "item2"]}
    elif version == "2.0":
        return {"version": "v2", "items": ["item1", "item2"], "additional_info": "v2 specific data"}
    else:
        raise HTTPException(status_code=400, detail="Unsupported API version")

클라이언트는 요청 시 ?version=1.0 형식으로 버전을 지정할 수 있습니다.

2-3. Header 기반 Versioning

버전을 경로 대신 헤더를 통해 전달받는 방식도 가능합니다. 이를 구현하려면 헤더 값을 확인하는 미들웨어를 추가하거나 엔드포인트에서 헤더를 직접 확인하면 됩니다.

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/items")
def read_items(x_api_version: str = Header(...)):
    if x_api_version == "1.0":
        return {"version": "v1", "items": ["item1", "item2"]}
    elif x_api_version == "2.0":
        return {"version": "v2", "items": ["item1", "item2"], "additional_info": "v2 specific data"}
    else:
        raise HTTPException(status_code=400, detail="Unsupported API version")

클라이언트는 x-api-version 헤더를 통해 원하는 버전을 지정할 수 있습니다.

단일 책임 원칙은 SOLID 원칙 중 하나로, 하나의 클래스 또는 모듈은 오직 하나의 책임만 가져야 한다는 원칙입니다. 즉, 특정 클래스나 함수가 하나의 명확한 역할을 수행하고, 변경의 이유가 하나여야 한다는 뜻입니다. 이를 통해 코드는 이해하기 쉽고, 유지보수와 확장이 용이해집니다.

백엔드 코드에서 SRP 적용 방법

백엔드 코드에 이 원칙을 적용하기 위해서는 다음과 같은 단계를 고려할 수 있습니다.

1. 역할(Role) 구분

   • API 요청 처리를 위한 엔드포인트, 비즈니스 로직(서비스), 데이터 접근 계층(리포지토리 또는 DAO)을 분리해 각각의 파일이나 클래스로 구성한다.

2. 책임(Responsibility) 분리

   • 엔드포인트는 HTTP 요청·응답 로직에 집중하고, 서비스 계층은 핵심 비즈니스 로직만 관리하며, 데이터 계층은 DB와의 연동만 책임진다.

3. 의존성 주입(Dependency Injection)

   • 각 계층을 서로 독립적으로 작성한 뒤 필요한 곳에서 의존성을 주입해 결합도를 낮춘다.

이것을 FastAPI와 같은 프레임워크에서는 다음과 같은 방식으로 SRP를 적용할 수 있습니다.

1. 라우팅과 비즈니스 로직 분리

   • API 엔드포인트의 정의는 라우팅 파일에서 관리하고, 비즈니스 로직은 별도의 서비스 계층으로 분리합니다.

2. 데이터베이스 접근 로직 분리

   • 데이터베이스와의 상호작용은 별도의 리포지토리 클래스로 분리하여 관리합니다.

3. 유틸리티 함수와 공통 기능 분리

   • 공통적으로 사용하는 함수(예: 인증, 데이터 검증)는 별도의 유틸리티 모듈에 작성합니다.

FastAPI 예시: SRP 적용

1. 프로젝트 구조 설계

project/
├── app/
│   ├── main.py               # FastAPI 애플리케이션 초기화
│   ├── routes/
│   │   └── user_routes.py    # 사용자 관련 API 엔드포인트 정의
│   ├── services/
│   │   └── user_service.py   # 사용자 비즈니스 로직
│   ├── repositories/
│   │   └── user_repository.py # 데이터베이스 접근 로직
│   ├── models/
│   │   └── user.py           # 사용자 모델 정의
│   ├── utils/
│   │   └── validators.py     # 공통 유틸리티(예: 데이터 검증)

2. 코드 작성

(1) models/user.py - 데이터 모델

from pydantic import BaseModel

class User(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool

(2) repositories/user_repository.py - 데이터베이스 접근 로직

from typing import List, Optional
from app.models.user import User

class UserRepository:
    def __init__(self):
        # 임시 데이터베이스로 리스트 사용
        self.users = []

    def get_user_by_id(self, user_id: int) -> Optional[User]:
        return next((user for user in self.users if user.id == user_id), None)

    def get_all_users(self) -> List[User]:
        return self.users

    def create_user(self, user: User) -> User:
        self.users.append(user)
        return user

(3) services/user_service.py - 비즈니스 로직

from typing import List
from app.models.user import User
from app.repositories.user_repository import UserRepository

class UserService:
    def __init__(self, repository: UserRepository):
        self.repository = repository

    def get_user_details(self, user_id: int) -> User:
        user = self.repository.get_user_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        return user

    def create_user(self, user_data: User) -> User:
        # 추가 비즈니스 로직 (예: 유효성 검사)
        if not user_data.email:
            raise ValueError("Email is required")
        return self.repository.create_user(user_data)

(4) routes/user_routes.py - 라우팅 정의

from fastapi import APIRouter, HTTPException
from app.models.user import User
from app.services.user_service import UserService
from app.repositories.user_repository import UserRepository

router = APIRouter()
repository = UserRepository()
service = UserService(repository)

@router.get("/users/{user_id}")
def get_user(user_id: int):
    try:
        return service.get_user_details(user_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

@router.post("/users")
def create_user(user: User):
    try:
        return service.create_user(user)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

(5) main.py - 애플리케이션 초기화

from fastapi import FastAPI
from app.routes.user_routes import router as user_router

app = FastAPI()

app.include_router(user_router, prefix="/api")

SRP가 적용된 장점

1. 유지보수 용이성: 라우트, 서비스, 리포지토리가 분리되어 특정 로직을 변경해도 다른 부분에 영향을 미치지 않음.
2. 재사용성 증가: 서비스 계층과 리포지토리 계층은 다른 API에서 재사용 가능.
3. 테스트 용이성: 각 계층별로 단위 테스트 작성이 가능.