This section includes guides for advanced usage patterns.
To add your own custom location handler, write a function that receives a request, an argument name, and a Field
, then decorate that function with Parser.location_handler
.
from webargs import fields
from webargs.flaskparser import parser
@parser.location_handler("data")
def parse_data(request, name, field):
return request.data.get(name)
# Now 'data' can be specified as a location
@parser.use_args({"per_page": fields.Int()}, locations=("data",))
def posts(args):
return "displaying {} posts".format(args["per_page"])
When you need more flexibility in defining input schemas, you can pass a marshmallow Schema
instead of a dictionary to Parser.parse
, Parser.use_args
, and Parser.use_kwargs
.
from marshmallow import Schema, fields
from webargs.flaskparser import use_args
class UserSchema(Schema):
id = fields.Int(dump_only=True) # read-only (won't be parsed by webargs)
username = fields.Str(required=True)
password = fields.Str(load_only=True) # write-only
first_name = fields.Str(missing="")
last_name = fields.Str(missing="")
date_registered = fields.DateTime(dump_only=True)
# NOTE: Uncomment below two lines if you're using marshmallow 2
# class Meta:
# strict = True
@use_args(UserSchema())
def profile_view(args):
username = args["userame"]
# ...
@use_kwargs(UserSchema())
def profile_update(username, password, first_name, last_name):
update_profile(username, password, first_name, last_name)
# ...
# You can add additional parameters
@use_kwargs({"posts_per_page": fields.Int(missing=10, location="query")})
@use_args(UserSchema())
def profile_posts(args, posts_per_page):
username = args["username"]
# ...
Warning
If you’re using marshmallow 2, you should always set strict=True
(either as a class Meta
option or in the Schema’s constructor) when passing a schema to webargs. This will ensure that the parser’s error handler is invoked when expected.
Warning
Any Schema
passed to use_kwargs
MUST deserialize to a dictionary of data. Keep this in mind when writing post_load
methods.
If you need to parametrize a schema based on a given request, you can use a “Schema factory”: a callable that receives the current request
and returns a marshmallow.Schema
instance.
Consider the following use cases:
Filtering via a query parameter by passing only
to the Schema.
Handle partial updates for PATCH requests using marshmallow’s partial loading API.
from flask import Flask
from marshmallow import Schema, fields
from webargs.flaskparser import use_args
app = Flask(__name__)
class UserSchema(Schema):
id = fields.Int(dump_only=True)
username = fields.Str(required=True)
password = fields.Str(load_only=True)
first_name = fields.Str(missing="")
last_name = fields.Str(missing="")
date_registered = fields.DateTime(dump_only=True)
def make_user_schema(request):
# Filter based on 'fields' query parameter
fields = request.args.get("fields", None)
only = fields.split(",") if fields else None
# Respect partial updates for PATCH requests
partial = request.method == "PATCH"
# Add current request to the schema's context
return UserSchema(only=only, partial=partial, context={"request": request})
# Pass the factory to .parse, .use_args, or .use_kwargs
@app.route("/profile/", methods=["GET", "POST", "PATCH"])
@use_args(make_user_schema)
def profile_view(args):
username = args.get("username")
# ...
We can reduce boilerplate and improve [re]usability with a simple helper function:
from webargs.flaskparser import use_args
def use_args_with(schema_cls, schema_kwargs=None, **kwargs):
schema_kwargs = schema_kwargs or {}
def factory(request):
# Filter based on 'fields' query parameter
only = request.args.get("fields", None)
# Respect partial updates for PATCH requests
partial = request.method == "PATCH"
return schema_cls(
only=only, partial=partial, context={"request": request}, **schema_kwargs
)
return use_args(factory, **kwargs)
Now we can attach input schemas to our view functions like so:
@use_args_with(UserSchema)
def profile_view(args):
# ...
get_profile(**args)
See the “Custom Fields” section of the marshmallow docs for a detailed guide on defining custom fields which you can pass to webargs parsers: https://marshmallow.readthedocs.io/en/latest/custom_fields.html.
To add your own parser, extend Parser
and implement the parse_*
method(s) you need to override. For example, here is a custom Flask parser that handles nested query string arguments.
import re
from webargs import core
from webargs.flaskparser import FlaskParser
class NestedQueryFlaskParser(FlaskParser):
"""Parses nested query args
This parser handles nested query args. It expects nested levels
delimited by a period and then deserializes the query args into a
nested dict.
For example, the URL query params `?name.first=John&name.last=Boone`
will yield the following dict:
{
'name': {
'first': 'John',
'last': 'Boone',
}
}
"""
def parse_querystring(self, req, name, field):
return core.get_value(_structure_dict(req.args), name, field)
def _structure_dict(dict_):
def structure_dict_pair(r, key, value):
m = re.match(r"(\w+)\.(.*)", key)
if m:
if r.get(m.group(1)) is None:
r[m.group(1)] = {}
structure_dict_pair(r[m.group(1)], m.group(2), value)
else:
r[key] = value
r = {}
for k, v in dict_.items():
structure_dict_pair(r, k, v)
return r
If you’d prefer validation errors to return status code 400
instead
of 422
, you can override DEFAULT_VALIDATION_STATUS
on a Parser
.
from webargs.falconparser import FalconParser
class Parser(FalconParser):
DEFAULT_VALIDATION_STATUS = 400
parser = Parser()
use_args = parser.use_args
use_kwargs = parser.use_kwargs
In order to parse a JSON array of objects, pass many=True
to your input Schema
.
For example, you might implement JSON PATCH according to RFC 6902 like so:
from webargs import fields
from webargs.flaskparser import use_args
from marshmallow import Schema, validate
class PatchSchema(Schema):
op = fields.Str(
required=True,
validate=validate.OneOf(["add", "remove", "replace", "move", "copy"]),
)
path = fields.Str(required=True)
value = fields.Str(required=True)
@app.route("/profile/", methods=["patch"])
@use_args(PatchSchema(many=True), locations=("json",))
def patch_blog(args):
"""Implements JSON Patch for the user profile
Example JSON body:
[
{"op": "replace", "path": "/email", "value": "mynewemail@test.org"}
]
"""
# ...
Arguments for different locations can be specified by passing location
to each field individually:
@app.route("/stacked", methods=["POST"])
@use_args(
{
"page": fields.Int(location="query"),
"q": fields.Str(location="query"),
"name": fields.Str(location="json"),
}
)
def viewfunc(args):
page = args["page"]
# ...
Alternatively, you can pass multiple locations to use_args
:
@app.route("/stacked", methods=["POST"])
@use_args(
{"page": fields.Int(), "q": fields.Str(), "name": fields.Str()},
locations=("query", "json"),
)
def viewfunc(args):
page = args["page"]
# ...
However, this allows page
and q
to be passed in the request body and name
to be passed as a query parameter.
To restrict the arguments to single locations without having to pass location
to every field, you can call the use_args
multiple times:
query_args = {"page": fields.Int(), "q": fields.Int()}
json_args = {"name": fields.Str()}
@app.route("/stacked", methods=["POST"])
@use_args(query_args, locations=("query",))
@use_args(json_args, locations=("json",))
def viewfunc(query_parsed, json_parsed):
page = query_parsed["page"]
name = json_parsed["name"]
# ...
To reduce boilerplate, you could create shortcuts, like so:
import functools
query = functools.partial(use_args, locations=("query",))
body = functools.partial(use_args, locations=("json",))
@query(query_args)
@body(json_args)
def viewfunc(query_parsed, json_parsed):
page = query_parsed["page"]
name = json_parsed["name"]
# ...
See the Framework Support page for framework-specific guides.
For example applications, check out the examples directory.