FastAPI를 사용하여 요청에 헤더가 없을 때 사용자 지정 응답을 반환하는 방법은 무엇입니까?

FastAPI를 사용하여 요청에 헤더가 없을 때 사용자 지정 응답을 반환하는 방법은 무엇입니까?

Fast에서 끝점을 만들고 싶습니다특정한 것을 요구하는 API는 가 없을 때 사용자 정의 코드를 생성하고, 를 Open에 표시합니다FastAPI에 의해 생성된 API 문서입니다.

예를 들어, 이 끝점을 다음과 같이 요구하는 경우:

@app.post("/")
async def fn(some_custom_header: str = Header(...)):
    pass

클라이언트 요청에 오류 코드가 없을 경우 서버는 오류 코드를 사용하여 를 생성합니다. 하지만 저는 그것을 로 바꿀 수 있으면 좋겠습니다. 다시 말해, 나는 나의 API를 사용하고 싶다.

나는 가능한 해결책이 기능 본체에서 를 사용하고 테스트를 하는 것이라고 생각했지만, 불행히도, 이것은 오픈으로 귀결된다헤더가 .임을 나타내는 API 문서입니다.

---

옵션 1

에서와 같이 상영해도 괜찮으시다면 다음과 같이 쉽게 상영할 수 있습니다:

from fastapi import Header, HTTPException
@app.post("/")
def some_route(some_custom_header: Optional[str] = Header(None)):
    if not some_custom_header:
        raise HTTPException(status_code=401, detail="Unauthorized")
    return {"some-custom-header": some_custom_header}

옵션 2

그러나 OpenAPI에서 처럼 보이기를 원하기 때문에 기본 예외 처리기를 재정의해야 합니다. 따라서 . .을(를) 재정의해야 하며, 는 Pydantic의 하위 클래스이므로 위 링크와 같이 오류에 액세스할 수 있습니다, 사용자 정의가 오류에 포함되어 있는지(요청에서 누락되었거나 유형이 아님을 의미) 확인하여 사용자 정의 응답을 반환할 수 있습니다.

예

from fastapi import FastAPI, Request, Header, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

routes_with_custom_header = ["/"]

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    if request.url.path in routes_with_custom_header:
        for err in exc.errors():
            if err['loc'][0] == "header" and err['loc'][1] == 'some-custom-header':
                return JSONResponse(content={"401": "Unauthorized"}, status_code=401)
            
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

@app.get("/")
def some_route(some_custom_header: str = Header(...)):
    return {"some-custom-header": some_custom_header}

옵션 3

또 다른 해결책은 (토론에서 영감을 받아) 을 사용하는 것입니다. 하위 앱(또는 필요한 경우 그 이상)을 생성하고 이를 기본 앱에 마운트할 수 있습니다(사용자 지정이 필요한 경로 포함). 따라서 해당 하위 앱의 for를 재정의하는 것은 이전 솔루션에서 설명한 대로 를 확인할 필요 없이 해당 경로에만 적용되고 기본 앱은 평소와 같이 나머지 경로를 사용합니다. 에 따라:

빠른 속도로 장착API 응용 프로그램

"장착"은 특정 경로에 완전히 "독립적인" 응용 프로그램을 추가하는 것을 의미하며, 이는 해당 하위 응용 프로그램에서 선언된 것과 함께 해당 경로 아래의 모든 것을 처리한다.

예

경로를 사용하여 하위 응용 프로그램(예: 아래 예제)을 마운트한 경우 해당 페이지의 API 문서에는 메인 앱의 경로만 포함되므로 의 경로에 액세스할 수 없습니다. 또한 메인 API의 경로(메인 API에 이러한 경로가 존재하는 경우)를 방해할 수 있으며, 이후 에 요청을 발행하면 실제로 메인 API의 해당 경로를 호출할 수 있습니다(아래 설명 참조). 따라서 아래 그림과 같이 다른 경로(예: )를 사용하여 마운트하고 에서 하위 API 문서에 액세스하는 것이 좋습니다. Python 요청 예제도 아래에 제공되어 앱을 테스트하는 방법을 보여줍니다.

from fastapi import FastAPI, Request, Header
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get('/')
async def main():
    return {'message': 'Hello from main API'}
    

subapi = FastAPI()
   
@subapi.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(content={'401': 'Unauthorized'}, status_code=401)

    
@subapi.get('/')
async def sub_api_route(some_custom_header: str = Header(...)):
    return {'some-custom-header': some_custom_header}    


app.mount('/sub', subapi)

위의 예를 테스트합니다

import requests

# Test main API
url = 'http://127.0.0.1:8000/'

r = requests.get(url=url)
print(r.status_code, r.json())

# Test sub API
url = 'http://127.0.0.1:8000/sub/'

r = requests.get(url=url)
print(r.status_code, r.json())

headers = {'some-custom-header': 'this is some custom header'}
r = requests.get(url=url, headers=headers)
print(r.status_code, r.json())

옵션 4

추가 솔루션은 의 옵션 2에서 설명한 대로 와 함께 를 사용하고 에 설명된 대로 블록 내(예외를 포착하는 데 사용됨) 요청을 처리하는 것입니다. 예외가 발생하면 원하는 대로 오류를 처리하고 사용자 지정 응답을 반환할 수 있습니다.

예

from fastapi import FastAPI, APIRouter, Response, Request, Header, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from typing import Callable


class ValidationErrorHandlingRoute(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 e:
                raise HTTPException(status_code=401, detail={'401': 'Unauthorized'})
                            
        return custom_route_handler


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


@app.get('/')
async def main():
    return {'message': 'Hello from main API'}
    

@router.get('/custom')
async def custom_route(some_custom_header: str = Header(...)):
    return {'some-custom-header': some_custom_header}


app.include_router(router) 

위의 예를 테스트합니다

import requests

# Test main API
url = 'http://127.0.0.1:8000/'

r = requests.get(url=url)
print(r.status_code, r.json())

# Test custom route
url = 'http://127.0.0.1:8000/custom'

r = requests.get(url=url)
print(r.status_code, r.json())

headers = {'some-custom-header': 'this is some custom header'}
r = requests.get(url=url, headers=headers)
print(r.status_code, r.json())