Skip to content

Pydantic Boundary Contracts

Pydantic 경계 계약 (Pydantic Boundary Contracts)

Section titled “Pydantic 경계 계약 (Pydantic Boundary Contracts)”
  • HTTP 경계의 입력 계약을 route 함수가 아니라 Pydantic request DTO가 설명하도록 고정한다.
  • proposal_id, ttl_seconds, requested_scopes, tenant_id, limit 같은 교차 관심사를 presentation imperative code에서 걷어낸다.
  • 디버그 시 “왜 실패했는가”를 route 분기 추적이 아니라 모델 validation 에러로 바로 읽을 수 있게 한다.

흐름 (RequestIn -> CommandIn -> Service)

Section titled “흐름 (RequestIn -> CommandIn -> Service)”
flowchart LR
  http["HTTP 입력\n(query · body · header)"] --> reqin["RequestIn\n(Pydantic DTO · domain contract)"]
  reqin -->|validate_request_model| valid{검증}
  valid -->|실패| err["RequestValidationError\n(일관된 422)"]
  valid -->|성공| cmd["CommandIn\n(유스케이스 입력)"]
  cmd --> svc["application Service\n(부수효과는 Temporal로 enqueue)"]
  • 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를 가진다.

Pydantic validation을 FastAPI의 RequestValidationError로 승격하는 공통 helper입니다. 즉 도메인 모델의 ValueError까지도 같은 422 형식으로 통일됩니다.

backend/src/fractalops/foundation/presentation/request_models.py
def validate_request_model(model_type: type[TModel], /, **raw: object) -> TModel:
try:
return model_type.model_validate(raw)
except ValidationError as exc:
raise RequestValidationError(exc.errors()) from exc
except ValueError as exc:
field_name = str(exc).split(" ", 1)[0] if " " in str(exc) else "body"
raise RequestValidationError(
[{"loc": (field_name,), "msg": str(exc), "type": "value_error"}]
) from exc

query parameter alias의 단일 원천입니다. Annotated[..., Query(...)]로 제약을 한 곳에 박아 둡니다.

backend/src/fractalops/foundation/presentation/query_models.py
ProposalIdQuery = Annotated[str, Query(min_length=1)] # proposal_id 강제
TenantIdQuery = Annotated[str, Query()]
AdminLimitQuery = Annotated[int, Query(ge=1, le=200)]

proposal_id가 빈 문자열이면 Query(min_length=1)가 422를 내므로, route 본문에서 if not proposal_id 같은 수동 검증이 사라집니다.

contexts/*/application/** 또는 contexts/*/domain/**

Section titled “contexts/*/application/** 또는 contexts/*/domain/**”

RequestIn, CommandIn, Out을 둡니다. request DTO도 presentation이 아니라 해당 bounded context의 계약으로 관리합니다. 예: ProjectionPublishRequestIncontexts/semantics/application/proposal_contracts.py에 있고 tach.tomlsemantics.domain.proposals 인터페이스로 노출됩니다.

AdminLimitQueryAnnotated[int, ...]이므로 기본값은 route 시그니처에서 명시합니다(타입 alias 자체에 = 50을 박지 않습니다).

def _publish_request(
proposal_id: ProposalIdQuery,
tenant_id: TenantIdQuery = "",
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 함수 본문에 검증 분기가 전혀 없다는 점에 주목하세요. proposal_id 강제, limit 범위(ge=1, le=200)는 전부 타입/모델이 책임집니다.

  • 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는 모델에 둔다.