pydantic.error_wrappers.ValidationError: 1 validation error for '' with FastAPI
pydantic.error_wrappers.ValidationError: 1 validation error for '' with FastAPI
我的 FastAPI 应用程序中有这些模式:
class Run(BaseModel):
id: int = Field(..., description="Run id")
type: str = Field(..., description="Type")
class RunsResponse(BaseModel):
data: list[Run] = Field(..., description="List of existing runs")
links: dict = Field(..., description="link to runs")
然后我将这个装饰器用于所有端点:
def handle_request(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
response = func(*args, **kwargs)
except (requests.exceptions.SSLError, requests.exceptions.RequestException) as ex:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(ex))
return response
return wrapper
这条路线:
@router.get(
"/runs",
response_model=RunsResponse,
)
@handle_request
async def get_runs(request: Request) -> JSONResponse:
response = send_request(request, SOME_URL)
return JSONResponse(content=response)
因此,如果我在装饰器中没有这些代码的情况下尝试 api,一切正常。但是当我把代码放在装饰器中并在我的路由器功能上使用它时,我得到这个错误:
pydantic.error_wrappers.ValidationError: 1 validation error for RunsResponse
response
value is not a valid dict (type=type_error.dict)
我无法调试代码,因为我在 Swagger 中按下 Execute 按钮后立即发生错误。我怀疑它与 functools
中的 @wraps
有关,但我不知道如何解决它。
谁能帮我解决这个问题?
更新:
我正在添加 send_request
方法,以防万一:
def send_request(request: Request, url: str) -> dict:
headers = {
"Content-type": "application/json",
"Accept": "application/json",
}
response = requests.get(url, headers=headers)
data = response.json()
return data
我可以通过在返回之前使用 await
将装饰器中的函数解压缩到一个非协程对象中来解决这个问题:
def handle_request(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
response = await func(*args, **kwargs) # <- Added await here
except (requests.exceptions.SSLError, requests.exceptions.RequestException) as ex:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(ex))
return response
return wrapper
现在一切正常。
我的 FastAPI 应用程序中有这些模式:
class Run(BaseModel):
id: int = Field(..., description="Run id")
type: str = Field(..., description="Type")
class RunsResponse(BaseModel):
data: list[Run] = Field(..., description="List of existing runs")
links: dict = Field(..., description="link to runs")
然后我将这个装饰器用于所有端点:
def handle_request(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
response = func(*args, **kwargs)
except (requests.exceptions.SSLError, requests.exceptions.RequestException) as ex:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(ex))
return response
return wrapper
这条路线:
@router.get(
"/runs",
response_model=RunsResponse,
)
@handle_request
async def get_runs(request: Request) -> JSONResponse:
response = send_request(request, SOME_URL)
return JSONResponse(content=response)
因此,如果我在装饰器中没有这些代码的情况下尝试 api,一切正常。但是当我把代码放在装饰器中并在我的路由器功能上使用它时,我得到这个错误:
pydantic.error_wrappers.ValidationError: 1 validation error for RunsResponse
response
value is not a valid dict (type=type_error.dict)
我无法调试代码,因为我在 Swagger 中按下 Execute 按钮后立即发生错误。我怀疑它与 functools
中的 @wraps
有关,但我不知道如何解决它。
谁能帮我解决这个问题?
更新:
我正在添加 send_request
方法,以防万一:
def send_request(request: Request, url: str) -> dict:
headers = {
"Content-type": "application/json",
"Accept": "application/json",
}
response = requests.get(url, headers=headers)
data = response.json()
return data
我可以通过在返回之前使用 await
将装饰器中的函数解压缩到一个非协程对象中来解决这个问题:
def handle_request(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
response = await func(*args, **kwargs) # <- Added await here
except (requests.exceptions.SSLError, requests.exceptions.RequestException) as ex:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(ex))
return response
return wrapper
现在一切正常。