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는 크게 두 가지 핵심 라이브러리 위에서 동작합니다.
- Starlette: ASGI 기반의 비동기 웹 프레임워크로, FastAPI의 웹 서버 기능을 담당합니다. 라우팅, 미들웨어, Request/Response 처리 등 웹 프레임워크의 기본 기능을 제공합니다.
- 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)
요청이 들어오면:
- Router가 등록된 모든 Route를 순회
- 각 Route의
matches()메서드로 경로 매칭 확인 - 매칭되는 Route를 찾으면 해당 endpoint 함수 실행
- path parameter는 자동으로 추출되어 함수 인자로 전달
- 매칭되는 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의 주요 내장 미들웨어
- CORSMiddleware: Cross-Origin Resource Sharing 처리
- GZipMiddleware: 응답 압축
- TrustedHostMiddleware: 허용된 호스트만 접근 가능
- HTTPSRedirectMiddleware: HTTP를 HTTPS로 리다이렉트
- 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 요청이 들어오면:
- 로깅 미들웨어 실행 (before) -
[Before] GET /test - CORS 미들웨어 실행
- 라우트 핸들러 실행 -
{"message": "test"} - CORS 미들웨어 완료 (CORS 헤더 추가)
- 로깅 미들웨어 완료 (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-core의SchemaValidator로 매우 빠른 검증 성능 제공__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+$') |
description | API 문서 설명 | 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 타입 힌트만으로 간결하게 구현할 수 있다는 것이 가장 큰 장점입니다.