https://ooknimm.tistory.com/7 글에서 이어집니다. 전체 코드는 ooknimm/flask-parameter-validator에서 확인할 수 있습니다.
추가해야 하는 기능은 다음과 같습니다.
- parameter type annotation으로 pydantic model 외에도 python typing 지원
- form data 지원
- path, query, header를 지원
- multiple request parameter를 지원
1편에서는 기초적인 기능만 존재하여 코드가 심플했습니다. 이제 보다 복잡한 요구사항을 수행할 수 있어야 하므로 ParameterValidator와 Dependant만 존재하는 클래스 구조에서 확장하였습니다. 클래스 관계를 직관적으로 확인하기 위해 UML을 그려보겠습니다.
UML
클래스 UML을 잘 모른다면 클래스 UML에 대한 글을 참고바랍니다. 클래스 구조는 fastapi의 내부를 참고하여 디자인하였습니다.
여기서 핵심적인 역할을 하는 클래스는 ParameterValidator, Dependant, FieldAdapter입니다.
- ParameterValidator
- 데코레이터로써 함수의 기능을 확장합니다.
- 타겟 함수의 parameter를 분석해 Dependant에 저장합니다.
- API가 호출되면 Dependant에 validation을 요청하고 validation 된 데이터를 타겟 함수에 주입합니다.
- Dependant
- FieldAdapter를 저장하고 validation하는 클래스
- FieldAdapter
- 해당 클래스는 BaseModel의 각 Field인 pydantic.FieldInfo를 확장하였습니다.
- BaseModel이 아닌 type을 validation하기 위해서는 pydantic.TypeAdapter라는 특별한 클래스가 필요합니다. 그래서 클래스 이름이 FieldAdapter입니다.
- API 함수에서 정의할 때 request parameter type을 구분할 수 있도록 Header, Path, Query, Body, Form 하위클래스들이 존재합니다.
- 이제 parameter는 FieldAdapter의 하위 타입으로 정의됩니다.
API 예제
추가하려는 기능이 포함되었다고 가정하고 API를 만들어보겠습니다. path paramter, body parameter, query paramter, header parameter가 모두 요구되는 API를 만들 것입니다.
- path_paramter
- item_id
- query_paramter
- q (optional)
- body_paramter
- item
- user (optional)
- importance: int
- header_parameter
- x-token
from typing import Annotated, Optional
from flask import Flask, jsonify
from pydantic import BaseModel
from flask_parameter_validator import Header, Path, Query, parameter_validator
class Item(BaseModel):
name: str
description: str
price: int
class User(BaseModel):
name: str
address: str
app = Flask(__name__)
@app.post("/items/<item_id>")
@parameter_validator
def update_item(
item_id: Annotated[int, Path(gt=10)],
item: Item,
importance: int,
x_token: Annotated[str, Header()],
user: Optional[User] = None,
q: Annotated[Optional[str], Query()] = None,
):
results = {"item_id": item_id, "item": item.model_dump(), "importance": importance, "x_token": x_token}
if user:
results.update({"user": user.model_dump()})
if q:
results.update({"q": q})
return jsonify(results)Path, Query, Header, typing.Annotated라는 새로운 개념들이 등장해서 혼란스러울 수 있습니다. API 함수에서 데이터를 정의할 때 path, query, body, header를 구분해야 합니다. 그래야 ParameterValidator 데코레이터에서 어떤 parameter을 validation 할지 결정할 수 있습니다.
앞서 봤던 UML에서 FieldAdapter를 상속받은 Path, Query, Header, Body가 이를 해결할 수 있습니다. 함수 parameter의 type annodation을 정의할 때 Path, Query, Header, Body로 정의하면 ParameterValidator에서 어떤 parameter인지 알 수 있을 것입니다.
@app.post("/items/<item_id>")
def update_item(item_id: Path): ...하지만 이렇게 하면 데이터의 type annodation이 깨져버립니다. API함수 Body에서 item_id가 어떤 type인지 모릅니다.
이를 해결하기 위해 2가지 방법이 있습니다.
- 기본값으로 지정
@app.post("/items/<item_id>")
def update_item(item_id: int = Path()): ...ParameterValidator는 함수 signature 정보를 접근할 수 있습니다. 그래서 parameter의 기본값으로 지정된 Path를 보고 path parameter라는 것을 알 수 있습니다. 또, 데이터의 type annotation이 깨지지 않아 API함수 Body에서 item_id의 type 확인이 가능합니다. 만약 item_id의 기본값을 지정하고 싶다면 Path(default=5)같이 지정합니다.
- typing.Annotated
@app.post("/items/<item_id>")
def update_item(item_id: Annotated[int, Path()]): ...typing.Annodated는 type에 metadata를 추가하는 기능입니다. int라는 type에 메타데이터로 Path가 추가된 것입니다. type check시에는 Annotated의 첫 번째 인자(type)를 확인합니다. 자세한 사항은 typing document에 나와있습니다.
ParameterValidator에서 Annodated의 두 번째 인자 Path를 확인할 수 있어 path parameter라는 것을 알 수 있고, 데이터의 type annotation이 깨지지 않아 API함수 Body에서 type 확인이 가능합니다. 단점은 python 3.9 이상이어야 한다는 것입니다.
fastapi는 두 가지 방식 모두 지원하는데 여기서는 typing.Annotated 방식으로 만들어볼 것입니다.
Models
FieldAdapter와 이를 상속받은 클래스를 살펴봅니다. FieldAdapter는 각 parameter의 슈퍼클래스이며 pydantic.FieldInfo + pydantic.TypeAdapter의 기능으로 구성됩니다.
pydantic.FieldInfo
BaseModel의 각 Field를 나타내는 클래스로 pydantic에서는 BaseModel 외부에서 사용하지 않도록 권장하지만 field의 정보를 저장할 수 있고, gt, ge, lt, le, min_length, max_length 같은 추가적인 validation을 제공하여 사용합니다.
pydantic.TypeAdapter
일반적으로 pydantic validation는 BaseModel의 메소드이기 때문에 BaseModel을 상속받아 정의된 model type만 사용할 수 있습니다. Item, User가 이에 해당합니다. python typing의 type을 갖는 경우 pydantic validation을 사용할 때 TypeAdapter라는 클래스를 지원합니다. 일반 type을 TypeAdapter으로 wrapping 하고 TypeAdapter의 validate_python 메소드를 사용하여 데이터를 validation 합니다.
from pydantic import BaseModel, TypeAdapter
int_validator = TypeAdapter(int)
int_validator.validate_python(5)
# BaseModel도 사용 가능하다.
class User(BaseModel):
name: str
address: str
user_validator = TypeAdapter(User)
user_validator.validate_python({"name": "foo", "address": "bar"})FieldAdapter
pydantic.FieldInfo와 pydantic.TypeAdapter 기능을 결합한 베이스 클래스입니다.
class FieldAdapter:
def __init__(
self,
default: Any = PydanticUndefined,
*,
alias: Optional[str] = None,
title: Optional[str] = None,
description: Optional[str] = None,
gt: Optional[float] = None,
ge: Optional[float] = None,
lt: Optional[float] = None,
le: Optional[float] = None,
min_length: Optional[int] = None,
max_length: Optional[int] = None,
**extra: Any,
) -> None:
self._field_info = FieldInfo(
default=default,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
**extra,
)
self._type_adapter: TypeAdapter[Any] = self._get_type_adapter()
def _get_type_adapter(self) -> TypeAdapter[Any]:
return TypeAdapter(Annotated[self.field_info.annotation, self.field_info]) # type: ignore
def validate(self, obj: Any, loc: Tuple[str, ...]) -> Tuple[Any, List[Dict[str, Any]]]:
value, errors = None, []
try:
value = self._type_adapter.validate_python(obj)
except ValidationError as exc:
errors = self._regenerate_with_loc(exc.errors(), loc=loc)
return value, errors
# 중략...- _get_type_adapter: 클래스가 초기화될 때 TypeAdapter를 생성합니다. Annodated를 사용하여 self.field_info.annotation 타입에 self.field_info 메타데이터를 추가합니다.
- validate: 생성된 TypeAdapter로 데이터(obj)를 validation 합니다.
FieldAdapter를 상속받아 Param, Body, Query 등을 구현하면 됩니다.
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():
if get_origin(param.annotation) is Annotated: # 1
annotated_param = get_args(param.annotation) # 1
field = annotated_param[1]
if isinstance(field, Body) or isinstance(field, _params.Form): # 2
self._update_field_info(field, param_name, param)
dependant.body_params[param_name] = field
elif isinstance(field, Path):
self._update_field_info(field, param_name, param)
dependant.path_params[param_name] = field
elif isinstance(field, Query):
self._update_field_info(field, param_name, param)
dependant.query_params[param_name] = field
elif isinstance(field, Header):
self._update_field_info(field, param_name, param)
dependant.header_params[param_name] = field
else:
if param.annotation is inspect._empty:
continue
field = _params.Body(title=param_name, default=param.default, annotation=param.annotation) # 3
dependant.body_params[param_name] = field
return dependant- typing.get_origin으로 parameter의 type이 Annotated인지 체크하고 typing.get_args로 Annotated 인자(type, metadata)를 접근할 수 있습니다.
- isinstance로 param type 별로 구분하여 dependant에 저장합니다.
- parameter가 Annodated로 정의되지 않았다면 body로 취급합니다.
solve dependant
def _solve_dependencies(self) -> Tuple[Dict[str, BaseModel], List[Union[Dict[str, Any], ErrorDetails]]]:
solved_params: Dict[str, BaseModel] = {}
errors: List[Union[Dict[str, Any], ErrorDetails]] = []
headers: Dict[str, Any] = request.headers
_params, _errors = self.dependant.solve_header_params(headers)
errors.extend(_errors)
solved_params.update(_params)
path: Dict[str, Any] = request.view_args or {}
_params, _errors = self.dependant.solve_path_params(path)
errors.extend(_errors)
solved_params.update(_params)
query: Dict[str, str] = dict(request.args) or {}
_params, _errors = self.dependant.solve_query_params(query)
errors.extend(_errors)
solved_params.update(_params)
if self.dependant.body_params:
if self.dependant.is_form_type:
received_body = dict(request.form)
else:
received_body = request.json or {}
_params, _errors = self.dependant.solve_body(received_body)
errors.extend(_errors)
solved_params.update(_params)
return solved_params, errorsrequest parameter의 값을 validation 합니다. self.dependant의 solve_* 메소드가 이를 수행합니다.
dependant
solve_* 메소드중에 solve_body를 대표적으로 살펴봅니다.
def solve_body(self, received_body: Dict[str, Any]) -> Tuple[Dict[str, BaseModel], List[Union[Dict[str, Any], ErrorDetails]]]:
# 중략..
for param_name, param in self.body_params.items():
# 중략..
validated_param, _errors = param.validate(_received_body, loc=loc)
if _errors:
errors.extend(_errors)
if validated_param:
solved[param_name] = validated_param
return solved, errors저장하고 있던 parameter(FieldAdapter)를 하나씩 꺼내 validate 합니다.
이번 편에서 중요한 개념은 3가지입니다.
- pydantic.FieldInfo
- Field의 정보를 저장하고 ge, gt 등의 validation을 제공하는 pydantic 클래스
- pydantic.TypeAdapter
- type이 BaseModel이 아니어도 pydantic validation을 지원하는 pydantic 클래스
- Annotated
- type에 metadata를 더할 수 있는 python3.9부터 지원하는 typing 기능
이 3가지를 사용하면 API 함수 parameter에 request parameter type을 구분하여 정의할 수 있고 어떤 type이든 pydantic validation을 사용할 수 있습니다.
이제 ooknimm/flask-parameter-validator나 fastapi 코드를 참고하여 직접 validation을 만들어 보는 것을 추천드립니다.
'Python' 카테고리의 다른 글
| Flask Request Parameter 우아하게 검사하기 - 1 (0) | 2023.09.24 |
|---|
