FastAPI - Query Parameter에 Custom Validate 적용하기

문제상황

FastAPI에서 Query Parameter에 Validate를 적용하고 싶었다. 기본적인 건 min_length, max_length, pattern, regex 등은 기본적으로 지원해 주지만(공식문서), 조금 복잡한 validate를 적용할 때는 마땅한 게 없었다. 

View에서 물론 처리해도 되겠지만, 같은 코드 블록이 다른 View에서 계속 쓰이는 것은 보기 불편했다.(validate 함수를 지정하고 view에서 계속 호출하는 그게 너무 싫었다)

해결 과정

1. 처음 생각한 건 BaseModel을 적용해서 @field_validator 를 적용할까 생각했지만, 한 두 개의 query param을 위해 BaseModel을 적용하는 건 필요 이상으로 복잡하게 모델을 구현한다는 생각이 들었다.
2. 생각해보다가 FastAPI에서 제공해 주는 Depends를 활용하면 어떨까란 생각이 들었다. 마침 Depends에 대한 공식문서의 설명에서 내가 생각한 게 아주 틀리진 않다는 확신을 가졌다.

"Dependency Injection" means, in programming, that there is a way for your code (in this case, your path operation functions) to declare things that it requires to work and use: "dependencies".
And then, that system (in this case FastAPI) will take care of doing whatever is needed to provide your code with those needed dependencies ("inject" the dependencies).
This is very useful when you need to:
- Have shared logic (the same code logic again and again).
- Share database connections.
- Enforce security, authentication, role requirements, etc.
- And many other things..

첫 번째 줄의 Have shared logic 이란 부분에서 같은 validate를 적용하면 되지 않을까 싶었다.

 

구현해 보기

우선 비교를 위해 처음 구현되었던 부분을 보자

@app.get("/duplicate/validate/query1")
async def duple_validate_query_1(
    q: Annotated[str, Query(description="이것은 query string 이다.")]
):
    if not do_validate_something(q):
        return {"message": "Cheer Up!!!!"}
    ...



@app.get("/duplicate/validate/query2")
async def duple_validate_query_1(
    q: Annotated[str, Query(description="이것은 query string 이다.")]
):
    if not do_validate_something(q):
        return {"message": "Cheer Up!!"}
    ...

물론 '이게 뭐가 거슬리냐?'라고 말할 수 있다. 하지만 이러한 validate 가 조금씩 늘어난다면 어떻게 될까?

@app.get("/duplicate/validate/query1")
async def duple_validate_query_1(
    q: Annotated[str, Query(description="이것은 query string 이다.")],
    qq: Annotated[str, Query(description="이것은 query string 이다.")],
):
    if not do_validate_something(q):
        return {"message": "Cheer Up!!!!"}
    if not do_validate_something(qq):
        return {"message": "More Cheer Up!!!"}
    ...


@app.get("/duplicate/validate/query2")
async def duple_validate_query_1(
    q: Annotated[str, Query(description="이것은 query string 이다.")],
    qq: Annotated[str, Query(description="이것은 query string 이다.")],
):
    if not do_validate_something(q):
        return {"message": "Cheer Up!!!!"}
    if not do_validate_something(qq):
        return {"message": "More Cheer Up!!!"}
    ...

조금씩 코드뭉치가 늘어나는 느낌이다. 

query param이 많다면 다음과 같은 방법도 괜찮을 것이다.(참고자료)

class QueryParam(BaseModel):
    q: str = Query(description="이것은 query string 이다.")
    qq: str = Query(description="이것은 query string 이다.")
    qqq: int = Query(description="이것은 query string 이다.")

    @field_validator("q", "qq")
    @classmethod
    def q_must_is_goal(cls, v: str):
        if do_validate_something(v):
            return v
        raise HTTPException(
            status_code=HTTP_400_BAD_REQUEST, detail={"message": "Cheer Up"}
        )

    @field_validator("qqq")
    @classmethod
    def qqq_must_match(cls, v: int):
        expected_value = [10000, 10200, 30000]
        if v in expected_value:
            return v
        raise HTTPException(
            status_code=HTTP_400_BAD_REQUEST,
            detail={"message": f"qqq is {','.join(map(str, expected_value))}"},
        )


@app.get("/duplicate/validate/query2")
async def duple_validate_query_base_model(query_param: QueryParam = Depends()):
    return query_param.model_dump()

다만 이 방법은 너무 적은 수의 Query를 class로 적용하면, 복잡도를 높인다고 생각든다.

 

Depends 사용해 보기

그럼 이제 Depends를 사용해 보자.

async def depends_query_validate(
    q: Annotated[str, Query(description="이것은 Depends 로 검사하는 query string")]
):
    if do_validate_something(q):
        return q
    return "sorry invalid q"


@app.get("/duplicate/validate/query/depends")
async def duple_validate_query_depends(
    q: Annotated[str, Depends(depends_query_validate)]
):
    result = q
    return {"message": result}

성공!!!

실패

마치며

사실 Validator를 사용하는 방법이 있긴 한데 어째서인지 router로 serving하면 validate가 세 번이 돌고 있어서 사용하지 않았다. 이 글을 쓰면서 계속 이유를 찾고 있어서 만약 해결이 된다면 따로 포스팅할 예정이다.(app으로 serving 한 경우에는 이런 문제가 발생하지 않는다.)

def result_validate(value):
    return value + "1"
    
    
@test_router.get(
    "/TEST",
    name="test",
)
async def test(
    foo: Annotated[
        str | None,
        Query(description="foo"),
        AfterValidator(result_validate),
    ],
):
    return {"serving_router": foo}
    
    
@app.get(
    "/TEST",
    name="test",
)
async def test(
    foo: Annotated[
        str | None,
        Query(description="foo"),
        AfterValidator(result_validate),
    ],
):
    return {"serving_app": foo}

router로 서빙한 결과

app 으로 서빙한 결과