1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160
|
import marshmallow as ma
from bottle import Bottle, HTTPResponse, debug, request, response
from webargs import fields, validate
from webargs.bottleparser import parser, use_args, use_kwargs
from webargs.core import json
hello_args = {"name": fields.Str(load_default="World", validate=validate.Length(min=3))}
hello_multiple = {"name": fields.List(fields.Str())}
class HelloSchema(ma.Schema):
name = fields.Str(load_default="World", validate=validate.Length(min=3))
hello_many_schema = HelloSchema(many=True)
# variant which ignores unknown fields
hello_exclude_schema = HelloSchema(unknown=ma.EXCLUDE)
app = Bottle()
debug(True)
@app.route("/echo", method=["GET"])
def echo():
return parser.parse(hello_args, request, location="query")
@app.route("/echo_form", method=["POST"])
def echo_form():
return parser.parse(hello_args, location="form")
@app.route("/echo_json", method=["POST"])
def echo_json():
return parser.parse(hello_args, location="json")
@app.route("/echo_json_or_form", method=["POST"])
def echo_json_or_form():
return parser.parse(hello_args, location="json_or_form")
@app.route("/echo_use_args", method=["GET"])
@use_args(hello_args, location="query")
def echo_use_args(args):
return args
@app.route(
"/echo_use_args_validated",
method=["POST"],
apply=use_args(
{"value": fields.Int()},
validate=lambda args: args["value"] > 42,
location="form",
),
)
def echo_use_args_validated(args):
return args
@app.route("/echo_ignoring_extra_data", method=["POST"])
def echo_json_ignore_extra_data():
return parser.parse(hello_exclude_schema, unknown=None)
@app.route(
"/echo_use_kwargs", method=["GET"], apply=use_kwargs(hello_args, location="query")
)
def echo_use_kwargs(name):
return {"name": name}
@app.route("/echo_multi", method=["GET"])
def echo_multi():
return parser.parse(hello_multiple, request, location="query")
@app.route("/echo_multi_form", method=["POST"])
def multi_form():
return parser.parse(hello_multiple, location="form")
@app.route("/echo_multi_json", method=["POST"])
def multi_json():
return parser.parse(hello_multiple)
@app.route("/echo_many_schema", method=["POST"])
def echo_many_schema():
arguments = parser.parse(hello_many_schema, request)
return HTTPResponse(body=json.dumps(arguments), content_type="application/json")
@app.route(
"/echo_use_args_with_path_param/<name>",
apply=use_args({"value": fields.Int()}, location="query"),
)
def echo_use_args_with_path_param(args, name):
return args
@app.route(
"/echo_use_kwargs_with_path_param/<name>",
apply=use_kwargs({"value": fields.Int()}, location="query"),
)
def echo_use_kwargs_with_path_param(name, value):
return {"value": value}
@app.route("/error", method=["GET", "POST"])
def always_error():
def always_fail(value):
raise ma.ValidationError("something went wrong")
args = {"text": fields.Str(validate=always_fail)}
return parser.parse(args)
@app.route("/echo_headers")
def echo_headers():
return parser.parse(hello_args, request, location="headers")
@app.route("/echo_cookie")
def echo_cookie():
return parser.parse(hello_args, request, location="cookies")
@app.route("/echo_file", method=["POST"])
def echo_file():
args = {"myfile": fields.Raw()}
result = parser.parse(args, location="files")
myfile = result["myfile"]
content = myfile.file.read().decode("utf8")
return {"myfile": content}
@app.route("/echo_nested", method=["POST"])
def echo_nested():
args = {"name": fields.Nested({"first": fields.Str(), "last": fields.Str()})}
return parser.parse(args)
@app.route("/echo_nested_many", method=["POST"])
def echo_nested_many():
args = {
"users": fields.Nested({"id": fields.Int(), "name": fields.Str()}, many=True)
}
return parser.parse(args)
@app.error(400)
@app.error(422)
def handle_error(err):
response.content_type = "application/json"
return err.body
|