커스텀 Request 및 APIRoute 클래스¶
일부 경우에는 Request와 APIRoute 클래스에서 사용되는 로직을 오버라이드하고 싶을 수 있습니다.
특히, 이는 middleware에 있는 로직의 좋은 대안이 될 수 있습니다.
예를 들어, 애플리케이션에서 처리되기 전에 요청 바디를 읽거나 조작하고 싶을 때가 그렇습니다.
위험
이 기능은 "고급" 기능입니다.
FastAPI를 이제 막 시작했다면 이 섹션은 건너뛰는 것이 좋습니다.
사용 사례¶
사용 사례에는 다음이 포함됩니다:
- JSON이 아닌 요청 바디를 JSON으로 변환하기(예:
msgpack). - gzip으로 압축된 요청 바디 압축 해제하기.
- 모든 요청 바디를 자동으로 로깅하기.
커스텀 요청 바디 인코딩 처리하기¶
커스텀 Request 서브클래스를 사용해 gzip 요청의 압축을 해제하는 방법을 살펴보겠습니다.
그리고 그 커스텀 요청 클래스를 사용하기 위한 APIRoute 서브클래스도 함께 보겠습니다.
커스텀 GzipRequest 클래스 만들기¶
팁
이 예시는 동작 방식 시연을 위한 장난감 예제입니다. Gzip 지원이 필요하다면 제공되는 GzipMiddleware를 사용할 수 있습니다.
먼저, GzipRequest 클래스를 만듭니다. 이 클래스는 Request.body() 메서드를 덮어써서, 적절한 헤더가 있는 경우 바디를 압축 해제합니다.
헤더에 gzip이 없으면 바디를 압축 해제하려고 시도하지 않습니다.
이렇게 하면 동일한 route 클래스가 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)}
🤓 Other versions and variants
import gzip
from typing import Annotated, Callable
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)}
Tip
Prefer to use the Annotated version if possible.
import gzip
from collections.abc import Callable
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: list[int] = Body()):
return {"sum": sum(numbers)}
Tip
Prefer to use the Annotated version if possible.
import gzip
from typing import Callable
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: list[int] = Body()):
return {"sum": sum(numbers)}
커스텀 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)}
🤓 Other versions and variants
import gzip
from typing import Annotated, Callable
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)}
Tip
Prefer to use the Annotated version if possible.
import gzip
from collections.abc import Callable
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: list[int] = Body()):
return {"sum": sum(numbers)}
Tip
Prefer to use the Annotated version if possible.
import gzip
from typing import Callable
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: 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)
🤓 Other versions and variants
from typing import Annotated, Callable
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)
Tip
Prefer to use the Annotated version if possible.
from collections.abc import Callable
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: list[int] = Body()):
return sum(numbers)
Tip
Prefer to use the Annotated version if possible.
from typing import Callable
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: 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)
🤓 Other versions and variants
from typing import Annotated, Callable
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)
Tip
Prefer to use the Annotated version if possible.
from collections.abc import Callable
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: list[int] = Body()):
return sum(numbers)
Tip
Prefer to use the Annotated version if possible.
from typing import Callable
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: 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)
🤓 Other versions and variants
import time
from typing 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)
🤓 Other versions and variants
import time
from typing 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)