Pydantic Validators에 컨텍스트를 전달하는 방법
Source: Dev.to
위에 제공된 링크만으로는 번역할 실제 텍스트가 포함되어 있지 않습니다. 번역이 필요한 전체 내용(마크다운 형식 포함)을 제공해 주시면, 요청하신 대로 한국어로 번역해 드리겠습니다.
호출하는 사람에 따라 다르게 동작해야 하는 함수가 있었습니다.
함수가 무엇을 하는지는 상관없고, 오류를 기록할지 조용히 할지만 달랐습니다.
간단해 보였죠. 그렇지 않았습니다.
설정
우리는 전화번호 정규화 함수를 가지고 있습니다. 이 함수는 다양한 데이터 소스에서 들어오는 지저분한 전화번호들을 일관된 형식으로 변환합니다. 때때로 전화번호가 쓰레기처럼(자리수가 부족하거나, 형식이 틀리거나, 무작위 텍스트가 포함된) 될 수 있습니다. 이런 경우 함수는 원래 값을 그대로 반환하고 다음 단계로 넘어갑니다.
이 함수는 Pydantic 모델에 BeforeValidator 로 연결되어 있습니다:
from typing import Annotated
from pydantic import BeforeValidator
PhoneNumber = Annotated[str, BeforeValidator(normalize_phone_number)]Pydantic은 전화번호 필드가 있는 모델을 생성할 때마다 자동으로 이 함수를 호출합니다. 우리는 언제, 어떻게 호출되는지를 제어하지 않으며—Pydantic은 값을 전달하고 반환값을 기대할 뿐입니다.
문제
이 함수는 두 가지 매우 다른 상황에서 실행됩니다:
- API 요청 – 사용자에게 데이터를 제공합니다. 잘못된 전화번호가 예상됩니다(원본 데이터가 지저분할 수 있습니다). 여기서 모든 잘못된 번호를 로그에 기록하면 각 요청마다 로그가 소음으로 가득 차게 됩니다.
- 데이터 검증 파이프라인 – 배포하기 전에 새로운 데이터 파일을 검사합니다. 여기서는 모든 잘못된 번호를 보고 싶으며 문제를 조기에 발견할 수 있습니다.
같은 함수. 같은 Pydantic 모델. 두 가지 다른 요구사항.
명백한 아이디어들 (그리고 왜 작동하지 않는가)
1. 매개변수 추가
def normalize_phone_number(value, log_errors=False):
...BeforeValidator는 함수에 단일 인수 (value)만 전달합니다. 추가 인수를 전달할 수 없으며, 시그니처가 고정되어 있습니다.
2. 전역 플래그 사용
_log_errors = False
def enable_logging():
global _log_errors
_log_errors = True한 번에 하나의 요청만 처리할 때만 작동합니다. FastAPI에서는 여러 요청이 동시에 실행되므로, 요청 A에 의해 설정된 플래그가 요청 B에서도 보이게 됩니다(비록 B는 로깅을 요청하지 않았더라도). 전역 변수는 모든 동시 요청 간에 공유되므로 우리가 원하는 바가 아닙니다.
3. 함수 복제
def normalize_phone_number(value):
... # no logging
def normalize_phone_number_with_logging(value):
... # with logging이제 정규화 로직을 중복하게 됩니다. 함수가 BeforeValidator를 통해 Pydantic 모델에 바인딩되어 있기 때문에, 검증 파이프라인 전용 별도의 모델이 필요합니다. 이는 단순한 로깅 플래그를 위해 많은 절차를 추가하는 셈입니다.
ContextVar 사용하기
Python의 contextvars 모듈(3.7부터 표준 라이브러리) 은 실행 컨텍스트마다 격리된 변수를 제공합니다. 하나의 async 작업에서 ContextVar 값을 설정하면 동시에 실행되는 다른 작업에서는 그 값을 볼 수 없으며, 각 작업이 자체 복사본을 갖게 됩니다.
다음이 해결 방법입니다:
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any
import logging
logger = logging.getLogger(__name__)
_validation_logging_enabled: ContextVar[bool] = ContextVar(
"validation_logging_enabled", default=False
)
@contextmanager
def enable_validation_logging():
token = _validation_logging_enabled.set(True)
try:
yield
finally:
_validation_logging_enabled.reset(token)
def log_validation_error_if_enabled(message: str, **kwargs: Any):
if _validation_logging_enabled.get():
logger.error(message, **kwargs)이것이 전체 모듈이며, 표준 라이브러리 외에 추가 의존성이 없습니다.
어떻게 함께 작동하는가
정규화 함수는 조건부 로거를 호출합니다:
def normalize_phone_number(value):
...
try:
parsed = phonenumbers.parse(value)
except NumberParseException as e:
log_validation_error_if_enabled(
"Invalid phone number format", value=value, error=str(e)
)
return value
...데이터‑검증 파이프라인은 작업을 컨텍스트 매니저로 감쌉니다:
def validate_incoming_data(data, schema):
with enable_validation_logging():
result = validate_records(data, model=schema)
return result
def validate_records(data, model):
"""Validate each data record through a Pydantic model."""
...with 블록 내부에서는 오류가 기록됩니다. 블록 외부(예: API 요청 중)에서는 기록되지 않습니다. 호출 체인에 매개변수를 전달할 필요가 없으며, 신경 쓸 전역 상태도 없습니다.
Source:
토큰 패턴
이해하면 좋은 세부 사항: set() / reset(token) 패턴.
token = _validation_logging_enabled.set(True)
# ... do work ...
_validation_logging_enabled.reset(token)set()은 이전 값을 나타내는 토큰을 반환합니다. reset(token)은 그 이전 값을 복원하는데, 반드시 False일 필요는 없습니다. 따라서 중첩된 enable_validation_logging() 호출도 올바르게 동작합니다 — 내부 호출은 외부 호출에 의해 설정된 값을 복원합니다.
이를 try/finally 블록으로 감싸면 예외가 발생하더라도 정리 작업이 보장됩니다.
Python 3.14+ 단축형
Python 3.14부터는 토큰 자체가 컨텍스트 매니저가 되므로 래퍼를 생략할 수 있습니다:
with _validation_logging_enabled.set(True):
result = validate_records(data, model=schema)with 블록이 종료될 때 자동으로 reset(token)을 호출합니다. 3.14 이상을 사용하고 있다면 가장 간결한 형태입니다.
트레이드‑오프
단점을 솔직히 말하자면, 이것은 “거리에서의 행동” 입니다. 정규화 함수가 매개변수로 받지 않은 플래그를 읽습니다. 모듈을 처음 보는 경우, 로깅이 함수 외부의 무언가에 의해 제어된다는 것을 깨닫지 못할 수 있습니다.
좋은 docstring이 도움이 됩니다. 그리고 우리의 사용 사례(단순한 불리언 플래그, 두 개의 명확한 컨텍스트)에서는 이 트레이드‑오프가 충분히 가치가 있습니다. 대안은 로깅 플래그를 전달하기 위해 Pydantic 검증 방식을 재구성하는 것이 될 것입니다.
ContextVar를 사용해야 할 때
ContextVar는 다음과 같은 경우에 적합합니다:
- 제어할 수 없는 레이어(예: Pydantic 검증기, 미들웨어, 서드파티 콜백)를 통해 컨텍스트를 전달해야 할 때.
- 비동기 환경에서 전역 상태가 안전하지 않을 때.
- 컨텍스트가 단순할 때(플래그, 요청 ID, 상관 토큰 등).
- 컨텍스트가 교차 관심사(로깅, 트레이싱, 모니터링)를 위한 것이고, 비즈니스 로직을 위한 것이 아닐 때.
복잡한 상태 관리를 위해서는 적절하지 않습니다. 오류를 리스트에 수집하거나 풍부한 컨텍스트를 구축해야 한다면 더 좋은 패턴이 있습니다. 그러나 “지금 이 코드 경로가 약간 다르게 동작해야 하는가?”와 같은 경우에는 정확히 맞는 선택입니다.
테스트하기
테스트는 동작의 양쪽을 모두 검증합니다. 우리는 pytest의 caplog 픽스처를 사용하여 로그 레코드를 캡처하고, 메시지가 출력되었는지 확인할 수 있습니다:
class TestLogPhoneNumberErrors:
def test_no_logging_outside_context(self, caplog):
normalize_phone_number("not a phone number")
assert caplog.records == []
def test_logs_error_inside_context(self, caplog):
with enable_validation_logging():
normalize_phone_number("not a phone number")
assert "Invalid phone number format" in caplog.text
def test_context_manager_resets_after_exit(self, caplog):
with enable_validation_logging():
pass
normalize_phone_number("another invalid number")
assert caplog.records == []기본값은 False(무음)이며, 로깅을 원할 경우 직접 활성화합니다. 기존 코드는 동작이 변경되지 않습니다.
마무리
전체 변경 사항은 작은 새 모듈과 몇 개의 테스트 케이스였습니다. 새로운 의존성은 없습니다. 데이터‑검증 파이프라인이 이제 잘못된 전화번호를 일찍 잡아내고, API 요청 로그가 깔끔하게 유지됩니다.
때로는 올바른 해결책이 새로운 라이브러리나 영리한 아키텍처가 아닐 수도 있습니다. 때로는 아직 사용해 보지 않은 표준 라이브러리 모듈일 수도 있습니다.
프로젝트에서 ContextVar를 사용해 본 적이 있나요? 댓글로 사용 사례를 알려 주세요.