Home FastAPI 톺아보기
Post
Cancel

FastAPI 톺아보기

FastAPI 공식 문에서는 FastAPI를 다음과 같이 정의하고 있습니다.

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python based on standard Python type hints.

즉, FastAPI는 표준 Python 타입 힌트를 기반으로 API를 구축하기 위한 현대적이고 빠른 고성능 웹 프레임워크입니다.

FastAPI가 어떻게 구성되어 있고 어떤 장점이 있는지 하나씩 공식 문서를 참고하여 살펴보겠습니다.

분석 중 나오는 모든 코드는 아래 환경을기준으로 작성되었습니다.

🍳 분석 환경

  • fastAPI: 0.121.1
  • python: 3.13
  • pydantic: 2.12.4

✅ FastAPI의 구성

FastAPI는 크게 두 가지 핵심 라이브러리 위에서 동작합니다.

  1. Starlette: ASGI 기반의 비동기 웹 프레임워크로, FastAPI의 웹 서버 기능을 담당합니다. 라우팅, 미들웨어, Request/Response 처리 등 웹 프레임워크의 기본 기능을 제공합니다.
  2. Pydantic: 타입 힌트 기반의 데이터 검증 라이브러리로, 요청 및 응답 데이터의 자동 검증과 직렬화를 담당합니다.

FastAPI는 이 두 라이브러리를 기반으로 개발자 편의성을 대폭 강화한 프레임워크라고 볼 수 있습니다. 먼저 Starlette의 역할부터 자세히 알아보겠습니다.

📌 Starlette

Starlette는 FastAPI의 기반이 되는 ASGI 웹 프레임워크입니다. FastAPI는 실제로 Starlette를 상속받아 구현되어 있습니다.

코드를 통해 자세히 살펴보겠습니다.

1
2
3
4
5
6
from fastapi import FastAPI

app.get("/root")
async def root():
    return {"message": "Hello World"}

위 코드에서는 FastAPI 인스턴스를 생성하고 @app.get("/root") 데코레이터로 /root 경로에 대한 GET 요청 핸들러를 등록합니다.

FastAPI 클래스의 내부 구현을 들여다보면 (fastapi/applications.py), Starlette를 import하고 상속받고 있는 것을 확인할 수 있습니다.

FastAPI Class 들어가면 application.py 파일이 나오는데 거기서 Starlette를 import 하고 있습니다.

1
2
3
4
from starlette.applications import Starlette

class FastAPI(Starlette):
    def __init__(self, ...):

이처럼 FastAPI는 Starlette의 핵심 기능을 그대로 사용하면서, 그 위에 추가적인 편의 기능들을 올린 구조입니다.


1️⃣ 라우팅 (Routing)

Starlette의 가장 핵심적인 기능은 라우팅입니다. 라우팅이란 클라이언트의 HTTP 요청을 적절한 핸들러 함수로 연결해주는 작업을 말합니다. ex) Spring의 Dispatcher Servlet 역할

/starlette/routing.py 파일을 살펴보면 라우팅의 핵심 클래스들을 확인할 수 있습니다.

Router 클래스

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Router:
    def __init__(
        self,
        routes: Sequence[BaseRoute] | None = None,
        redirect_slashes: bool = True,
        default: ASGIApp | None = None,
        on_startup: Sequence[Callable[[], Any]] | None = None,
        on_shutdown: Sequence[Callable[[], Any]] | None = None,
        lifespan: Lifespan[Any] | None = None,
        *,
        middleware: Sequence[Middleware] | None = None,
    ) -> None:
        self.routes = [] if routes is None else list(routes)
        self.redirect_slashes = redirect_slashes
        self.default = self.not_found if default is None else default
        # ...

Router 클래스는 여러 개의 Route를 관리하는 핵심 클래스입니다. 주요 파라미터를 살펴보면:

  • routes: Route 객체들의 리스트
  • redirect_slashes: 슬래시(/) 리다이렉션 여부 (예: /users//users)
  • default: 매칭되는 라우트가 없을 때 실행될 기본 핸들러
  • on_startup, on_shutdown: 애플리케이션 시작/종료 시 실행될 함수들 (deprecated)
  • lifespan: 애플리케이션 생명주기 관리를 위한 컨텍스트 매니저 (최신 방식)
  • middleware: 요청/응답 처리에 적용될 미들웨어 리스트

Router는 등록된 모든 라우트를 순회하며 요청과 매칭되는 라우트를 찾아 해당 핸들러를 실행시킵니다.

FastAPI에서 @app.get(), @app.post() 같은 데코레이터를 사용하면, 내부적으로는 이 Route 객체가 생성되어 Router에 등록되는 방식입니다.

예를 들어:

1
2
3
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

이 코드는 내부적으로 다음과 같이 동작합니다:

1
2
3
4
5
6
7
# 내부 동작 (simplified)
route = Route(
    path="/users/{user_id}",
    endpoint=get_user,
    methods=["GET"]
)
app.router.routes.append(route)

요청이 들어오면:

  1. Router가 등록된 모든 Route를 순회
  2. 각 Route의 matches() 메서드로 경로 매칭 확인
  3. 매칭되는 Route를 찾으면 해당 endpoint 함수 실행
  4. path parameter는 자동으로 추출되어 함수 인자로 전달
  5. 매칭되는 Route가 없으면 default 핸들러 실행 (보통 404 에러)


2️⃣ 미들웨어 (Middleware)

미들웨어는 요청이 라우트 핸들러에 도달하기 전과 응답이 클라이언트에게 전달되기 전에 실행되는 컴포넌트입니다. 로깅, 인증, CORS 처리, 압축 등 공통적인 처리 로직을 미들웨어로 구현할 수 있습니다.

ex) Spring의 Filter 또는 interceptor 역할

Starlette는 starlette/middleware 패키지에서 다양한 미들웨어를 제공합니다.

미들웨어의 동작 방식

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from starlette.middleware.base import BaseHTTPMiddleware

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # 요청 처리 전 로직
        print(f"Request: {request.url}")
        
        # 다음 미들웨어 또는 라우트 핸들러 호출
        response = await call_next(request)
        
        # 응답 처리 후 로직
        print(f"Response status: {response.status_code}")
        
        return response

미들웨어는 다음과 같은 순서로 실행됩니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
Client Request
    ↓
Middleware 1 (before)
    ↓
Middleware 2 (before)
    ↓
Route Handler
    ↓
Middleware 2 (after)
    ↓
Middleware 1 (after)
    ↓
Client Response

Middleware 클래스

1
2
3
4
5
6
7
# starlette/middleware/__init__.py
@dataclass
class Middleware:
  def __init__(self, cls: _MiddlewareFactory[P], *args: P.args, **kwargs: P.kwargs) -> None:
    self.cls = cls
    self.args = args
    self.kwargs = kwargs

Middleware 클래스는 미들웨어의 메타데이터를 담는 데이터 클래스입니다:

  • cls: 실제 미들웨어 클래스
  • args: 미들웨어 생성자에 전달할 위치 인자
  • kwargs: 미들웨어 생성자에 전달할 키워드 인자

FastAPI에서 미들웨어를 등록하는 방법은 크게 두 가지입니다:

방법 1: 데코레이터 사용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    import time
    start_time = time.time()
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

방법 2: add_middleware 메서드 사용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()

# CORS 미들웨어
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip 압축 미들웨어
app.add_middleware(GZipMiddleware, minimum_size=1000)

Starlette의 주요 내장 미들웨어

  1. CORSMiddleware: Cross-Origin Resource Sharing 처리
  2. GZipMiddleware: 응답 압축
  3. TrustedHostMiddleware: 허용된 호스트만 접근 가능
  4. HTTPSRedirectMiddleware: HTTP를 HTTPS로 리다이렉트
  5. SessionMiddleware: 세션 관리

미들웨어 스택 구성

Router 클래스에서 미들웨어는 다음과 같이 구성됩니다:

1
2
3
4
5
6
7
8
9
class Router:
    def __init__(
        self,
        routes: Sequence[BaseRoute] | None = None,
        *,
        middleware: Sequence[Middleware] | None = None,
    ) -> None:
        self.routes = [] if routes is None else list(routes)
        self.middleware = [] if middleware is None else list(middleware)

내부적으로 미들웨어들은 스택 구조로 쌓이며, 요청이 들어오면 역순으로 실행됩니다:

1
2
3
4
5
6
7
8
# 미들웨어 등록 순서
app.add_middleware(Middleware1)  # 첫 번째
app.add_middleware(Middleware2)  # 두 번째
app.add_middleware(Middleware3)  # 세 번째

# 실행 순서
# Request: Middleware3 → Middleware2 → Middleware1 → Handler
# Response: Handler → Middleware1 → Middleware2 → Middleware3

실제 예시로 보면:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, Request
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

# 1. CORS 미들웨어 등록
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
)

# 2. 커스텀 로깅 미들웨어
@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"[Before] {request.method} {request.url}")
    response = await call_next(request)
    print(f"[After] Status: {response.status_code}")
    return response

@app.get("/test")
async def test():
    return {"message": "test"}

위 코드에서 /test 요청이 들어오면:

  1. 로깅 미들웨어 실행 (before) - [Before] GET /test
  2. CORS 미들웨어 실행
  3. 라우트 핸들러 실행 - {"message": "test"}
  4. CORS 미들웨어 완료 (CORS 헤더 추가)
  5. 로깅 미들웨어 완료 (after) - [After] Status: 200

위처럼 처리되어 순서를 염두에 두고 미들웨어를 설계하는 것이 중요합니다.


3️⃣ Request/Response 처리

Starlette는 HTTP 요청과 응답을 처리하기 위한 Request와 다양한 Response 클래스를 제공합니다. 이를 통해 클라이언트와의 통신을 효율적으로 처리할 수 있습니다.
ex) Spring의 HttpServletRequest, HttpServletResponse 역할

  • Request와 Response는 제외할까 하다가 FastAPI에서 의외로 다룰일이 있어서 포함시켰습니다.

Request 클래스

1
2
3
4
5
6
7
8
9
10
11
12
# starlette/requests.py
class Request(HTTPConnection):
  _form: FormData | None

  def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send):
    super().__init__(scope)
    assert scope["type"] == "http"
    self._receive = receive
    self._send = send
    self._stream_consumed = False
    self._is_disconnected = False
    self._form = None

Request 객체는 클라이언트로부터 받은 HTTP 요청의 모든 정보를 담고 있습니다. 주요 메서드와 프로퍼티는 다음과 같습니다:

요청 데이터 파싱

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# JSON 데이터 파싱
async def json(self) -> Any:
    if not hasattr(self, "_json"):
        body = await self.body()
        self._json = json_loads(body)
    return self._json

# Raw body 읽기
async def body(self) -> bytes:
    if not hasattr(self, "_body"):
        chunks: list[bytes] = []
        async for chunk in self.stream():
            chunks.append(chunk)
        self._body = b"".join(chunks)
    return self._body

요청 정보 접근

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@property
def query_params(self) -> QueryParams:
    # URL 쿼리 파라미터 (예: ?page=1&limit=10)
    if not hasattr(self, "_query_params"):
        self._query_params = QueryParams(self._scope["query_string"])
    return self._query_params

@property
def path_params(self) -> dict[str, Any]:
    # Path 파라미터 (예: /users/{user_id})
    return self._scope.get("path_params", {})

@property
def headers(self) -> Headers:
    # HTTP 헤더
    if not hasattr(self, "_headers"):
        self._headers = Headers(scope=self._scope)
    return self._headers

@property
def cookies(self) -> dict[str, str]:
    if not hasattr(self, "_cookies"):
        cookies: dict[str, str] = {}
        cookie_header = self.headers.get("cookie")
        if cookie_header:
            cookies = parse_cookie(cookie_header)
        self._cookies = cookies
    return self._cookies

실제 사용 예시:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from starlette.requests import Request
from starlette.responses import JSONResponse

async def handle_request(request: Request):
    # 쿼리 파라미터
    page = request.query_params.get("page", "1")
    
    # Path 파라미터
    user_id = request.path_params.get("user_id")
    
    # 헤더
    auth_token = request.headers.get("Authorization")
    
    # JSON body
    data = await request.json()
    
    # 쿠키
    session_id = request.cookies.get("session_id")
    
    return JSONResponse({
        "page": page,
        "user_id": user_id,
        "has_auth": auth_token is not None
    })

Response 클래스

Starlette는 다양한 상황에 맞는 여러 Response 클래스를 제공합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
# starlette/responses.py

# 1. 기본 Response
class Response:
  media_type = None
  charset = "utf-8"

  def __init__(
    self,
    content: Any = None,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
  ) -> None:
    self.status_code = status_code
    if media_type is not None:
      self.media_type = media_type
    self.background = background
    self.body = self.render(content)
    self.init_headers(headers)

# 2. JSON 응답
class JSONResponse(Response):
    media_type = "application/json"

    def __init__(
      self,
      content: Any,
      status_code: int = 200,
      headers: Mapping[str, str] | None = None,
      media_type: str | None = None,
      background: BackgroundTask | None = None,
    ) -> None:
      super().__init__(content, status_code, headers, media_type, background)
      
    def render(self, content: Any) -> bytes:
        return json_dumps(
            content,
            ensure_ascii=False,
            allow_nan=False,
            indent=None,
            separators=(",", ":"),
        ).encode("utf-8")

# 3. HTML 응답
class HTMLResponse(Response):
    media_type = "text/html"

# 4. 파일 응답
class FileResponse(Response):
    chunk_size = 64 * 1024  # 64KB
    
    def __init__(
        self,
      path: str | os.PathLike[str],
      status_code: int = 200,
      headers: Mapping[str, str] | None = None,
      media_type: str | None = None,
      background: BackgroundTask | None = None,
      filename: str | None = None,
      stat_result: os.stat_result | None = None,
      method: str | None = None,
      content_disposition_type: str = "attachment",
    ) -> None:
        # ...

# 6. 리다이렉트 응답
class RedirectResponse(Response):
    def __init__(
        self,
        url: str | URL,
        status_code: int = 307,
        headers: Mapping[str, str] | None = None,
      background: BackgroundTask | None = None,
    ) -> None:
        # ...

Response 사용 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from starlette.responses import (
    JSONResponse, 
    HTMLResponse, 
    FileResponse,
    StreamingResponse,
    RedirectResponse
)

# JSON 응답
@app.get("/api/users")
async def get_users():
    return JSONResponse({"users": [{"id": 1, "name": "John"}]})

# HTML 응답
@app.get("/")
async def home():
    html_content = "<html><body><h1>Hello World</h1></body></html>"
    return HTMLResponse(content=html_content)

# 파일 다운로드
@app.get("/download")
async def download_file():
    return FileResponse(
        path="files/document.pdf",
        filename="download.pdf",
        media_type="application/pdf"
    )


# 리다이렉트
@app.get("/old-url")
async def redirect():
    return RedirectResponse(url="/new-url")

FastAPI에서의 활용

FastAPI는 Starlette의 Request/Response를 그대로 사용하면서, 추가적인 편의 기능을 제공합니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse

app = FastAPI()

# Starlette Request 직접 사용 가능
@app.get("/custom")
async def custom_handler(request: Request):
    # Starlette Request 객체 그대로 사용
    user_agent = request.headers.get("user-agent")
    client_host = request.client.host
    
    return JSONResponse({
        "user_agent": user_agent,
        "client_ip": client_host
    })

# FastAPI는 자동으로 dict를 JSONResponse로 변환
@app.get("/simple")
async def simple_handler():
    # 내부적으로 JSONResponse로 변환됨
    return {"message": "Hello"}



📌 Pydantic

Pydantic은 Python 타입 힌트를 기반으로 데이터 검증과 직렬화를 수행하는 라이브러리입니다. FastAPI에서 요청/응답 데이터의 자동 검증, 문서화, 직렬화를 담당하는 핵심 컴포넌트입니다.

ex) Spring의 @Valid, @RequestBody, DTO, Bean Validation 역할


1️⃣ BaseModel - 데이터 검증의 핵심

Pydantic의 가장 기본이 되는 클래스는 BaseModel입니다. 모든 데이터 모델은 이 클래스를 상속받아 정의합니다.

1
2
3
4
5
6
7
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool = True  # 기본값 설정

Spring에서는 DTO를 다음과 같이 정의합니다:

1
2
3
4
5
6
7
8
9
10
11
12
public class UserDto {
    @NotNull
    private Long id;

    @NotBlank
    private String name;

    @Email
    private String email;

    private boolean isActive = true;
}

Pydantic은 타입 힌트만으로 자동 검증이 이루어지므로, 별도의 어노테이션 없이도 타입 안전성을 보장합니다.

BaseModel 내부 구조

pydantic/main.py를 살펴보면:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class BaseModel(metaclass=_model_construction.ModelMetaclass):
    """A base class for creating Pydantic models."""

    model_config: ClassVar[ConfigDict] = ConfigDict()
    """Configuration for the model, should be a dictionary conforming to ConfigDict."""

    __pydantic_complete__: ClassVar[bool] = False
    """Whether model building is completed, or if there are still undefined fields."""

    __pydantic_core_schema__: ClassVar[CoreSchema]
    """The core schema of the model."""

    __pydantic_validator__: ClassVar[SchemaValidator | PluggableSchemaValidator]
    """The `pydantic-core` SchemaValidator used to validate instances of the model."""

    __pydantic_serializer__: ClassVar[SchemaSerializer]
    """The `pydantic-core` SchemaSerializer used to dump instances of the model."""

    __pydantic_fields__: ClassVar[Dict[str, FieldInfo]]
    """A dictionary of field names and their corresponding FieldInfo objects.
    This replaces `Model.__fields__` from Pydantic V1."""

    __pydantic_extra__: Dict[str, Any] | None = _model_construction.NoInitField(init=False)
    """A dictionary containing extra values, if `extra` is set to 'allow'."""

    __pydantic_fields_set__: set[str] = _model_construction.NoInitField(init=False)
    """The names of fields explicitly set during instantiation."""

    __slots__ = '__dict__', '__pydantic_fields_set__', '__pydantic_extra__', '__pydantic_private__'

    def __init__(self, /, **data: Any) -> None:
        """Create a new model by parsing and validating input data from keyword arguments."""
        __tracebackhide__ = True
        validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
        if self is not validated_self:
            warnings.warn(
                'A custom validator is returning a value other than `self`.\n'
                "Returning anything other than `self` from a top level model validator isn't supported when validating via `__init__`.\n"
                'See the `model_validator` docs for more details.',
                stacklevel=2,
            )

BaseModel의 주요 클래스 변수:

  • model_config: 모델 설정을 담는 ConfigDict 객체
  • __pydantic_validator__: Rust로 작성된 pydantic-coreSchemaValidator로 매우 빠른 검증 성능 제공
  • __pydantic_serializer__: 모델 인스턴스를 직렬화하는 SchemaSerializer
  • __pydantic_fields__: 필드명과 FieldInfo 객체의 딕셔너리 (Pydantic V1의 Model.__fields__를 대체)
  • __pydantic_fields_set__: 인스턴스 생성 시 명시적으로 설정된 필드명 집합

__init__ 메서드에서 __pydantic_validator__.validate_python()을 호출하여 입력 데이터를 검증합니다.

주요 클래스 변수 활용 예시

1
2
3
4
5
6
7
from pydantic import BaseModel, Field

class User(BaseModel):
    id: int
    name: str = Field(description="사용자 이름")
    email: str | None = None
    is_active: bool = True

1. __pydantic_fields__ - 필드 메타데이터 조회

1
2
3
4
5
6
7
8
9
10
11
12
# 모델에 정의된 모든 필드 정보 조회
for field_name, field_info in User.__pydantic_fields__.items():
    print(f"{field_name}: {field_info.annotation}, required={field_info.is_required()}")

# 출력:
# id: <class 'int'>, required=True
# name: <class 'str'>, required=False
# email: str | None, required=False
# is_active: <class 'bool'>, required=False

# 특정 필드의 description 조회
print(User.__pydantic_fields__['name'].description)  # "사용자 이름"

2. __pydantic_fields_set__ - 명시적으로 설정된 필드 확인

1
2
3
4
5
6
7
8
9
10
11
12
# 기본값을 사용한 경우
user1 = User(id=1, name="John")
print(user1.__pydantic_fields_set__)  # {'id', 'name'}

# 명시적으로 기본값과 같은 값을 설정한 경우
user2 = User(id=1, name="John", is_active=True)
print(user2.__pydantic_fields_set__)  # {'id', 'name', 'is_active'}

# 활용: PATCH 요청에서 실제로 전달된 필드만 업데이트
def update_user(user_id: int, update_data: User):
    for field in update_data.__pydantic_fields_set__:
        print(f"Updating {field}: {getattr(update_data, field)}")

3. model_config - 모델 설정 조회

1
2
3
4
5
6
7
8
9
10
11
from pydantic import ConfigDict

class StrictUser(BaseModel):
    model_config = ConfigDict(strict=True, frozen=True)

    id: int
    name: str

# 모델 설정 조회
print(StrictUser.model_config.get('strict'))  # True
print(StrictUser.model_config.get('frozen'))  # True

4. __pydantic_complete__ - 모델 빌드 완료 여부

1
2
3
4
5
6
7
8
9
10
11
12
from typing import TYPE_CHECKING

class Post(BaseModel):
    id: int
    author: 'User'  # Forward Reference

# Forward Reference가 해결되기 전
print(Post.__pydantic_complete__)  # False (미완성 상태일 수 있음)

# model_rebuild()로 강제 재빌드
Post.model_rebuild()
print(Post.__pydantic_complete__)  # True

FastAPI에서의 활용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str
    age: int

@app.post("/users")
async def create_user(user: UserCreate):
    # user는 이미 검증된 UserCreate 인스턴스
    return {"message": f"User {user.name} created"}

잘못된 데이터가 들어오면 FastAPI가 자동으로 422 에러를 반환합니다:

1
2
3
4
5
6
7
8
9
10
11
// POST /users {"name": "John", "email": "john@example.com", "age": "not_a_number"}
{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["body", "age"],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "not_a_number"
        }
    ]
}


2️⃣ 타입 힌트 기반 자동 검증

Pydantic은 Python의 타입 힌트를 활용하여 다양한 타입을 자동으로 검증합니다.

기본 타입

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from pydantic import BaseModel
from typing import Optional, List, Dict
from datetime import datetime
from enum import Enum

class Status(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

class Product(BaseModel):
    id: int                          # 정수
    name: str                        # 문자열
    price: float                     # 실수
    is_available: bool               # 불리언
    tags: List[str]                  # 문자열 리스트
    metadata: Dict[str, Any]         # 딕셔너리
    created_at: datetime             # 날짜/시간
    status: Status                   # Enum
    description: Optional[str] = None  # 선택적 필드

자동 타입 변환 (Coercion)

Pydantic은 가능한 경우 자동으로 타입을 변환합니다:

1
2
3
4
5
6
7
8
class Item(BaseModel):
    id: int
    price: float

# 문자열 "123"이 int 123으로 자동 변환
item = Item(id="123", price="99.99")
print(item.id)     # 123 (int)
print(item.price)  # 99.99 (float)

이 동작은 model_config에서 제어할 수 있습니다:

1
2
3
4
5
6
7
8
9
10
from pydantic import BaseModel, ConfigDict

class StrictItem(BaseModel):
    model_config = ConfigDict(strict=True)  # 엄격 모드

    id: int
    price: float

# strict=True일 경우 문자열 "123"은 에러 발생
# StrictItem(id="123", price=99.99)  # ValidationError!


3️⃣ Field를 통한 상세 검증

더 세밀한 검증이 필요한 경우 Field를 사용합니다.

ex) Spring의 @Size, @Min, @Max, @Pattern 역할

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from pydantic import BaseModel, Field

class User(BaseModel):
    name: str = Field(
        min_length=2,           # 최소 길이
        max_length=50,          # 최대 길이
        description="사용자 이름"  # API 문서에 표시
    )
    age: int = Field(
        ge=0,                   # >= 0 (greater than or equal)
        le=150,                 # <= 150 (less than or equal)
        description="나이"
    )
    email: str = Field(
        pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$',  # 정규식 패턴
        description="이메일 주소"
    )
    score: float = Field(
        gt=0,                   # > 0 (greater than)
        lt=100,                 # < 100 (less than)
        description="점수"
    )

Spring에서 동일한 검증:

1
2
3
4
5
6
7
8
9
10
public class UserDto {
    @Size(min = 2, max = 50)
    private String name;

    @Min(0) @Max(150)
    private int age;

    @Pattern(regexp = "^[\\w\\.-]+@[\\w\\.-]+\\.\\w+$")
    private String email;
}

Field의 주요 파라미터

pydantic/fields.py에서 Field 함수를 살펴보면:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def Field(
    default: Any = PydanticUndefined,
    *,
    default_factory: Callable[[], Any] | None = None,
    alias: str | None = None,
    validation_alias: str | AliasPath | AliasChoices | None = None,
    serialization_alias: str | None = None,
    title: str | None = None,
    description: str | None = None,
    examples: List[Any] | None = None,
    gt: float | None = None,
    ge: float | None = None,
    lt: float | None = None,
    le: float | None = None,
    multiple_of: float | None = None,
    min_length: int | None = None,
    max_length: int | None = None,
    pattern: str | None = None,
    # ...
) -> Any:
파라미터설명예시
default기본값Field(default="unknown")
default_factory기본값 팩토리 함수Field(default_factory=list)
alias필드 별칭Field(alias="userName")
gt, ge, lt, le숫자 범위 검증Field(ge=0, le=100)
min_length, max_length문자열 길이Field(min_length=1)
pattern정규식 패턴Field(pattern=r'^\d+$')
descriptionAPI 문서 설명Field(description="사용자 ID")

Alias 활용

API 응답과 내부 필드명이 다를 때 유용합니다:

1
2
3
4
5
6
7
8
9
10
11
12
from pydantic import BaseModel, Field

class User(BaseModel):
    user_name: str = Field(alias="userName")  # camelCase -> snake_case
    created_at: datetime = Field(alias="createdAt")

    model_config = ConfigDict(populate_by_name=True)

# JSON에서 camelCase로 받아도 snake_case로 접근 가능
data = {"userName": "john", "createdAt": "2024-01-01T00:00:00"}
user = User(**data)
print(user.user_name)  # "john"


4️⃣ Custom Validator

기본 검증으로 부족한 경우 커스텀 검증 로직을 추가할 수 있습니다.

ex) Spring의 Custom Validator, @AssertTrue 역할

@field_validator - 단일 필드 검증

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from pydantic import BaseModel, field_validator

class User(BaseModel):
    name: str
    email: str
    password: str

    @field_validator('name')
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('이름은 비어있을 수 없습니다')
        return v.strip()

    @field_validator('email')
    @classmethod
    def email_must_contain_at(cls, v: str) -> str:
        if '@' not in v:
            raise ValueError('유효한 이메일 형식이 아닙니다')
        return v.lower()  # 소문자로 변환

    @field_validator('password')
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError('비밀번호는 8자 이상이어야 합니다')
        if not any(c.isupper() for c in v):
            raise ValueError('비밀번호에 대문자가 포함되어야 합니다')
        if not any(c.isdigit() for c in v):
            raise ValueError('비밀번호에 숫자가 포함되어야 합니다')
        return v

@model_validator - 모델 전체 검증

여러 필드 간의 관계를 검증할 때 사용합니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pydantic import BaseModel, model_validator

class DateRange(BaseModel):
    start_date: datetime
    end_date: datetime

    @model_validator(mode='after')
    def check_dates(self) -> 'DateRange':
        if self.start_date >= self.end_date:
            raise ValueError('시작일은 종료일보다 이전이어야 합니다')
        return self

class UserCreate(BaseModel):
    password: str
    password_confirm: str

    @model_validator(mode='after')
    def check_passwords_match(self) -> 'UserCreate':
        if self.password != self.password_confirm:
            raise ValueError('비밀번호가 일치하지 않습니다')
        return self

mode 옵션

  • mode='before': 타입 변환 전에 실행 (raw 데이터 접근)
  • mode='after': 타입 변환 후에 실행 (검증된 데이터 접근)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pydantic import BaseModel, model_validator
from typing import Any

class FlexibleModel(BaseModel):
    data: dict

    @model_validator(mode='before')
    @classmethod
    def convert_string_to_dict(cls, values: Any) -> Any:
        # JSON 문자열을 dict로 변환
        if isinstance(values.get('data'), str):
            import json
            values['data'] = json.loads(values['data'])
        return values


5️⃣ Nested Model (중첩 모델)

복잡한 데이터 구조를 표현할 때 모델을 중첩하여 사용합니다.

ex) Spring의 중첩 DTO 구조

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from pydantic import BaseModel
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    zip_code: str
    country: str = "Korea"

class Company(BaseModel):
    name: str
    address: Address  # 중첩 모델

class User(BaseModel):
    id: int
    name: str
    email: str
    address: Optional[Address] = None  # 선택적 중첩 모델
    company: Optional[Company] = None
    friends: List['User'] = []  # 자기 참조 (Forward Reference)

사용 예시

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 중첩된 JSON 데이터
data = {
    "id": 1,
    "name": "John",
    "email": "john@example.com",
    "address": {
        "street": "123 Main St",
        "city": "Seoul",
        "zip_code": "12345"
    },
    "company": {
        "name": "Tech Corp",
        "address": {
            "street": "456 Business Ave",
            "city": "Seoul",
            "zip_code": "67890"
        }
    }
}

user = User(**data)
print(user.address.city)          # "Seoul"
print(user.company.address.city)  # "Seoul"

FastAPI에서 중첩 모델 활용

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class OrderItem(BaseModel):
    product_id: int
    quantity: int
    price: float

class Order(BaseModel):
    order_id: int
    customer_name: str
    items: List[OrderItem]

    @property
    def total_price(self) -> float:
        return sum(item.price * item.quantity for item in self.items)

@app.post("/orders")
async def create_order(order: Order):
    return {
        "order_id": order.order_id,
        "total_price": order.total_price,
        "item_count": len(order.items)
    }

요청 예시:

1
2
3
4
5
6
7
8
{
    "order_id": 1,
    "customer_name": "John",
    "items": [
        {"product_id": 101, "quantity": 2, "price": 10.0},
        {"product_id": 102, "quantity": 1, "price": 25.0}
    ]
}


6️⃣ 직렬화와 역직렬화

Pydantic 모델은 다양한 형식으로 변환할 수 있습니다.

모델 → dict/JSON

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    password: str  # 민감 정보

user = User(id=1, name="John", email="john@example.com", password="secret123")

# dict로 변환
user_dict = user.model_dump()
# {'id': 1, 'name': 'John', 'email': 'john@example.com', 'password': 'secret123'}

# 특정 필드 제외
user_dict = user.model_dump(exclude={'password'})
# {'id': 1, 'name': 'John', 'email': 'john@example.com'}

# 특정 필드만 포함
user_dict = user.model_dump(include={'id', 'name'})
# {'id': 1, 'name': 'John'}

# JSON 문자열로 변환
user_json = user.model_dump_json()
# '{"id":1,"name":"John","email":"john@example.com","password":"secret123"}'

dict/JSON → 모델

1
2
3
4
5
# dict에서 생성
user = User(**{"id": 1, "name": "John", "email": "john@example.com", "password": "secret"})

# JSON 문자열에서 생성
user = User.model_validate_json('{"id": 1, "name": "John", "email": "john@example.com", "password": "secret"}')

응답 모델 분리 패턴

보안을 위해 요청과 응답 모델을 분리하는 것이 좋습니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from pydantic import BaseModel
from typing import Optional

# 요청용 모델
class UserCreate(BaseModel):
    name: str
    email: str
    password: str

# 응답용 모델 (비밀번호 제외)
class UserResponse(BaseModel):
    id: int
    name: str
    email: str

# DB 모델 (내부용)
class UserInDB(BaseModel):
    id: int
    name: str
    email: str
    hashed_password: str

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    # password를 해싱하여 저장
    hashed_password = hash_password(user.password)

    # DB에 저장 (가정)
    db_user = UserInDB(
        id=1,
        name=user.name,
        email=user.email,
        hashed_password=hashed_password
    )

    # response_model=UserResponse이므로 password 관련 필드는 자동 제외
    return db_user


7️⃣ 실전 활용 팁

Config를 통한 모델 설정

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,      # 문자열 앞뒤 공백 제거
        str_min_length=1,               # 모든 문자열 최소 길이
        frozen=True,                    # 불변 객체로 만들기 (immutable)
        extra='forbid',                 # 정의되지 않은 필드 금지
        use_enum_values=True,           # Enum 값 자동 변환
        validate_assignment=True,       # 할당 시에도 검증
    )

    name: str
    email: str

ORM 모드 (from_attributes)

SQLAlchemy 등 ORM 객체를 Pydantic 모델로 변환할 때:

1
2
3
4
5
6
7
8
9
10
11
12
from pydantic import BaseModel, ConfigDict

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    email: str

# SQLAlchemy 모델에서 직접 변환
db_user = session.query(UserModel).first()
user_response = UserResponse.model_validate(db_user)

Generic Model

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from pydantic import BaseModel
from typing import TypeVar, Generic, List

T = TypeVar('T')

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

    @property
    def pages(self) -> int:
        return (self.total + self.size - 1) // self.size

# 사용
class User(BaseModel):
    id: int
    name: str

response = PaginatedResponse[User](
    items=[User(id=1, name="John"), User(id=2, name="Jane")],
    total=100,
    page=1,
    size=10
)

이처럼 Pydantic은 FastAPI에서 데이터 검증, 직렬화, API 문서화를 자동으로 처리해주는 핵심 라이브러리입니다. Spring의 Bean Validation과 DTO 패턴을 Python 타입 힌트만으로 간결하게 구현할 수 있다는 것이 가장 큰 장점입니다.

This post is written by PRO.