API Gateway → Lambda → Pydantic 모델 연결 흐름 완전 이해하기

Sue·2025년 7월 24일

API Gateway → Lambda → Pydantic 모델 연결 흐름에 대한 전체 구조를, 개념 → 원리 → 예시 코드 → 실전 활용 포인트까지 정리

AWS Lambda + API Gateway에서 Pydantic을 활용한 입력 검증 흐름 정리

1. 전체 흐름 개요

Lambda 기반의 API는 보통 다음과 같은 구조로 동작한다:

클라이언트가 API Gateway를 통해 HTTP 요청을 보냄
API Gateway는 이 요청을 event 객체 형태로 Lambda에 전달
Lambda 함수는 이 event에서 필요한 데이터를 추출하고 처리함
추출한 입력값을 Pydantic 모델로 검증하고,
처리 결과를 HTTP 응답으로 반환

2. API Gateway 이벤트 구조 이해

API Gateway는 Lambda 함수에 요청 정보를 event 딕셔너리 형태로 전달한다.
이 event 안에는 요청 관련 다양한 정보가 들어 있다.

예시 요청: GET /users?age=25

전달되는 event 구조:

{
  "resource": "/users",
  "path": "/users",
  "httpMethod": "GET",
  "queryStringParameters": {
    "age": "25"
  },
  "body": null,
  "headers": {
    "Content-Type": "application/json"
  }
}

필요한 정보는 다음 경로에서 찾을 수 있다:

데이터 종류	위치	예시
쿼리 파라미터 (`?age=25`)	`event["queryStringParameters"]`	`{"age": "25"}`
요청 바디 (POST 등)	`event["body"]`	문자열로 인코딩된 JSON
경로 파라미터 (`/users/{id}`)	`event["pathParameters"]`	`{"id": "123"}`

3. Lambda 함수에서 데이터 추출 위치

요청의 종류에 따라 추출 위치가 다르다:

GET /users?age=25 → event["queryStringParameters"]
POST /users + JSON body → json.loads(event["body"])
GET /users/123 → event["pathParameters"]["id"]

추출 후에는 문자열로 넘어오기 때문에, 숫자 등으로 형 변환이 필요하다.

이 부분에서 Pydantic이 자동 타입 변환과 검증을 해줄 수 있다.

4. Pydantic으로 입력 검증하는 위치와 방식

BaseModel을 상속받아 입력 스키마를 정의한 뒤, 모델(**입력 딕셔너리) 형식으로 검증할 수 있다.

예시:

from pydantic import BaseModel

class QueryParams(BaseModel):
    age: int

그 후 Lambda 내부에서:

params = QueryParams(**event["queryStringParameters"])

이렇게 하면 다음을 자동으로 해준다:

age가 int로 변환 가능한지 확인
누락되었는지 검사
실패하면 ValidationError 발생

5. 예외 처리 및 상태 코드 설계

검증이 실패했을 때는 Pydantic이 ValidationError를 발생시킨다.

이를 try/except로 감싸고, 적절한 HTTP 상태 코드로 응답하면 된다.

400 Bad Request: 파라미터 누락 또는 잘못된 요청 구조
422 Unprocessable Entity: 파라미터 구조는 맞지만 값이 부적절

예시:

from pydantic import ValidationError

try:
    params = QueryParams(**event["queryStringParameters"])
except ValidationError as ve:
    return {
        "statusCode": 422,
        "body": json.dumps({"message": "Invalid input", "detail": str(ve)})
    }

6. 전체 예제 코드

import json
from pydantic import BaseModel, ValidationError
from http import HTTPStatus

class QueryParams(BaseModel):
    age: int

def handler(event, context):
    try:
        params = QueryParams(**event.get("queryStringParameters", {}))

        return {
            "statusCode": HTTPStatus.OK,
            "body": json.dumps({"message": f"You are {params.age} years old"})
        }

    except ValidationError as ve:
        return {
            "statusCode": HTTPStatus.UNPROCESSABLE_ENTITY,
            "body": json.dumps({"message": "Invalid input", "detail": str(ve)})
        }

    except Exception as e:
        return {
            "statusCode": HTTPStatus.INTERNAL_SERVER_ERROR,
            "body": json.dumps({"message": "Internal server error"})
        }

7. 정리 및 실무 팁

항목	내용
이벤트 구조 출력해보기	`print(json.dumps(event, indent=2))` 로 구조 확인
Pydantic 사용 위치	쿼리스트링, 경로 파라미터, 바디 파싱 후에 검증용으로 사용
try/except 구조	`ValidationError`, `KeyError`, `Exception` 순서대로 잡기
응답 통일성	항상 `{ "statusCode", "body (stringified json)" }` 형식 유지
바디 파싱 시	`json.loads(event["body"])` 필요함 (POST/PUT 요청의 경우)