简而言之:一个符合 mypy 标准的可继承版本,能够生成正确的 OpenAPI 模式字段类型,而不是任何/未知类型。
现有的解决方案将 FastAPI 参数设置为 typing.Any
,以防止验证两次失败,这会导致生成的 API 规范具有这些表单字段的任何/未知参数类型。
此解决方案在模式生成之前暂时注入正确的注释,并在其他解决方案中重置它们。
# Example usage
class ExampleForm(FormBaseModel):
name: str
age: int
@api.post("/test")
async def endpoint(form: ExampleForm = Depends(ExampleForm.as_form)):
return form.dict()
form_utils.py
import inspect
from pydantic import BaseModel, ValidationError
from fastapi import Form
from fastapi.exceptions import RequestValidationError
class FormBaseModel(BaseModel):
def __init_subclass__(cls, *args, **kwargs):
field_default = Form(...)
new_params = []
schema_params = []
for field in cls.__fields__.values():
new_params.append(
inspect.Parameter(
field.alias,
inspect.Parameter.POSITIONAL_ONLY,
default=Form(field.default) if not field.required else field_default,
annotation=inspect.Parameter.empty,
)
)
schema_params.append(
inspect.Parameter(
field.alias,
inspect.Parameter.POSITIONAL_ONLY,
default=Form(field.default) if not field.required else field_default,
annotation=field.annotation,
)
)
async def _as_form(**data):
try:
return cls(**data)
except ValidationError as e:
raise RequestValidationError(e.raw_errors)
async def _schema_mocked_call(**data):
"""
A fake version which is given the actual annotations, rather than typing.Any,
this version is used to generate the API schema, then the routes revert back to the original afterwards.
"""
pass
_as_form.__signature__ = inspect.signature(_as_form).replace(parameters=new_params) # type: ignore
setattr(cls, "as_form", _as_form)
_schema_mocked_call.__signature__ = inspect.signature(_schema_mocked_call).replace(parameters=schema_params) # type: ignore
# Set the schema patch func as an attr on the _as_form func so it can be accessed later from the route itself:
setattr(_as_form, "_schema_mocked_call", _schema_mocked_call)
@staticmethod
def as_form(parameters=[]) -> "FormBaseModel":
raise NotImplementedError
# asgi.py
from fastapi.routing import APIRoute
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from fastapi.dependencies.utils import get_dependant, get_body_field
api = FastAPI()
def custom_openapi():
if api.openapi_schema:
return api.openapi_schema
def create_reset_callback(route, deps, body_field):
def reset_callback():
route.dependant.dependencies = deps
route.body_field = body_field
return reset_callback
# The functions to call after schema generation to reset the routes to their original state:
reset_callbacks = []
for route in api.routes:
if isinstance(route, APIRoute):
orig_dependencies = list(route.dependant.dependencies)
orig_body_field = route.body_field
is_modified = False
for dep_index, dependency in enumerate(route.dependant.dependencies):
# If it's a form dependency, set the annotations to their true values:
if dependency.call.__name__ == "_as_form": # type: ignore
is_modified = True
route.dependant.dependencies[dep_index] = get_dependant(
path=dependency.path if dependency.path else route.path,
# This mocked func was set as an attribute on the original, correct function,
# replace it here temporarily:
call=dependency.call._schema_mocked_call, # type: ignore
name=dependency.name,
security_scopes=dependency.security_scopes,
use_cache=False, # Overriding, so don't want cached actual version.
)
if is_modified:
route.body_field = get_body_field(dependant=route.dependant, name=route.unique_id)
reset_callbacks.append(
create_reset_callback(route, orig_dependencies, orig_body_field)
)
openapi_schema = get_openapi(
title="foo",
version="bar",
routes=api.routes,
)
for callback in reset_callbacks:
callback()
api.openapi_schema = openapi_schema
return api.openapi_schema
api.openapi = custom_openapi # type: ignore[assignment]
form_data
。但是如果不看你提交了什么,就无法提供更多帮助。 - SColvinno=1&nm=abcd
- shanmuga