Marshmallow를 이용해 편리하게 data validation을 구현하는 방법

개발을 하면서 Marshmallow를 왜 사용하게 되었는지, 그리고 Marshmallow를 어떻게 사용하는 지에 대한 방법 소개

Marshmallow를 이용해 편리하게 data validation을 구현하는 방법

안녕하세요, 저는 AX에서 외부 판매 채널들과 연동작업을 진행하고 있는 Noah라고 합니다. 이 글은 제가 AX에서 개발을 하면서 고민하던 부분에 대해서 말씀드리고, 어떤 식으로 해결을 했는지에 대해서 적어 보려고 합니다.

지금이야 프레임워크들이 많이 좋아져서 다양한 기능을 기본적으로 제공하고 있습니다. 프레임워크에서 시키는 대로만 개발을 하면, 자동으로 리퀘스트 데이터에 대해서 유효성 검사도 해 주고, 자동으로 response를 만들어 주고, 심지어 개발 문서도 자동으로 만들어 줍니다.

하지만 옛날부터 프레임워크가 이렇게 많은 걸 해주지 않았습니다.

저희는 flask를 씁니다

Flask는 python기반의 web framework 중에서 사용률 2위를 차지하는 잘 나가는 framework입니다. 이 flask의 특징으로 lightweight가 있습니다. 덕분에 간단하고 빠르게 웹 페이지를 만들 수 있어서, 튜토리얼을 보면 10줄도 안되는 코드로 웹 페이지가 뜹니다.

하지만, 그 반면에 flask는 많은 부분에 있어서 flask를 사용하는 개발자에게 맡기기도 합니다. 덕분에 flask 기반의 라이브러리를 많이 찾아 볼 수 있습니다. http과 관련된 어떤 개념을 사용하고 싶을 때, 그 라이브러리를 가져와서 flask와 사용할 수 있습니다.

그러나 이것은 종종 부작용을 낳는데, 개발자는 flask 기반의 라이브러리를 사용하지 않는 선택을 할 수도 있다는 점입니다. 프로젝트가 개발 속도가 매우 중요한 생명줄이라면, 번거로운걸 모두 건너 뛴 채, 개발 속도에 집중할 수 있습니다. 하지만 개발 속도 뿐 아니라 다양한 요소들이 고려되어야 하는 상황이라면, 아무것도 선택하지 않는 것은 독이 됩니다.

저는 앞에서 언급했듯, AX의 API 팀에서 외부 채널과 연동 작업을 하고 있습니다. 물론 개발 속도 중요합니다. 그렇다고 해서 완성도 같은 측면을 아예 무시해 버릴 수 없습니다. 외부 회사에게 정확하게 스팩대로 작동하는 API를 제공해야 합니다. 그래야지 API의 완성도를 높히면서 커뮤니케이션 코스트를 낮추고 계속 중요한 것을 해 나갈 수 있습니다.

소프트웨어의 구조를 설계해 보자

그래서 저는 web request가 발생해서 flask가 처리를 시작한 시점에서 최종적으로 response를 보내줄 때까지, 개발자가 소프트웨어로서 필요한 부분들에 대해서 단계를 나눌 필요가 있었습니다.

그리고 저희 소프트웨어의 구조는 다음과 같았습니다.

  1. 주소를 확인 if not 404!
  2. 로그인 정보 확인 if not 401!
  3. 권한 확인 if not 403!
  4. 주어진 데이터가 구조가 잘 맞는지 확인 if not 400!
  5. 외부 저장소로부터 필요한 값을 가져옵니다 if not 404!
  6. 로직에 따라서 데이터를 처리한 후
  7. 처리한 결과를 외부 저장소에 다시 적용합니다.
  8. 그리고 response 값을 생성해서 넘겨줍니다. 200!
    (눈치 채셨겠지만, 뒤의 숫자는 http response status code입니다.)

저희는 각각의 단계를 처리해 주기 위해서 다양한 모듈들의 기능을 사용하고 있습니다. 예로 로그인 정보 확인을 위해서는 flask-login이라는 모듈을 사용하고 있습니다. 그리고 저희는 외부 저장소로 PostgreSQL을 사용하고 있었기에, SQL을 사용하기 위해서 ORM 기반의 SQLAlchemy와 flask에서 사용할 수 있도록 Flask-SQLAlchemy를 사용하고 있습니다. (이 부분은 이미 저희 첫 아티클을 통해 접해 보셨을 거라 생각합니다.) 그렇게 각 단계를 해결해 주기 위한 라이브러리를 이미 사용하고 있었습니다. 하지만, 그렇지 않은 곳도 있었습니다. 앞의 구조에서 4번 “주어진 데이터가 구조가 잘 맞는지 확인”과 8번 “response 값을 생성”하는 것이었습니다.

저는 저 부분에 대해서 채워줄 무언가를 찾기 위해 노력했습니다.

Hello, Marshmallow!

마시멜로우, 달콤해서 너무 좋아

Marshmallow 공식 주소 - https://marshmallow.readthedocs.io/en/stable/

드디어 이 글의 주인공을 소개할 수 있게 되었습니다! Marshmallow는 JSON과 같은 raw 데이터를 validate해주고, python의 object로 deserialize해줍니다. 그리고 python object를 raw 데이터로 serialize 해주는 역할을 합니다. 너무 정확하게도 제가 필요했던 기능인 “주어진 데이터 구조 확인(4번)”과 “response 값 생성(8번)”을 해 주는 라이브러리 입니다.

이제 간단히 어떻게 사용하는 지에 대해서 예시를 통해서 소개하겠습니다.

from marshmallow import Schema, fields

class SubtaskSchema(Schema):
    name = fields.Str()
    is_done = fields.Boolean(default=False)

class TaskSchema(Schema):
    name = fields.Str()
    is_done = fields.Boolean(default=False)
    subtask_list = fields.Nested(SubtaskSchema(many=True))

데이터 구조는 이미 구면이라 생각합니다. 이전 글에서 사용했던 데이터와 같은 작업과 그 밑에 하위 작업이 있는 구조입니다. 단지, SQLAlchemy가 아니라, Marshmallow로 작성되었을 뿐입니다. 이제 Schema들을 이용해서 필요했던 기능들이었던 4번과 8번을 해보겠습니다.

task_schema = TaskSchema()
task = {
    "name": "task1",
    "is_done": False,
    "subtask_list": [
        {
            "name": "task1-1",
            "is_done": True
        },
        {
            "name": "task1-2",
            "is_done": False
        }
    ]
}
task_validated = task_schema.load(task)

Schema에 정의되어있는 함수인 load를 이용해서 task의 구조를 validation하고 deserialize할 수 있습니다. 이것은 앞의 4번 과정을 해 줄 수 있습니다.

return jsonify(
    task_schema.dump(
        task_validated
    )
)

또한, schema의 dump함수를 이용해서 serialize를 할 수 있습니다. jsonify는 파이썬 dict object를 json으로 바꿔 HTTP Response를 만들어 주는 함수입니다. 이것을 통해, 우리는 역시 8번 과정인 response를 생성할 수 있게 되었습니다.

하지만 이것만으로 Marshmallow의 매력을 나타내기에는 부족한 부분이 많습니다. 좀 더 알아보죠.

Marshmallow의 장점

Marshmallow를 통해서 validation을 하면서 일관적인 플로우를 통해 데이터가 잘못되었을 때에 대한 처리를 할 수 있습니다. 한 가지 예시를 들어보겠습니다.

invalid_task = {
    "name": "task1",
    "is_done": "Done"
}
task_schema.load(invalid_task)

이 코드를 확인해보면 한가지 이상하다는 것을 알 수 있습니다. is_doneboolean타입이어야 합니다. 하지만, boolean 타입이 아닙니다. Marshmallow의 경우 이럴때 ValidationError 라는 에러를 발생시킵니다. 게다가 어디가 어떻게 잘못되었는지를 자세하게 제공까지 합니다. 다음은 에러 메세지입니다.

marshmallow.exceptions.ValidationError: {'is_done': ['Not a valid boolean.']}

에러 메세지 구조를 보고 어느 필드가 어떻게 잘못되었는지에 대해서 알 수 있습니다. 만약 저 에러메세지의 내용을 API를 통해 제공해 준다면 data의 어느 부분이 기대한 구조와 맞지 않았다는 것을 일괄적으로 제공할 수 있을 것입니다. 이것은 특히 데이터가 schema에서 명세한 것과 여러 부분에서 다를 때에는 더 풍부해 집니다.

invalid_task = {
    "name": "task1",
    "is_done": False,
    "subtask_list": [
        {
            "name": "task1-1",
            "is_done": "Done"
        },
        {
            "name": 12,
            "is_done": False
        }
    ]
}
task_schema.load(invalid_task)

subtask에 여러 곳에 값이 이상합니다! 이런 경우일 때에도 융통성 있게 schema와 다른 부분을 모두 차근히 알려줍니다.

marshmallow.exceptions.ValidationError: {
    'subtask_list': {
        0: {'is_done': ['Not a valid boolean.']},
        1: {'name': ['Not a valid string.']}
    }
}

이 정보를 API로 제공해 준다면, 어느 부분에서 자신이 데이터를 잘못 보냈는지 API를 사용하는 개발자가 확실히 알 수 있습니다. 이제 이 정보를 API를 통해서 제공해 보겠습니다.

@app.post("/task")
def view():
    try:
        request_data = task_schema.load(request.json)
    except ValidationError as e:
        return jsonify(e.data), 401
    
    # do_something_awesome()

eValidationError 타입의 에러를 나타내는 object로 e.data 에 방금 확인했던, 어떤 필드의 값이 어떻게 이상한지에 대한 데이터를 담고 있습니다. 이제 validation을 실패 했을 때 어떤 데이터가 schema와 맞지 않는지를 API를 통해 제공할 수 있게 되었습니다.

아! 아직 완벽한 코드는 아닙니다. 저대로라면, 모든 endpoint에 대해서 validation 에러 처리해주는 코드를 넣어줘야 합니다. DRY (Don’t Repeat Yourself) 원칙에 위배되기에 살짝 수정해 보겠습니다.

@app.errorhandler(ValidationError)
def handle_validation(e):
    return jsonify(e.data), 401

@app.post("/task")
def view():
    request_data = task_schema.load(request.json)
  
    # do_something_awesome()

# add new awesome endpoint!

app.errorhandler라는 함수를 통해 어떤 endpoint에서 실행해도 해당 에러가 발생하면 fallback으로 실행할 함수를 지정할 수 있습니다. 이제 endpoint를 많이 만들어도 간단하게 validation을 해 주는 코드가 자연스럽게 녹아들어 가 있습니다.

Marshmallow를 사용하면서

Marshmallow를 사용하면서 저런 좋은 점도 있었지만, 불편한 점도 있었습니다.

가장 큰 문제는 앞의 구조를 생각하지 않고 손 가는 대로 쓰던 습관을 버려야 하는 것이었습니다. 대표적으로는, 더 이상 schema.dump를 한 데이터를 후 처리 하면 안된다는 점입니다.  Marshmallow를 사용하는 이유가 API를 통해 주고 받는 데이터들에 대한 명세였는데, 그것을 위배해 버리니까요.

# 완료된 task에는 `완료`라는, 완료되지 않은 task에는 `미완료`라는 값을 가지는
# `status_name`이라는 필드를 API에 추가해 달라는 요청을 받았습니다.

# Do not try at home
# view 함수에서 dump이후에 `status_name` 키를 추가합니다.
task_dump = task_schema.dump(task_validated)
task_dump['status_name'] = "완료" if task_validated.is_done else "미완료"
return jsonify(task_dump)

# Do it yourself
# Schema에서 status_code라는 필드를 추가합니다.
class TaskSchema(Schema):
    # ... 생략
    status_name = fields.Function(lambda obj: "완료" if obj.is_done else "미완료")

그 외에도 모듈이 여러 개가 되면서 그에 따른 라이브러리 추가도 있었습니다. Marshmallow의 도입으로 인해 추가로 필요해진 모듈이 2개가 더 있었습니다. 그것이 하나는 Marshmallow와 SQLAlchemy를 같이 쓸 수 있도록 해주는 라이브러리였고 다른 하나는, Marshmallow와 Flask를 같이 쓸 수 있도록 한 라이브러리였습니다. 그런데, SQLAlchemy도 Flask와 같이 쓸 수 있도록 해주는 라이브러리가 있다보니 이 3개의 라이브러리 간의 삼각관계도 조심스러웠습니다. 다행히 현재 각 라이브러리들이 활발하게 업데이트를 해 주고 있기 때문에 라이브러리끼리 의 충돌은 최소화 하면서 유지보수를 할 수 있었습니다.

마치며

Marshmallow를 도입하면서 했던 고민들 덕분에 어떻게 하면 더 체계적이고 구조적으로 개발을 할 수 있을 지에 대해서 더 잘 알 수 있게 되었습니다. 덕분에 꼭 Flask가 아니라 어떤 framework를 쓰더라도, 기능들이 어떤 목적으로 제공되는지 더 잘 이해하게 되는 계기가 되었습니다.

저는 앞으로도 계속 개발하면서 불편한 부분을 끊임없이 찾아내고 문제화하여 해결책을 찾기 위해 노력할 것입니다. 그 때에는 또 다른 답을 찾으려 할 것이고, 저보다 더 치열하게 고민하고 행동하여 답을 라이브러리로서 제시한 사람들을 보며 영감을 받게 될 것입니다.

지금까지 Noah였습니다. 읽어주셔서 감사합니다.

액스(AX)로 다 채널 여행 상품 동시 판매 및 운영


액스는 투어 & 액티비티 판매자의 온라인 판매를 위한 최적화된 다 채널 자동 판매 서비스 입니다.
지금 무료로 시작해보세요. 써보면서 이해하는 것이 가장 빠릅니다!