使用webargs对Starlette进行声明性请求解析和验证
项目描述
webargs-starlette是一个库，用于使用Starlette进行声明性请求解析和验证，建立在webargs之上。
它包含了webargs的所有优点，并为类型注解添加了一些额外的功能。
import uvicorn
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs_starlette import use_annotations
app = Starlette()
@app.route("/")
@use_annotations(location="query")
async def index(request, name: str = "World"):
return JSONResponse({"Hello": name})
if __name__ == "__main__":
uvicorn.run(app, port=5000)
# curl 'http://localhost:5000/'
# {"Hello": "World"}
# curl 'http://localhost:5000/?name=Ada'
# {"Hello": "Ada"}安装
pip install -U webargs-starlette
用法
解析器用法
使用 parser.parse 解析带有字段字典的Starlette Request。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs import fields
from webargs_starlette import parser
app = Starlette()
@app.route("/")
async def homepage(request):
args = {
"name": fields.Str(required=True),
"greeting": fields.Str(load_default="hello"),
}
parsed = await parser.parse(args, request)
greeting = parsed["greeting"]
name = parsed["name"]
return JSONResponse({"message": f"{greeting} {name}"})装饰器
使用 use_args 装饰器将解析后的参数字典注入到处理器函数中。以下代码片段与第一个示例等价。
重要：装饰的函数必须是协程函数。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs import fields
from webargs_starlette import use_args
app = Starlette()
@app.route("/")
@use_args(
{"name": fields.Str(required=True), "greeting": fields.Str(load_default="hello")}
)
async def homepage(request, args):
greeting = args["greeting"]
name = args["name"]
return JSONResponse({"message": f"{greeting} {name}"})使用 use_kwargs 装饰器将解析后的参数作为关键字参数注入。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs import fields
from webargs_starlette import use_args
app = Starlette()
@app.route("/")
@use_kwargs(
{"name": fields.Str(required=True), "greeting": fields.Str(load_default="hello")}
)
async def homepage(request, name, greeting):
return JSONResponse({"message": f"{greeting} {name}"})有关 use_args 和 use_kwargs 的更完整示例，请参阅 decorator_example.py。
错误处理
当验证失败时，解析器将引发一个 WebargsHTTPException 异常，它与 Starlette 的 HTTPException 相同，但增加了 messages（验证消息）、headers、exception（底层异常）和 schema（marshmallow Schema）属性。
您可以使用自定义异常处理器以 JSON 格式返回错误消息。
from starlette.responses import JSONResponse
from webargs_starlette import WebargsHTTPException
@app.exception_handler(WebargsHTTPException)
async def http_exception(request, exc):
return JSONResponse(exc.messages, status_code=exc.status_code, headers=exc.headers)注解
use_annotations 装饰器允许您使用类型注解解析请求对象。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs_starlette import use_annotations
app = Starlette()
@app.route("/")
@use_annotations(location="query")
async def welcome(request, name: str = "Friend"):
return JSONResponse({"message": f"Welcome, {name}!"})
# curl 'http://localhost:5000/'.
# {"message":"Welcome, Friend!"}
# curl 'http://localhost:5000/?name=Ada'.
# {"message":"Welcome, Ada!"}任何没有默认值的注解参数都是必需的。例如，如果我们在上面的示例中移除 name 的默认值，如果未传递 ?name，则返回 422 错误响应。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs_starlette import use_annotations, WebargsHTTPException
app = Starlette()
@app.route("/")
@use_annotations(location="query")
async def welcome(request, name: str):
return JSONResponse({"message": f"Welcome, {name}!"})
@app.exception_handler(WebargsHTTPException)
async def http_exception(request, exc):
return JSONResponse(exc.messages, status_code=exc.status_code, headers=exc.headers)
# curl "http://localhost:5000/"
# {"name":["Missing data for required field."]}当需要更多控制时，也可以使用 Field 实例注解参数。例如，您可能想添加一个验证器。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from webargs import fields
from marshmallow import validate
from webargs_starlette import use_annotations, WebargsHTTPException
app = Starlette()
@app.route("/")
@use_annotations(location="query")
async def welcome(request, name: fields.Str(validate=validate.Length(min=2))):
return JSONResponse({"message": f"Welcome, {name}!"})
@app.exception_handler(WebargsHTTPException)
async def http_exception(request, exc):
return JSONResponse(exc.messages, status_code=exc.status_code, headers=exc.headers)
# curl "http://localhost:5000/?name=A"
# {"name":["Shorter than minimum length 2."]}HTTPEndpoint 方法也可以用 use_annotations 装饰。
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.endpoints import HTTPEndpoint
from webargs_starlette import use_annotations
app = Starlette()
@app.route("/")
class WelcomeEndpoint(HTTPEndpoint):
@use_annotations(location="query")
async def get(self, request, name: str = "World"):
return JSONResponse({"message": f"Welcome, {name}!"})有关 use_annotations 的更完整示例，请参阅 annotation_example.py。
更多信息
有关如何使用 webargs 的更多信息，请参阅 webargs 文档。
许可证
MIT 许可。有关更多详细信息，请参阅 LICENSE 文件。
