Skip to content

Pydantic Boundary Contracts

  • HTTP 경계의 입력 계약을 route 함수가 아니라 Pydantic request DTO가 설명하도록 고정한다.
  • proposal_id, ttl_seconds, requested_scopes, tenant_id, limit 같은 교차 관심사를 presentation imperative code에서 걷어낸다.
  • 디버그 시 “왜 실패했는가”를 route 분기 추적이 아니라 모델 validation 에러로 바로 읽을 수 있게 한다.
  • write route는 가능한 한 RequestIn -> CommandIn -> Service 순서만 가진다.
  • route 함수는 입력 병합과 서비스 호출만 담당한다.
  • query/body/header 정규화는 RequestIn validator나 공통 dependency helper에서 수행한다.
  • proposal_id 강제는 HTTPException(400, "proposal_id_required") 대신 Field(min_length=1) 기반 validation으로 표현한다.
  • 내부 이벤트/스케줄 입력처럼 runtime이 자동 생성하는 command는 optional proposal_id를 가질 수 있다.
  • 사람이나 UI가 직접 치는 write request는 route 전용 RequestIn에서 required proposal_id를 가진다.
  • foundation/presentation/request_models.py
    • validate_request_model()
    • Pydantic validation을 RequestValidationError로 승격하는 공통 helper
  • foundation/presentation/query_models.py
    • ProposalIdQuery
    • TenantIdQuery
    • AdminLimitQuery
    • query parameter alias의 단일 원천
  • contexts/*/domain/**
    • RequestIn, CommandIn, Out를 둔다.
    • request DTO도 presentation이 아니라 해당 bounded context domain contract로 관리한다.
def _publish_request(
tenant_id: TenantIdQuery = "",
proposal_id: ProposalIdQuery,
limit: AdminLimitQuery = 50,
) -> ProjectionPublishRequestIn:
return ProjectionPublishRequestIn(
tenant_id=tenant_id,
proposal_id=proposal_id,
limit=limit,
)
@router.post("/publish")
def publish_projection(
query: ProjectionPublishRequestIn = Depends(_publish_request),
service=Depends(get_projection_publish_service),
) -> dict:
return service.publish(
tenant_id=resolve_tenant_id(query.tenant_id),
proposal_id=query.proposal_id,
limit=query.limit,
)
  • route 본문에서 if not proposal_id 같은 수동 검증
  • body와 query를 route 내부에서 ad-hoc로 merge
  • presentation 파일 안에 route 전용 contract truth를 중복 선언
  • 동일 의미의 입력이 route마다 다른 에러 형식으로 떨어지는 것
  • 외부 provider webhook처럼 wire format이 강하게 고정된 ingress는 codec/adapter 레이어에서 shape 변환을 허용한다.
  • 이 경우에도 route는 codec 호출만 하고 validation truth는 모델에 둔다.