커스텀 Request와 APIRoute 클래스

커스텀 Request와 APIRoute 클래스 (Custom Request and APIRoute class)

어떤 경우엔 Request와 APIRoute 클래스가 쓰는 로직을 오버라이드하고 싶을 수 있어요.

특히 미들웨어의 로직보다 이 방식이 더 나은 대안이 될 수 있어요. 예를 들어 요청 본문이 애플리케이션에 처리되기 전에 읽거나 조작하고 싶을 때죠.

위험 — 이건 "고급(advanced)" 기능이에요. FastAPI를 막 시작했다면 이 섹션은 건너뛰어도 좋아요.

출처: 공식문서

사용 사례

몇 가지 사용 사례는 이래요:

JSON이 아닌 요청 본문을 JSON으로 변환하기(예: msgpack).
gzip으로 압축된 요청 본문 압축 풀기.
모든 요청 본문을 자동으로 로깅하기.

커스텀 요청 본문 인코딩 다루기

커스텀 Request 서브클래스를 써서 gzip 요청을 압축 해제하는 방법을 볼게요.

그리고 그 커스텀 요청 클래스를 쓰는 APIRoute 서브클래스도 만들 거예요.

커스텀 `GzipRequest` 클래스 만들기

팁 — 이건 동작 방식을 보여주기 위한 장난감 예시예요. gzip 지원이 필요하다면 제공되는 GzipMiddleware를 쓰면 돼요.

먼저 GzipRequest 클래스를 만들어요. 이 클래스는 Request.body() 메서드를 오버라이드해서, 적절한 헤더가 있을 때 본문을 압축 해제해요.

헤더에 gzip이 없다면 본문을 압축 해제하지 않아요.

그렇게 하면 같은 라우트 클래스가 gzip 압축된 요청과 압축되지 않은 요청을 모두 처리할 수 있어요.

import gzip
from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute

class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body

class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler

app = FastAPI()
app.router.route_class = GzipRoute

@app.post("/sum")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

팁 — 가능하면 Annotated 버전을 쓰는 게 좋아요. (non-Annotated 버전도 있어요. numbers: list[int] = Body()처럼 쓰면 돼요.)

커스텀 `GzipRoute` 클래스 만들기

다음으로 GzipRequest를 쓰는 fastapi.routing.APIRoute의 커스텀 서브클래스를 만들어요.

이번에는 APIRoute.get_route_handler() 메서드를 오버라이드해요.

이 메서드는 함수를 반환해요. 그리고 그 함수가 요청을 받아 응답을 반환하는 역할을 해요.

여기서는 이걸 써서 원래 요청으로부터 GzipRequest를 만들어요.

import gzip
from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute

class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body

class GzipRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler

app = FastAPI()
app.router.route_class = GzipRoute

@app.post("/sum")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

기술적 세부사항 — Request에는 request.scope 속성이 있는데, 이건 요청과 관련된 메타데이터를 담은 Python dict예요. Request의 request.receive는 요청의 본문을 "받는(receive)" 함수예요. 이 scope dict와 receive 함수는 둘 다 ASGI 명세의 일부예요. 그리고 바로 이 두 가지, scope와 receive가 새 Request 인스턴스를 만들 때 필요한 것들이에요. Request에 대해 더 알고 싶다면 Starlette의 Requests 문서를 확인해 보세요.

GzipRequest.get_route_handler가 반환하는 함수가 다르게 하는 유일한 일은 Request를 GzipRequest로 변환하는 거예요.

이렇게 하면 우리 GzipRequest가 데이터를 경로 연산에 넘기기 전에 (필요하다면) 압축을 풀어줘요.

그 후의 모든 처리 로직은 같아요.

하지만 GzipRequest.body의 변경 덕분에, FastAPI가 필요할 때 요청 본문을 불러오면 자동으로 압축이 해제돼요.

예외 핸들러에서 요청 본문에 접근하기

팁 — 같은 문제를 해결하려면 RequestValidationError용 커스텀 핸들러에서 body를 쓰는 게 훨씬 쉬워요(에러 처리하기 참고). 하지만 이 예시도 여전히 유효하고, 내부 컴포넌트들과 어떻게 상호작용하는지 보여줘요.

이와 같은 접근 방식을 써서 예외 핸들러에서 요청 본문에 접근할 수도 있어요.

try/except 블록 안에서 요청을 처리하기만 하면 돼요:

from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute

class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler

app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute

@app.post("/")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return sum(numbers)

예외가 발생해도 Request 인스턴스는 여전히 스코프 안에 있으므로, 에러를 처리할 때 요청 본문을 읽어 쓸 수 있어요:

from collections.abc import Callable
from typing import Annotated

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute

class ValidationErrorLoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler

app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute

@app.post("/")
async def sum_numbers(numbers: Annotated[list[int], Body()]):
    return sum(numbers)

라우터 안에서의 커스텀 `APIRoute` 클래스

APIRouter의 route_class 파라미터를 설정할 수도 있어요:

import time
from collections.abc import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute

class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler

app = FastAPI()
router = APIRouter(route_class=TimedRoute)

@app.get("/")
async def not_timed():
    return {"message": "Not timed"}

@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}

app.include_router(router)

이 예시에서 router 아래의 경로 연산들은 커스텀 TimedRoute 클래스를 쓰고, 응답에 응답 생성에 걸린 시간이 담긴 X-Response-Time 헤더가 추가돼요:

import time
from collections.abc import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute

class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler

app = FastAPI()
router = APIRouter(route_class=TimedRoute)

@app.get("/")
async def not_timed():
    return {"message": "Not timed"}

@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}

app.include_router(router)

더 알아보기 (Learn more)

원문: Custom Request and APIRoute class - FastAPI

커스텀 Request와 APIRoute 클래스