还有一部分security安全相关将单独开一章，本章将FastAPI剩余值得读的部分过一遍
params.py
这里定义了例如Depends()，Body()等类，内容比较简单。在此将它们列出，具体用法可直接参考官方文档。
class ParamTypes(Enum)
class Param(FieldInfo)
class Path(Param)
class Query(Param)
class Header(Param)
class Cookie(Param)
class Body(FieldInfo)
class Form(Body)
class File(Form)
class Depends
class Security(Depends)
param_functions.py
上述类的工厂函数模式，返回实例。例如：
def Path(
default: Any,
*,
alias: Optional[str] = None,
title: Optional[str] = None,
description: Optional[str] = None,
gt: Optional[float] = None,
ge: Optional[float] = None,
lt: Optional[float] = None,
le: Optional[float] = None,
min_length: Optional[int] = None,
max_length: Optional[int] = None,
regex: Optional[str] = None,
deprecated: Optional[bool] = None,
**extra: Any,
) -> Any:
return params.Path(
default=default,
alias=alias,
title=title,
description=description,
gt=gt,
ge=ge,
lt=lt,
le=le,
min_length=min_length,
max_length=max_length,
regex=regex,
deprecated=deprecated,
**extra,
)
utils.py
工具类，其中大部分函数，都已经在前几篇中被提及
encoders.py
一个JSON解码的工具
concurrency.py
相比Starlette添加了一个线程池上下文管理器
@asynccontextmanager
async def contextmanager_in_threadpool(cm: Any) -> Any:
try:
yield await run_in_threadpool(cm.__enter__)
except Exception as e:
ok = await run_in_threadpool(cm.__exit__, type(e), e, None)
if not ok:
raise e
else:
await run_in_threadpool(cm.__exit__, None, None, None)
datastructures.py
对Starlette的UploadFile进行了封装
class UploadFile(StarletteUploadFile):
@classmethod
def __get_validators__(cls: Type["UploadFile"]) -> Iterable[Callable]:
yield cls.validate
@classmethod
def validate(cls: Type["UploadFile"], v: Any) -> Any:
if not isinstance(v, StarletteUploadFile):
raise ValueError(f"Expected UploadFile, received: {type(v)}")
return v
exception_handlers.py
加了两种异常处理
async def http_exception_handler(request: Request, exc: HTTPException):
headers = getattr(exc, "headers", None)
if headers:
return JSONResponse(
{"detail": exc.detail}, status_code=exc.status_code, headers=headers
)
else:
return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)
async def request_validation_exception_handler(
request: Request, exc: RequestValidationError
) -> JSONResponse:
return JSONResponse(
status_code=HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": jsonable_encoder(exc.errors())},
)
exceptions.py
新加了四种异常
class HTTPException(StarletteHTTPException):
def __init__(
self,
status_code: int,
detail: Any = None,
headers: Optional[Dict[str, Any]] = None,
) -> None:
super().__init__(status_code=status_code, detail=detail)
self.headers = headers
RequestErrorModel = create_model("Request")
WebSocketErrorModel = create_model("WebSocket")
class FastAPIError(RuntimeError):
"""
A generic, FastAPI-specific error.
"""
class RequestValidationError(ValidationError):
def __init__(self, errors: Sequence[ErrorList], *, body: Any = None) -> None:
self.body = body
super().__init__(errors, RequestErrorModel)
class WebSocketRequestValidationError(ValidationError):
def __init__(self, errors: Sequence[ErrorList]) -> None:
super().__init__(errors, WebSocketErrorModel)
responses.py
添加了ORJSONResponse，orjson是一种现在python性能最高的json库。有需求可以指定该种response。
class ORJSONResponse(JSONResponse):
media_type = "application/json"
def render(self, content: Any) -> bytes:
assert orjson is not None, "orjson must be installed to use ORJSONResponse"
return orjson.dumps(content)
网友评论