FastAPI Pydantic 고급

pydantic 고급

Tip

• email : EmailStr | None = None
  • EmailStr 이메일 형식 검증, pydnatic 모듈에 있음
  • Or none 선택적이라고 명시하는것 Optional 과 같음 3.10 버전 이후부터 존재
  • EamilStr = None 이랑 같음

✅ 1. Pydantic vs Django Serializer 비교

기능	FastAPI (Pydantic)	Django (Serializer)
데이터 검증	✅ 가능 (BaseModel 사용)	✅ 가능 (serializers.Serializer 사용)
JSON 직렬화	✅ 가능 (Python 객체 → JSON 변환)	✅ 가능 (.data 사용)
DB 모델 연동	❌ X (Pydantic은 DB 모델과 분리됨)	✅ 가능 (serializers.ModelSerializer 사용 가능)
데이터 역직렬화	✅ 가능 (JSON → Pydantic 객체)	✅ 가능 (.is_valid() 사용)
요청 데이터 처리	✅ request.body 자동 변환	✅ request.data (Django REST Framework 필요)

🔹 결론:

• Pydantic은 DB 모델과 완전히 독립적이며, 요청 데이터를 검증하는 데 사용됨.
• Django Serializer는 Django 모델과 연동 가능 (ModelSerializer 지원)

pydnatic의 validator 개념 및 사용법

• 구조
  • @field_validator("<필드 이름>"), @classmethod 데코레이터를 사용하여 특정 필드의 값을 검증
  • 검증 함수는 항상 클래스 메서드 형태로 정의되며, 첫 번째 인자로 필드값을 받음
  • 검증 실패 시 예외를 발생시켜야 함 : ValueError, TypeError...
• 예시

from pydantic import BaseModel, field_validator

class ExampleModel(BaseModel):
    number: int

    @field_validator("number")
    @classmethod
    def validator_number(cls, value):
        if value < 1 :
            raise ValueError("Number must be at least 1")
        if value > 100 :
            raise ValueError("Number must not exceed 100")
        return Value

• 동작
  • 모델 초기화 시, 해당 필드의 값이 검증 로직을 통과해야 함
  • 통과하지 못하면 예외 메세지가 반환

example = ExampleModel(number = 50)
print(example) # Output : number 50

• 데이터 검증이 필요한 이유
  • 사용자가 입력하는 데이터가 항상 올바르지는 않음
  • 보안문제 : 잘못 입력된 SQL injection, XSS등 보안 취약점을 유발할 수 있음
  • 비즈니스 로직 보장 : 이메일 형식, 전화번호 국가별 형식, 나이 제한 등

고급 pydantic 검증

• 단일 필드 검증을 넘어 필드 간 관계를 고려한 검증이 필요
• 실제 서비스에서 발생할 수 있는 논리적 오류를 방지
• 보안, 데이터 정합성을 보장하기 위해 더 정교한 검증이 필요

@field_validator

• 비밀번호 검증

from pydantic import BaseModel, field_validator, ValidationError

special_word = "!@#$%^&*()-_+="

class UserRegister(BaseModel):
    username : str
    password : str

    @field_validator("password")
    @classmethod
    def validate_password(cls, value):
        if len(value) < 8:
            raise ValueError("password must be at least 8 characters")
        flag = False
        for char in value:
            if char.isupper():
                flag = True
                break
        if count == 0 :
            raise ValueError("password must contain at least one uppercase letter")

        if not any(char in special_word for char in value):
            raise ValueError("password must contain at least one special character")

        return value

• re 정규식 사용

class UserRegister(BaseModel):
    username: str
    password: str

    @field_validator("password")
    @classmethod
    def validate_password(cls, value):
        if len(value) < 8:
            raise ValueError("Password must be at least 8 characters long")
        if not re.search(r"[A-Z]", value):
            raise ValueError("Password must contain at least one uppercase letter")
        if not re.search(r"\d", value):
            raise ValueError("Password must contain at least one number")
        if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", value):
            raise ValueError("Password must contain at least one special character")
        return value

mode, @model_validator

poetry add email-validator

• before

  • 시점: 데이터 유효성 검증 전에 실행(데이터 받기전에 변환시키고)
  • 데이터 접근 방식: dict 형태로 원시 데이터 접근
  • 예제: 입력데이터 변환(소문자변환, 공백 제거, ...), 타입변환(str -> int)

• after

  • 시점: 데이터 유효성 검증 후 실행(데이터 받고 검사)
  • 데이터 접근 방식: self 객체를 사용해 필드값 접근
  • 예제: 여러 필드간 검증, 값의 논리적 검증, 비즈니스 로직 반영(role 기반 접근 제한)

class UserRegister(BaseModel):
    username : str
    password : str
    confirm_password : str

    @model_validator(mode="after")
    def check_password_match(self):
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self

from pydantic import BaseModel, model_validator, Field, EmailStr

class ContactInfo(BaseModel):
    email : EmailStr | None = None
    phone_number : str | None = None # Optional , 선택적이란 의미

    @model_validator(mode="after")
    def exist_email_phone(self):
        if self.email is None and self.phone_number is None:
            raise ValueError("Must have email or phone_number")
        return self


# no_email = ContactInfo(phone_number="phone_number")
no_phone_number = ContactInfo(email="email@email.com")
# no = ContactInfo()
print(no_phone_number.email)
    
    @model_validator(mode="before")
    @classmethod
    def preprocess_username(cls, data):
        #  객체가 특정 클래스(또는 데이터 타입)의 인스턴스인지 확인하는 함수
        if isinstance(data, dict) and "username" in data:
            data["username"] = data["username"].lower()
        return data

@computed_field

• 자동계산 필드
  • 입렵값 기반으로 자동 계산되는 필드
    • 나이대신 생년원일을 입력받고 자동으로 계산
    • 정가와 할인율을 입력하면 할인가를 자동으로 계산

from pydantic import BaseModel, computed_field
from datetime import datetime

class User(BaseModel):
    name : str
    birth_year : int

    @computed_field
    @property
    def age(self) -> int:
    return datetime.now().yaer - self.birth_year

user = User(name="Alice", brith_year = 2000)
print(user.age) # output : 25

초기값 동적 설정, default_factory

• 초기값을 동적으로 설정할 떄 사용
  • 자동증가 ID
  • created_at을 현재시간으로 설정
  • verification code를 위해 6개의 랜덤숫자 설정
• default_factory는 해당 필드에 값이 제공되지 않았을 때, 기본적으로 어떤 값을 넣을지를 지정하는 기능
• 기본값을 제공하지 않으면 실행되는 기능

from pydantic import BaseModel, Field
from datetime import datetime

class LogEntry(BaseModel):
    message: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

log1 = LogEntry(message="System started")
print(log1.created_at) # 자동생성된 UTC 시간

# 
import uuid
from datetime import datetime
from pydantic import BaseModel, Field

class User(BaseModel):
    # uuid.uuid4() : 랜덤한 UUID 생성
    user_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    role: str = Field(default="user")
    created_at : str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

user1 = User(name="alice")
print(user1)

#
import random
from datetime import datetime, timedelta
from pydantic import BaseModel, Field, field_serializer

class Otp(BaseModel):
    phone_number : str
    otp: int=Field(default_factory=lambda: random.randint(100000,999999))
    otp_expiry: str=Field(default_factory=lambda : (datetime.now()+ timedelta(minutes=5)).isoformat())

    # @field_serializer("otp_expiry")
    # def serialize_datetime(self, value: datetime) -> str:
    #     return value.isoformat()


otp = Otp(phone_number="010")

print(otp)

여러타입 지원

• 특정 필드에 여러 타입 지원
  • product_id가 int또는 str일 경우
  • score가 int또는 float일 경우
• Union 사용

from pydantic import
from typing import Union

class Product(BaseModel):
    product_id: Union[int, str]
    name: str

p1 = Product(product_id=123, name="Laptop")
p2 = Product(product_id="XYZ-456", name="Phone")

print(p1.product_id) # 123
print(p2.product_id) # "XYZ-456"

직렬화, json_encoders, @field_serializer

• 직렬화 Serialization: 다양한 종류의 데이터를 기계가 쓰고 읽기 편리하게 나타낸 방식
  • 기본적으로 datetime, Decimal 등은 JSON으로 직렬화할 수 없음
  • API응답에서 날짜 형식을 통일하거나 소수점 자리수 제한할 때 필요
• json_encoders는 Pydantic에서 특정 데이터 타입을 JSON으로 변환할 때 사용하는 커스텀 인코더
• datetime을 직접 str 형식으로 변환해서 JSON으로 출력하고 싶을 때 사용
• 기본 FastAPI 서버내에서는 기본값 유지

from pydantic import BaseModel
from datetime import datetime

class Order(BaseModel):
    order_id: int
    total_price: float
    created_at: datetime

    # class Config:
    #     json_encoders = {
    #         datetime: lambda v: v.strftime("%Y-%m-%d %H:%M:%S")
    #     }

    @field_serializer("created_at")
    def serialize_datetime(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M:%S")
        
order = Order(order_id=1, total_price=100.5, created_at = datetime.now())
print(order.model_dump_json())

응답 필터링, response_model_exclude

• 응답에서 민감한정보(비밀번호, API Key 등)을 숨기고 싶을 떄
• 유저 권한에 따라 보여줄 데이터 조절
  • admin은 모든 필드
  • 일반 유저는 일부 피륻

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.get("/users/me", response_model=User, response_model_exclude={"password"})
async def get_user():
    user_data = {"username":"testuser","email":"test@test.com","password":"mypassword"}
    return User(**user_data)

# username, email 만 출력됨

JSON & XML 응답 포맷 변환, Response, xml.etree.ElementTree

• API가 JSON 뿐만 아니라 XML도 지원해야할 때
• 사용자가 원하는 응답 포맷을 동적으로 변경하도록 설정

from fastapi import FastAPI
from fastapi.responses import JSONResponse, Response
import xml.etree.ElementTree as ET

app = FastAPI()

@app.get("/data/")
async def get_data(format: str="json"):
    data = {"message": "Hello, FastAPI"}

    if format == "xml":
        root = ET.Element("response")
        message = ET.SubElement(root, "message")
        message.text = data["message"]
        xml_str = ET.tostring(root, encoding="utf-8", method="xml")
        return Response(content=xml_str, media_type="application/xml")

    return JSONResponse(content=data)

# GET /data/?fromat=xml
# <response>
#     <message>Hello, FastAPI</message>
# </response>

다국어 응답처리(Accept-Language)

• 글로벌 API에서는 사용자의 언어에 따라 다른 메세지 반환 필요
• 브라우저 요청 헤더 Accept-Lanugage를 사용하여 자동 감지
• 지원하는 언어가 없으면 기본값을 설정해야 함

from fastapi import FastAPI, Header

app = FastAPI()

    responses = {
        "en": {"message": "Hello, welcome!"},
        "ko": {"message": "안녕하세요, 환영합니다!"},
        "fr": {"message": "Bonjour, bienvenue!"}
    }
@app.get("/greet/")
async def greet(accept_language: str = Header("en")):
    return responses.get(accept_language, responses["en"]) # 기본값 en