FastAPI는 어떻게 타입을 검증할까?

배경

FastAPI는 비동기를 간편하게 구현할수 있고, OpenAPI 문서를 자동으로 만들어주는 파이썬의 웹 프레임워크 중 하나이다. FastAPI는 Starlette라는 웹 프레임워크 기반으로 Routing 시스템을 구축했고, 그 위에 OpenAPI 문서를 자동으로 만들어주거나 Request body의 데이터 유효성을 자동으로 검사하는 기능을 추가했다. 이번 글에서는 FastAPI에서 어떻게 데이터 유효성을 검사하는지 공부한 내용을 정리해보았다. 참고한 라이브러리 버전은 다음과 같다.

• FastAPI: 0.85.2
• Pydantic: 1.10.4

유효성 검사 방법

FastAPI는 request를 처리하기전 get_request_handler 매서드에서 유효성 검사를 진행한다.

유효성 검사는 크게 2 단계로 진행된다.

1. Parameter 관련 정보 추출
2. 데이터 유효성 검사

Parameter 정보 추출

Parameter 정보 추출은 get_dependant 함수에 진행한다. 이때 진행되는 작업은 다음과 같다.

• Parameter 구분: Path parameter, Query parameter를 구분하고, 의존성 주입이 들어갔는지 구분한다.
• Parameter 정보 추출: 각 parameter의 이름, 타입, 기본값에 대한 정보를 추출한다.

위 두 단계에서는 python inspect 모듈을 사용하여 parameter에 대한 정보를 추출한다. Inspect은 파이썬 객체에 대한 정보를 검색해주는 모듈로 이를 활용하면 객체의 이름, 인자, docstring, 소스 코드등을 가져올 수 있다.
fastapi에서는 이를 활용해 다음과 같이 parameter 정보를 추출한다.

# https://github.com/tiangolo/fastapi/blob/0.85.2/fastapi/dependencies/utils.py#L272
def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
    signature = inspect.signature(call)
    globalns = getattr(call, "__globals__", {})
    typed_params = [
        inspect.Parameter(
            name=param.name,
            kind=param.kind,
            default=param.default,
            annotation=get_typed_annotation(param, globalns),
        )
        for param in signature.parameters.values()
    ]
    typed_signature = inspect.Signature(typed_params)
    return typed_signature
 
def make_dict(value: int, key: str = "default_key") -> dict[str, int]:
    return {key: value}
 
get_typed_signature(make_dict)
""" Result
(value: int, key: str = 'default_key')
"""

이렇게 추출한 코드에서 parameter 종류별로 ModelField를 만들어준다.
Model Field는 주어진 입력이 ModelField에 사전 정의된 타입과 일치하는지 검사하는 클래스로 pydantic에서 정의되어 있다.
이는 이후 데이터 유효성 검사시에 사용된다.

간단한 함수에서 get_dependant를 통해 parameter를 추출하면 다음과 같다.
각 parameter의 이름, 타입, default 값 유무(required), default 값 등이 저장됨을 확인할 수 있다.

def make_dict(value: int, key: str = "default_key") -> dict[str, int]:
    return {key: value}
 
get_dependant(make_dict).query_params
""" Result
[
    ModelField(name='value', type=int, required=True),
    ModelField(name='key', type=str, required=False, default='default_key')
]
"""

데이터 유효성 검사

데이터 유효성 검사는 request_params_to_args에서 진행한다.

get_dependant에서 생성한 ModelField의 validate 매서드를 사용해 데이터 유효성을 검증함과 동시에 type casting을 해준다.
만약 유효하지 않는 데이터라면 데이터의 parameter 종류, 에러 종류 등을 저장하여 이후 에러메시지로 출력해준다.

def make_dict(value: int, key: str = "default_key") -> dict[str, int]:
    return {key: value}
 
make_dict_param = get_dependant(make_dict_param).query_params
valid_param = {"key": "valid", "value": 0}
request_params_to_args(request_params=make_dict_param, received_params=valid_param)
""" Result
(
	{'value': 0, 'key': 'valid'},
    []
)
"""
 
invalid_param = {"key": "invalid"}
request_params_to_args(request_params=make_dict_param, received_params=invalid_param)
""" Result
(
	{'key': 'invalid'},
    [ErrorWrapper(exc=NoneIsNotAllowedError(), loc=('query', 'value'))]
)
"""

요약

요약하면 다음과 같다.

• get_dependant 매서드에서 python inspect를 활용하여 타입 정보를 추출하여 pydantic class를 만든다
• request_params_to_args 매서드에서 pydantic class를 활용하여 데이터 유효성 검사를 진행한다

활용

 python inspect 결과에 따라 데이터 유효성 검사를 한다는 정보를 활용해 서비스 로직이 구현되었을 때 바로 이를 반환하는 간단한 API를 를 만들어주는 함수를 구현해보았다 (github repo).

원하는 기능은 다음과 같다.

아래와 같은 Service method 코드가 있을때 해당 service 코드를 바로 반환하는 controller 코드를 작성하는 것이다.

# Service Code
class SimpleService:
    def __init__(self):
        self.service_name = "Simple Service"
 
    def get_name(self) -> str:
        return self.service_name
 
    def get_dict(self, key: str, value: int) -> dict:
        return {key: value}
        
# Controller code
router = APIRouter()
@route.get("/get_dict")
async def get_dict( 
    key: str,
    value: int,
    service: SimpleService = Depends()
 ) -> str:
	return service.get_dict(key=key, value=value)

Parameter 추출

이를 위해 첫번째로 parameter 추출코드를 작성했다.

Fastapi의 코드와 다른 점은 다음 2가지이다.

1. self parameter를 제외한다.
2. list 타입일 경우 Query([])를 default 값으로 설정한다.

def get_typed_params(call: Callable[..., Any]) -> list[inspect.Parameter]:
    """Get typed parameters from a function except self."""
    signature = inspect.signature(call)
 
    typed_params = []
    for param in signature.parameters.values():
        if param.name == "self":
            continue
 
        if str(param.annotation).startswith("list"):
            param = param.replace(default=Query([]))
 
        typed_params.append(param)
    return typed_params

API 생성

그 다음은 Controller 코드를 만들어주는 부분을 작성했다.

1. 타입 추출: 위에서 구현한 ```get_typed_params``` 함수를 사용해 타입을 추출했다. 그리고 추가적으로 매서드가 존재하는 클래스 의존성을 삽입하기 위한 parameter를 추가하고, return type을 추출했다.
2. API 로직: 단순히 서비스 매서드를 바로 반환하는 함수 (simple_api)를 작성했다.
3. simple_api의 inspect 결과를 1번에서 추출한 타입과 method 이름으로 변경했다.

def make_simple_api( 
    router: APIRouter,
    http_method: str,
    url: str,
    service_klass: object,
    method_name: str,
 ) -> None:
    """Make just-call-service api from service method."""
    SERVICE_ARGUMENT_NAME = "service"
 
    method = getattr(service_klass, method_name)
 
    params = get_typed_params(method)
    params.append(
        inspect.Parameter(
            name=SERVICE_ARGUMENT_NAME,
            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
            default=Depends(),
            annotation=service_klass,
        ),
    )
    params = inspect.Signature(params)
    return_type = inspect.signature(method).return_annotation
 
    async def simple_api(*args, **kwargs):
        bound_arguments = params.bind(*args, **kwargs)
        bound_arguments.apply_defaults()
        method = getattr(kwargs[SERVICE_ARGUMENT_NAME], method_name)
        kwargs.pop(SERVICE_ARGUMENT_NAME)
        return method(*args, **kwargs)
 
    simple_api.__signature__ = params
    simple_api.__annotations__ = {"return": return_type}
    simple_api.__name__ = method_name
 
    router_decorator = getattr(router, http_method)
    router_decorator(url, response_model=return_type)(simple_api)

사용법

위 함수를 사용하여 api를 다음과 같이 추가하면 된다.

router = APIRouter(prefix="/simple")
 
 
make_simple_api(
    router=router,
    http_method="get",
    url="/get_name",
    service_klass=SimpleService,
    method_name="get_name",
)
 
 
make_simple_api(
    router=router,
    http_method="get",
    url="/get_dict",
    service_klass=SimpleService,
    method_name="get_dict",
)

결과

결과는 다음과 같이 잘 동작함을 확인했다.

기타

코드를 구현할 때 2가지 선택이 있었다.

1. 함수 vs Decorator
   처음에는 method에 decorator로 만드는 코드를 작성하고 싶었지만, 그러면 circular import error가 발생해 일반적인 함수로 만들었다.
2. Argument로 class object vs class name
   class object를 직접 주는 것이 아니라 class name을 입력으로 주고 싶었지만, 그러면 폴더 구조와, 파일 이름에 따라 코드가 바뀌어야 하기 때문에 class object를 직접 주는 것으로 코드를 변경했다.