from webargs.core import json
import falcon
import marshmallow as ma
from webargs import fields
from webargs.falconparser import parser, use_args, use_kwargs
from webargs.core import MARSHMALLOW_VERSION_INFO
hello_args = {"name": fields.Str(missing="World", validate=lambda n: len(n) >= 3)}
hello_multiple = {"name": fields.List(fields.Str())}
class HelloSchema(ma.Schema):
name = fields.Str(missing="World", validate=lambda n: len(n) >= 3)
strict_kwargs = {"strict": True} if MARSHMALLOW_VERSION_INFO[0] < 3 else {}
hello_many_schema = HelloSchema(many=True, **strict_kwargs)
class Echo(object):
def on_get(self, req, resp):
try:
parsed = parser.parse(hello_args, req)
except json.JSONDecodeError:
resp.body = json.dumps(["Invalid JSON."])
resp.status = falcon.HTTP_400
else:
resp.body = json.dumps(parsed)
on_post = on_get
class EchoQuery(object):
def on_get(self, req, resp):
parsed = parser.parse(hello_args, req, locations=("query",))
resp.body = json.dumps(parsed)
class EchoJSON(object):
def on_post(self, req, resp):
parsed = parser.parse(hello_args, req, locations=("json",))
resp.body = json.dumps(parsed)
class EchoForm(object):
def on_post(self, req, resp):
parsed = parser.parse(hello_args, req, locations=("form",))
resp.body = json.dumps(parsed)
class EchoUseArgs(object):
@use_args(hello_args)
def on_get(self, req, resp, args):
resp.body = json.dumps(args)
on_post = on_get
class EchoUseKwargs(object):
@use_kwargs(hello_args)
def on_get(self, req, resp, name):
resp.body = json.dumps({"name": name})
on_post = on_get
class EchoUseArgsValidated(object):
@use_args({"value": fields.Int()}, validate=lambda args: args["value"] > 42)
def on_get(self, req, resp, args):
resp.body = json.dumps(args)
on_post = on_get
class EchoMulti(object):
def on_get(self, req, resp):
resp.body = json.dumps(parser.parse(hello_multiple, req))
on_post = on_get
class EchoManySchema(object):
def on_get(self, req, resp):
resp.body = json.dumps(
parser.parse(hello_many_schema, req, locations=("json",))
)
on_post = on_get
class EchoUseArgsWithPathParam(object):
@use_args({"value": fields.Int()})
def on_get(self, req, resp, args, name):
resp.body = json.dumps(args)
class EchoUseKwargsWithPathParam(object):
@use_kwargs({"value": fields.Int()})
def on_get(self, req, resp, value, name):
resp.body = json.dumps({"value": value})
class AlwaysError(object):
def on_get(self, req, resp):
def always_fail(value):
raise ma.ValidationError("something went wrong")
args = {"text": fields.Str(validate=always_fail)}
resp.body = json.dumps(parser.parse(args, req))
on_post = on_get
class EchoHeaders(object):
def on_get(self, req, resp):
resp.body = json.dumps(parser.parse(hello_args, req, locations=("headers",)))
class EchoCookie(object):
def on_get(self, req, resp):
resp.body = json.dumps(parser.parse(hello_args, req, locations=("cookies",)))
class EchoNested(object):
def on_post(self, req, resp):
args = {"name": fields.Nested({"first": fields.Str(), "last": fields.Str()})}
resp.body = json.dumps(parser.parse(args, req))
class EchoNestedMany(object):
def on_post(self, req, resp):
args = {
"users": fields.Nested(
{"id": fields.Int(), "name": fields.Str()}, many=True
)
}
resp.body = json.dumps(parser.parse(args, req))
def use_args_hook(args, context_key="args", **kwargs):
def hook(req, resp, params):
parsed_args = parser.parse(args, req=req, **kwargs)
req.context[context_key] = parsed_args
return hook
@falcon.before(use_args_hook(hello_args))
class EchoUseArgsHook(object):
def on_get(self, req, resp):
resp.body = json.dumps(req.context["args"])
def create_app():
app = falcon.API()
app.add_route("/echo", Echo())
app.add_route("/echo_query", EchoQuery())
app.add_route("/echo_json", EchoJSON())
app.add_route("/echo_form", EchoForm())
app.add_route("/echo_use_args", EchoUseArgs())
app.add_route("/echo_use_kwargs", EchoUseKwargs())
app.add_route("/echo_use_args_validated", EchoUseArgsValidated())
app.add_route("/echo_multi", EchoMulti())
app.add_route("/echo_many_schema", EchoManySchema())
app.add_route("/echo_use_args_with_path_param/{name}", EchoUseArgsWithPathParam())
app.add_route(
"/echo_use_kwargs_with_path_param/{name}", EchoUseKwargsWithPathParam()
)
app.add_route("/error", AlwaysError())
app.add_route("/echo_headers", EchoHeaders())
app.add_route("/echo_cookie", EchoCookie())
app.add_route("/echo_nested", EchoNested())
app.add_route("/echo_nested_many", EchoNestedMany())
app.add_route("/echo_use_args_hook", EchoUseArgsHook())
return app