Flask에는 validation 기능이 내장되어 있지 않으므로 별도의 라이브러리를 사용해야 합니다. flask-request-validator, flask-parameter-validation, flask-inputs, flask-pydantic 등의 오픈소스 라이브러리들이 이미 존재하는데, 이 중에 flask-pydantic이 fastapi의 validation과 비슷한 콘셉트로 가장 만족도가 높은 라이브러리가 될 것 같습니다.
어쨌든 이러한 기능을 직접 만들어 보고 싶었는데, 요 몇 년간 python web framework 중 가장 핫한 fastapi에는 request parameter를 아주 우아하게 validation 하는 기능이 있으니 이를 흉내 내어 만들기로 하였습니다. 전체 코드는 ooknimm/flask-parameter-validator에서 확인할 수 있습니다.
fastapi 유효성 검사
fastapi는 type annotation + pydantic으로 함수 parameter를 정의합니다. API가 호출되면 request parameter를 validation하고 해당 API 함수에 데이터를 주입합니다.
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str
price: int
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return item.model_dump()type annotation으로 인해 데이터가 명확해지고 데이터 정의만 해주면 유효성 검사를 함수 body 밖에서 자동으로 해주니 가독성과 생산성이 향상됩니다. 참 멋진 유효성 검사 기능입니다.
API
앞서 봤던 코드에서 fastapi 부분을 flask로 변경합니다.
from flask import Flask, jsonify
from pydantic import BaseModel
from flask_parameter_validator import parameter_validator # new
class Item(BaseModel):
name: str
description: str
price: int
app = Flask(__name__)
@app.post("/items/")
@parameter_validator # new
def create_item(item: Item):
return jsonify(item.model_dump())fastapi와 다르게 여기서는 parameter_validator 데코레이터가 추가되었습니다. fastapi는 app.route 내부에서 validation을 하는데 flask의 app.route는 해당 기능이 없어서 데코레이터를 추가하여 기능을 더합니다. 이때 주의할 점은 parameter_validator가 app.route 아래에 위치해야 합니다.
기능에서 기대하는 input(request parameter)과 output(function parameter)이 분명합니다. 이럴 땐 테스트 코드를 먼저 작성한 후 개발을 진행하는 것이 현명합니다.
테스트 코드
# test_body_param.py
from flask_parameter_validator import parameter_validator
import pytest
from flask import Flask, jsonify
app = Flask(__name__)
client = app.test_client()
class Item(BaseModel):
name: str
description: str
price: int
@app.post("/items")
@parameter_validator
def create_item(item: Item):
return jsonify(item.model_dump())
@pytest.mark.parametrize(
"path,body,expected_status,expected_response",
[
("/items", {"name": "foo", "price": 3000, "description": "bar"}, 200, {"name": "foo", "price": 3000, "description": "bar"}),
],
)
def test_single_body_param(path, body, expected_status, expected_response):
response = client.post(path, json=body)
assert response.status_code == expected_status
assert response.get_json() == expected_response하나의 API에 대한 하나의 테스트 케이스를 정의한 간단한 테스트 코드입니다. pytest.mark.parametrize를 사용하면 테스트 케이스를 간단히 여러 개 추가할 수 있습니다. 이제 이 테스트를 통과시키도록 코드를 작성하면 됩니다.
parameter_validator
이제 본격적인 기능 개발을 하겠습니다. 우선 parameter_validator가 데코레이터 역할을 하도록 코드를 작성합니다.
from functools import update_wrapper
from typing import Callable, Any
class ParameterValidator: # 1
def __init__(self, call: Callable[..., Any]) -> None:
self.call = call
update_wrapper(self, call) # 3
def __repr__(self) -> str: # 3
return repr(self.call)
def parameter_validator(func) -> ParameterValidator: # 2
return ParameterValidator(func)- validation 관련 데이터를 저장했다가 호출할 때 데이터를 사용해야 하는데 closure 혹은 global로 저장하는 방식은 적절하지 않습니다. 그래서 클래스로 정의하였습니다.
- 클래스를 직접 데코레이터로 써도 되지만 데코레이터에 파스칼 케이스 사용을 피하고자 함수로 감쌌습니다.
- update_wrapper를 사용하고 __repr__를 정의한 이유는 데코레이터가 타겟 함수의 여러 속성을 덮어쓰지 않도록 하기 위함입니다. 이 부분에 대해 이해가 없다면 wraps에 관한 글을 읽어보면 좋습니다.
ParameterValidator
ParameterValidator 클래스를 구현합니다. fastapi 소스 코드를 확인하여 전반적으로 참고하였습니다.
from functools import update_wrapper
from flask import Response
from pydantic import BaseModel
from typing import Callable, Any, Dict
class Dependant:
def __init__(self):
self.body_params: Dict[str, BaseModel] = {}
class ParameterValidator:
def __init__(self, call: Callable[..., Any]) -> None:
self.call = call
update_wrapper(self, call)
self.dependant: Dependant = self.get_dependant() # 1
def get_dependant(self): ...
def solve_dependencies(self): ...
def __call__(self, *args, **kwargs) -> Response:
solved, errors = self.solve_dependencies() # 2
if errors:
return Response(json.dumps({"detail": errors}), status=422, mimetype="application/json") # 3
return self.call(*args, **{**kwargs, **solved}) # 4
def __repr__(self) -> str:
return repr(self.call)- get_dependant 메소드에서 함수의 signature를 분석하고 Dependant 클래스에 signature의 type annotation을 저장합니다.
- API가 호출되면 API함수를 wrapping한 ParameterValidator가 호출되게 됩니다. 그러므로 ParameterValidator가 callable 해야 하는데 __call__ 매직 메소드는 클래스를 callable 하게 만듭니다. solve_dependencies 메소드는 get_dependant에서 저장된 type annotation으로 request parameter를 validation 합니다.
- validation을 통과하지 못했다면 에러 내용과 422 Unprocessable Entity를 반환합니다.
- API를 호출할 때 validation된 데이터를 주입합니다.
get_dependant
ParameterValidator가 초기화될 때 타겟 함수의 signature를 분석합니다. python 내장 라이브러리인 inspect를 사용하면 손쉽게 함수 parameter의 type annotation을 파악할 수 있습니다.
import inspect
class ParameterValidator:
# 중략
def get_dependant(self) -> Dependant:
dependant = Dependant()
func_signatures = inspect.signature(self.call)
signature_params = func_signatures.parameters
for param_name, param in signature_params.items():
annotation = param.annotation
dependant.body_params[param_name] = annotation
return dependantAPI함수 parameter의 type annotation을 dependant.body_params에 저장합니다.
solve_dependencies
from pydantic import ValidationError
from pydantic_core import ErrorDetails
class ParameterValidator:
# 중략
def solve_dependencies(self) -> Tuple[Dict[str, BaseModel], List[ErrorDetails]]:
solved: Dict[str, BaseModel] = {}
errors: List[ErrorDetails] = []
received_body = request.get_json()
for param_name, param in self.dependant.body_params.items():
try:
validated_model = param.model_validate(received_body)
solved[param_name] = validated_model
except ValidationError as exc:
errors.extend(exc.errors())
return solved, errorsrequest의 body를 dependant.body_params에 저장된 type annotation으로 validation 합니다. model_validate은 pydantic.BaseModel의 메소드이고 validation 실패 시 ValidationError가 발생합니다. validation에 성공했다면 solved에 추가하고, ValidationError가 발생했다면 errors 목록에 추가합니다.
전체 코드
첫 번째 테스트 코드가 통과되도록 기능을 개발했습니다. 전체 코드는 다음과 같습니다.
import inspect
import json
from functools import update_wrapper
from typing import Any, Callable, Dict, List, Tuple
from flask import Response, request
from pydantic import BaseModel, ValidationError
from pydantic_core import ErrorDetails
class Dependant:
def __init__(self):
self.body_params: Dict[str, BaseModel] = {}
class ParameterValidator:
def __init__(self, call: Callable[..., Any]) -> None:
self.call = call
update_wrapper(self, call)
self.dependant: Dependant = self.get_dependant()
def get_dependant(self) -> Dependant:
dependant = Dependant()
func_signatures = inspect.signature(self.call)
signature_params = func_signatures.parameters
for param_name, param in signature_params.items():
annotation = param.annotation
dependant.body_params[param_name] = annotation
return dependant
def solve_dependencies(self) -> Tuple[Dict[str, BaseModel], List[ErrorDetails]]:
solved: Dict[str, BaseModel] = {}
errors: List[ErrorDetails] = []
received_body = request.get_json()
for param_name, param in self.dependant.body_params.items():
try:
validated_model = param.model_validate(received_body)
solved[param_name] = validated_model
except ValidationError as exc:
errors.extend(exc.errors())
return solved, errors
def __call__(self, *args, **kwargs) -> Response:
solved, errors = self.solve_dependencies()
if errors:
return Response(json.dumps({"detail": errors}), status=422, mimetype="application/json")
return self.call(*args, **{**kwargs, **solved})
def __repr__(self) -> str:
return repr(self.call)
def parameter_validator(func) -> ParameterValidator:
return ParameterValidator(func)테스트
이제 처음에 작성한 테스트 코드로 테스트해 봅니다.
pytest -v test_sample.py
통과했습니다.
fastapi 같은 request parameter validator의 기초적인 부분을 개발하였습니다. request parameter validator의 핵심은 데코레이터, pydantic, inspect, type annotation입니다.
- 데코레이터: 타겟 함수의 기능을 확장하고 타겟 함수에 데이터를 의존성 주입(DI)합니다.
- pydantic: 데이터 구조체 정의, validation을 합니다.
- inspect: 데코레이터에서 inspect로 타겟 함수의 signature 중 parameter를 파악합니다.
- type annotation: parameter에 type annotation으로 데이터를 정의하여 데코레이터에서 확인할 수 있도록 합니다.
아직 실전에 사용하기에는 기능이 부족합니다. 다음과 같은 경우를 처리할 수 있어야 합니다.
- parameter type annotation으로 pydantic을 사용하지 않는 경우
- form data인 경우
- path, query, header의 경우
- multiple request parameter의 경우
다음 편에서 이러한 기능을 어떻게 추가할 수 있는지 설명하겠습니다.
'Python' 카테고리의 다른 글
| Flask Request Parameter 우아하게 검사하기 - 2 (0) | 2023.09.25 |
|---|
